heriec

飞书发送每日阿里云CDN账单到飞书卡片

发布于 2024-11-10

目录

飞书的机器人实在太好用了，我的网站使用阿里云的CDN服务，但是我不会每天去看过去一天我的CDN使用多少，但是我还是想通过CDN的使用来观测我网站当前的用户怎么样，因为这相当于报表的形式，如果我有后端，我就去统计我自己数据库的数据发送到飞书了，但是我为了成本，所以只有前端，是一个工具网站，如果能看到我这篇博客的，肯定都是程序员了，那我也安利一下我的网站，即使你可能用不到easygif.cn

阿里云获取CDN数据代码

好了，言归正传，跟之前一篇博客一样，我先给出阿里云的接口https://api.aliyun.com/api/Cdn/2018-05-10/DescribeDomainUsageData

import datetime
import os

from typing import List

from alibabacloud_bssopenapi20171214.client import Client as BssOpenApi20171214Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_bssopenapi20171214 import models as bss_open_api_20171214_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient

from alibabacloud_cdn20180510.client import Client as Cdn20180510Client
from alibabacloud_cdn20180510 import models as cdn_20180510_models


class AliyunClient:
    def __init__(self):
        pass
    def createCdnClient(self) -> Cdn20180510Client:
        """
        使用AK&SK初始化账号Client
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        config.endpoint = f'cdn.aliyuncs.com'
        return Cdn20180510Client(config)

    def cdnUsage(self, day: datetime.date, domain: str, type: str) -> None:
        client = self.createCdnClient()
        describe_domain_usage_data_request = cdn_20180510_models.DescribeDomainUsageDataRequest(
            domain_name=domain,
            start_time=day.strftime('%Y-%m-%dT00:00:00Z'),
            end_time=day.strftime('%Y-%m-%dT23:59:59Z'),
            field=type
        )
        runtime = util_models.RuntimeOptions()
        try:
            response = client.describe_domain_usage_data_with_options(
                describe_domain_usage_data_request, runtime)
            return response.body.usage_data_per_interval.data_module
        except Exception as error:
            print(error.message)
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

这个比较简单，根据文档直接cv即可

获取这个数据我只看请求数和流量，然后我就要把这个数据发送给飞书卡片

飞书卡片

给出对应的飞书卡片组件的文档地址：组件概述 - 开发指南 - 开发文档 - 飞书开放平台 (feishu.cn)

然后就直接搞，我这个很简单，因为我用得少，我之前还有个代码是放服务器上跑，等跑完提醒我的卡片，数据就很详细

"""
飞书群机器人发送通知
"""
import json
from logging import Logger
import requests
import datetime


class FlybookRobotAlert():
    def __init__(self, webhook_url, logger=Logger("飞书通知")):
        self.webhook = webhook_url
        self.logger = logger

        self.headers = {'Content-Type': 'application/json; charset=UTF-8'}

    def post_to_robot(self, post_data):
        '''
        给飞书机器人发送请求
        :param data:
        :return:
        '''
        try:
            resp = requests.request(
                method="POST", url=self.webhook, data=post_data, headers=self.headers).json()
            if resp.get("StatusCode") == 0 and resp.get("msg") == "success":
                self.logger.info(f"飞书通知发送成功，msg={resp}")
            else:
                self.logger.warning(f"飞书通知发送失败,{resp}")
        except Exception as e:
            self.logger.warning("飞书通知发送异常")
            self.logger.warning(e)
            pass

    def send_message(self, date, acc, traf):
        # 飞书通知标题
        robot_headers = 'cdn用量报告'
        field_list = [
            {
                "is_short": False,
                "text": {
                    "tag": "lark_md",
                    "content": f"**请求数**：<font color=\"green\">{{}}</font>   **次**\n".format(acc)
                }
            },
            {
                "is_short": False,
                "text": {
                    "tag": "lark_md",
                    "content": f"**流量**：<font color=\"green\">{{:.2f}}</font>    **MB**\n".format(traf / 1e6)
                }
            }
        ]

        elements = [
            {
                "tag": "div",
                "text": {
                    "content": date.strftime("%Y年%m月%d日"),
                    "tag": "lark_md"
                }
            },
            {
                "tag": "div",
                "fields": field_list
            }
        ]

        card = json.dumps({
            "config": {
                "wide_screen_mode": True
            },
            "elements": elements,
            "header": {
                "template": "blue",
                "title": {
                    "content": robot_headers,
                    "tag": "plain_text"
                }
            }
        })

        msg_body = json.dumps({"msg_type": "interactive", "card": card})
        self.post_to_robot(msg_body)
        # {'StatusCode': 0, 'StatusMessage': 'success', 'code': 0, 'data': {}, 'msg': 'success'}
        return

最后调用一下就好了

if __name__ == '__main__':
    ################# cdn ##################
    result = {}
    for type in ['acc', 'traf']:
        cdnUsage = AliyunClient().cdnUsage(yesterday, "easygif.cn", type)
        result[type] = sum([item.value for item in cdnUsage])
            
    print(result)
    a = FlybookRobotAlert("https://open.feishu.cn/open-apis/bot/v2/hook/xxxxx")
    a.send_message(yesterday, result['acc'], result['traf'])

最后结果：

image-20241110162458084