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

阿里云获取CDN数据代码

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
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)

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
"""
飞书群机器人发送通知
"""
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

最后调用一下就好了

1
2
3
4
5
6
7
8
9
10
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