Python+企业微信API实现盒子端deluge完成后自动推送

本文最后更新于:3 个月前

前言


因为有部分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推送。

100cff34c4fb70a3e.png

27151b3e1ec9c01ee.png

点进来有很多有用的信息,如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

36022b6025acc03a8.png

大概就是之前我们获取到的信息了。然后我们开始给个人发消息测试:

# 发送个人消息,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 )
  1. * * * * * 表示每分钟执行一次,为了保证20秒一次,我需要一分钟执行三次,所以后边两行代码分别休眠20秒和40秒。

  2. PYTHONIOENCODING=utf-8是为了解决python3脚本运行过程中出现的编码问题,这个困扰了我一晚上。详见:https://blog.csdn.net/TH_NUM/article/details/80685389

  3. 在写的时候最后一个路径为脚本的绝对路径。

使用


如果想要使用,找本人获取源代码。将代码和config.json配置好放到固定目录,crontab -e添加即可。

附上一个效果截图:

4.md.png

进阶玩法


成员AAA发现可以使用企业微信插件,扫码之后就可以实现微信接收消息,不用下载企业微信了~~~,具体方案直接百度。。妙哉。

总结


折腾不易,且行且珍惜。

参考链接




本博客所有文章除特别声明外,均采用 CC BY-SA 4.0 协议 ,转载请注明出处!