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发现可以使用企业微信插件，扫码之后就可以实现微信接收消息，不用下载企业微信了~~~，具体方案直接百度。。妙哉。

总结

折腾不易，且行且珍惜。

参考链接

  1. python发送微信及企业微信消息 ↩
  2. 发送应用消息 ↩
  3. Python3中遇到UnicodeEncodeError: ‘ascii’ codec can’t encode characters in ordinal not in range(128) ↩
  4. Deluge Console DocumentationStart the console interface ↩
  5. crontab 每10秒 执行一次脚本 ↩
编程学习
编程 Python 学习

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