Python+企业微信API实现盒子端deluge完成后自动推送
本文最后更新于:7 个月前
前言
因为有部分PTER希望实现下载完成后推送进行抢发的需求,以前写过发邮件的代码,这次想着弄个不一样的,就折腾了一下午+一晚上简单实现了标题的内容。
主要是定时监控盒子端deluge客户端完成下载之后进行企业微信的推送。话不多说,直接上过程。
企业微信设置
网上材料也有蛮多,踩坑也有一些的。一部分基于微信的发送消息,看看网上找到的代码:
import logging
import itchat
log = logging.getLogger(__name__)
# itchat 微信官方教程:https://itchat.readthedocs.io/zh/latest/
# 微信登录 登录时如果断网,则此程序直接停止 启动热登录,并且生成 命令行 登录二维码
itchat.auto_login(hotReload=True, enableCmdQR=2)
# 保持心跳状态,防止自动退出登录
itchat.start_receiving()
# 获取群聊,注意群 必须保存到通讯录,否则可能会找不到群
itchat.get_chatrooms(update=True)
room = itchat.search_chatrooms('python')
if len(room) == 0:
log.error('没有找到群信息')
else:
try:
iRoom = room[0]['UserName']
# 发送消息
result = itchat.send('send message', iRoom)
try:
if result['BaseResponse']['ErrMsg'] == '请求成功':
log.info('send wechat success')
except Exception as e:
print('resolve wechat result fail,result is :{},error is {}'.format(result, e))
except Exception as e:
print('wechat send message fail,reason is :{} '.format(e))
需要生成命令行登录二维码,如果部署在盒子端,我希望每20秒运行一次,都要获取一次链接扫码登录,基本不现实,所以pass掉了~~~
所以选择了企业微信的消息推送。操作流程如下:
先开通企业微信,个人开通即可,但是验证会比较麻烦,写教程的时候并没有验证。申请地址 https://work.weixin.qq.com/ 之后从我的企业面板查找企业 id(在获取访问access-token时需要)
获取access-token才能通过api发送消息,其中还需要创建一个应用,通过这个应用来给群发消息,如何对应一个企业群的应用呢? corpsecret。首先在应用与小程序面板创建应用,然后通过手机企业微信(需要管理员)获取corpsecret。
在使用应用进行推送的时候,可以选择群聊,也可以发送给群里的某一个人,在这之前需要将PTER添加进来,然后应用授权。如下选择新建的一个应用PT推送。
点进来有很多有用的信息,如agentid,点击查看secret会知道corpsecret,后边代码都能用到,然后编辑可以选择企业部门或者部门里的人有权限访问PT推送的消息。悬停在用户名称上还可以看到他的用户名,也是后续代码需要用到的。
充分掌握了上述信息之后,就可以用代码进行自动发送消息给个人了。还是用网上的例子加以说明:
import json
import requests
# 请求API响应结果
result = requests.get("https://qyapi.weixin.qq.com/cgi-bin/gettoken",
params={'corpid': 'fg', # 这里替换成自己的企业id
'corpsecret': '45'}) # 企业应用的secret
access_token = None
if result.status_code != 200:
print('连接到服务器失败')
else:
result_json = json.loads(result.text)
if result_json['errcode'] != 0:
print('响应结果不正确')
else:
access_token = result_json['access_token']
print(access_token)
这一段是用来请求API获取access_token的,那么我们如果想要知道这个token具体包含哪些信息的话,可以上网页上查询:
网址:https://open.work.weixin.qq.com/devtool/query
大概就是之前我们获取到的信息了。然后我们开始给个人发消息测试:
# 发送个人消息,content为发送的内容
result = requests.post('https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={}'.format(access_token),
data=json.dumps({
"touser": "user_name", # 替换成为你刚才获取到的成员ID,单单给某一个人发送
"msgtype": "text",
"agentid": 23,
"text": {
"content": "你的快递已到,请携带工卡前往邮件中心领取"
},
"safe": 0
}
))
print(result.text)
经过测试,企业微信的消息应该已经发送了,支持页面内部HTML标签。我们也可以设计一下发送的内容~
盒子端
准备工作:首先使用python3运行脚本,模块使用了requests
sudo apt-get install python3-pip
pip3 install requests
既然知道了怎么发送消息,那么我们怎么实时监控deluge的下载状况并且实时的发送消息呢?
下面给一个网页介绍deluge的相关命令:https://whatbox.ca/wiki/Deluge_Console_Documentation
'deluge-console "connect 127.0.0.1:{port} {name} {pwd}; info -v --sort=seeding_time; exit"'
我们用到的就是这样一条。然后就可以通过python调用系统命令获取输出进行正则获取种子详情。我们看看正常输出大概是什么样子:
Name: 木棘证人.The.Witness.2021.4K.WEB-DL.H265.AAC-PTerWEB
ID: 82a6804b1e6b57be0de56fed5c4f1f19cffd8a92
State: Seeding Up Speed: 0.0 K/s
Seeds: 0 (39) Peers: 0 (1) Availability: 1.00 Seed Rank: 51
Size: 34.9 G Downloaded: 0 B Uploaded: 1.1 T Share Ratio: 32.80
ETA: - Seeding: 2d 10h Active: 1d 19h
Last Transfer: 6h 29m Complete Seen: 08/21/2021
Tracker: pterclub.com
Tracker status: Announce OK
Download Folder: /home/guai/deluge/download
Name: 机智的上半场.Ji.Zhi.De.Shang.Ban.Chang.2021.E05-E07.WEB-DL.4K.H265.AAC-PTerWEB
ID: 35d93424deb5c52da944839f1bb26bba46ca809c
State: Seeding Up Speed: 743.8 K/s
Seeds: 0 (59) Peers: 2 (2) Availability: 1.94 Seed Rank: 50
Size: 4.7 G Downloaded: 0 B Uploaded: 84.9 G Share Ratio: 18.05
ETA: - Seeding: 22h 31m Active: 6h 16m
Last Transfer: 1s Complete Seen: 08/21/2021
Tracker: pterclub.com
Tracker status: Announce OK
Download Folder: /home/guai/deluge/download
其中最重要的当属seeding_time了,原理就是每N秒看一次,如果某个种子做种时间不超过N秒,那么它就是刚刚完成的种子,当然了,也可以使用一个缓存记录hash之类的,这个可以自己去想。
import subprocess
import re
def get_data(_user_info):
post_data = []
command = 'deluge-console "connect 127.0.0.1:{port} {name} {pwd}; info -v --sort=seeding_time; exit"'.format(
port=_user_info['port'],
name=_user_info['name'],
pwd=_user_info['pwd'],
)
# 调用subprogress模块进行获取输出
data = subprocess.Popen(command, shell=True, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
# 进行种子识别,获取所有种子数量
torrents = str(data.stdout.read().decode()).split('\n\n')
logger.info('下载器中有{}个种子!!!'.format(len(torrents)))
# 因为我们对种子进行了排序,按照做种时间排序之后做种时间长的在后边
seeding_too_long = False
for torrent in torrents:
# 发现种子已经做种超过1分钟,就忽略后边的了,节约时间
if seeding_too_long:
break
infos = list(filter(None, torrent.split('\n')))
for info in infos:
if info.startswith('ETA:'):
seeding_time = re.search('Seeding: (.*)Active:', info).group(1)
# 判断做种时间在秒级的
if seeding_time.find('d') < 0 and seeding_time.find('m') < 0 and seeding_time.find('s') >= 0:
seeding_time = int(re.search('(\d+)s', seeding_time).group(1))
if seeding_time <= int(_user_info['time']):
logger.info('{}\n已经下载完成了,请留意!!!'.format(infos[0].replace('Name: ', '')))
post_data.append('{}\n已经下载完成了,请留意!!!'.format(infos[0].replace('Name: ', '')))
else:
if seeding_time.find('-') < 0:
seeding_too_long = True
return '\n'.join(post_data)
上边这个函数通过调用一次deluge-console命令获取到了做种时间在规定时间内的种子进行返回。接下来就可以进行推送了~
def send_data(raw_data, _user_info):
logger.info('获取access-token……')
result = requests.get("https://qyapi.weixin.qq.com/cgi-bin/gettoken",
params={'corpid': _user_info['corpid'],
'corpsecret': _user_info['corpsecret']})
access_token = None
if result.status_code != 200:
logger.info('连接到服务器失败')
else:
result_json = json.loads(result.text)
if result_json['errcode'] != 0:
logger.info('响应结果不正确')
else:
access_token = result_json['access_token']
if access_token:
logger.info('获取access-token成功!!!!')
result = requests.post('https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={}'.format(access_token),
data=json.dumps({
"touser": _user_info['user'],
"msgtype": "text",
"agentid": _user_info['agentid'],
"text": {
"content": raw_data,
},
"safe": 0,
"debug": 1
}
))
logger.info(result.text)
整个代码我们把所有的配置文件放在了一个json里,方便直接配置,上述的user_info就是配置文件里的内容:
def get_config():
config_path = os.path.join(base_path, 'config.json')
if not os.path.exists(config_path):
logger.info('请确定存在配置文件!!!')
return {}
with open(config_path, 'r') as f:
data = json.loads(f.read())
return data
config.json里边的内容如下:端口,用户名和密码是deluge的信息,时间20,数量级是秒。公司和应用secret以及发送信息接收用户id,部门id之类信息。
{
"port": "58846",
"name": "",
"pwd": "",
"time": 20,
"corpid": "",
"corpsecret": "",
"user": "",
"agentid": 1000002
}
自动化部署
上边解决了一次性的运行结果,那么我们如何进行自动化部署呢?这里需要搭配上述配置文件里的时间,20秒的话我希望20秒运行一次,linux下用crontab就好了。直接crontab -e:在最后一行加入:
* * * * * PYTHONIOENCODING=utf-8 python3 /home/deluge/push_pt_info.py
* * * * * ( sleep 20 ; PYTHONIOENCODING=utf-8 python3 /home/deluge/push_pt_info.py )
* * * * * ( sleep 40 ; PYTHONIOENCODING=utf-8 python3 /home/deluge/push_pt_info.py )
* * * * *
表示每分钟执行一次,为了保证20秒一次,我需要一分钟执行三次,所以后边两行代码分别休眠20秒和40秒。PYTHONIOENCODING=utf-8是为了解决python3脚本运行过程中出现的编码问题,这个困扰了我一晚上。详见:https://blog.csdn.net/TH_NUM/article/details/80685389
在写的时候最后一个路径为脚本的绝对路径。
使用
如果想要使用,找本人获取源代码。将代码和config.json配置好放到固定目录,crontab -e添加即可。
附上一个效果截图:
进阶玩法
成员AAA发现可以使用企业微信插件,扫码之后就可以实现微信接收消息,不用下载企业微信了~~~,具体方案直接百度。。妙哉。
总结
折腾不易,且行且珍惜。