搞点新花样——飞书签名自动更新

前两天,看见公司有一些同学利用一些技术手段动态更新飞书签名,有人实时更新微博热搜,有人春节放假倒计时,还是挺有趣的,正巧北京这几天疫情比较紧张,我就也写了一个服务在飞书签名上定时更新北京的风险地区。

梳理了一下步骤流程,首先获取到飞书的cookie,然后在字节的“轻服务”平台上线一个服务,用来利用cookie更新飞书签名,接着在开发机部署一个定时任务,定时在丁香园爬取北京的风险地区,然后将数据post到刚刚上线的轻服务上,数据打过去之后,轻服务就能更新签名了。至于为什么用轻服务而不在开发机走完整个流程,因为cookie里的信息比较重要,放在开发机不够安全。

获取cookie

登录飞书网页版,对任意飞书api请求均可,复制cookie

cookie

在轻服务平台创建轻函数并上线

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
module.exports = async function(params, context) {

const request = axios.create({
baseURL: 'https://internal-api-lark-api.feishu.cn',
headers: {
'content-type': 'text/plain;charset=UTF-8',
'cookie':[]
}
});

await request.put('/passport/users/details/',
{"description": params.data, "descriptionType": 0} )
.then(res => {
console.log(res.data)
})
.catch(error => {
console.log(error)
})

}

在开发机部署定时爬虫任务

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
import json
import re
import traceback
import requests
import datetime
import time


def crawl_dxy_data():
"""爬取丁香园实时统计数据,发送到字节轻服务,更新lark签名
"""
response = requests.get('https://ncov.dxy.cn/ncovh5/view/pneumonia') # 发送get请求
print(response.status_code) # 打印状态码

try:
today = datetime.datetime.now().strftime('%Y-%m-%d %H:%M')
url_text = response.content.decode() # 获取响应的html页面
# re.search()用于扫描字符串以查找正则表达式模式产生匹配项的第一个位置,然后返回相应的match对象
# 在字符串a中,包含换行符\n,这种情况下:如果不适用re.S参数,则只在每一行内进行匹配,如果一行没有,就换下一行重新开始匹配
url_content = re.search(r'window.getAreaStat = (.*?)}]}catch',
url_text, re.S)
texts = url_content.group() # 获取匹配正则表达式的整体结果
content = texts.replace('window.getAreaStat = ', '').replace('}catch', '') # 去除多余字符
json_data = json.loads(content)
string_danger_areas = "【" + today + "自动更新北京疫情风险地区】"
num_of_high = 0
num_of_mid = 0
danger_areas = []
for _json in json_data:
if _json["provinceShortName"] == "北京":
danger_areas = _json["dangerAreas"]
break
for danger_area in danger_areas:
if danger_area['dangerLevel'] == 1:
num_of_high += 1
elif danger_area['dangerLevel'] == 2:
num_of_mid += 1
if num_of_high > 0:
string_danger_areas += "高风险:"
for danger_area in danger_areas:
if danger_area['dangerLevel'] == 1:
string_danger_areas += danger_area['cityName'] + danger_area['areaName'] + ";"
if num_of_mid > 0:
string_danger_areas += "中风险:"
for danger_area in danger_areas:
if danger_area['dangerLevel'] == 2:
string_danger_areas += danger_area['cityName'] + danger_area['areaName'] + ";"

url = "轻服务url"
post_data = {"data": string_danger_areas}
# 字符串格式
res = requests.post(url=url, json=post_data)
print(res.status_code)
print(string_danger_areas)

except:
print(traceback.format_exc())


if __name__ == '__main__':
while (True):
crawl_dxy_data()
time.sleep(300)

在后台不挂起执行

1
$nohup python3 crawl.py &

最终效果