526互联

使用python-telegram-bot模块转发telegram机器人消息到钉钉平台

发布时间 2023-05-21 17:53:18作者: Egu0

脚本：将电报频道机器人的消息转发到钉钉上（工作中钉钉用的多）

步骤：

  1. 创建电报机器人，将感兴趣的消息分享给机器人
  2. 使用本程序监听机器人的消息，收到消息时转发 markdown 到钉钉
# encoding: utf-8

import json
import time
import hmac
import hashlib
import base64
import urllib.parse
from datetime import datetime, timedelta, timezone

import requests
from telegram import Update
from telegram.ext import MessageHandler, filters, Application

"""
环境：python 3.8

依赖：
------------------------ requirements.txt
python-telegram-bot
requests==2.29
------------------------

部署：
nohup python3 tgbot.py > out.log 2>&1 &
"""


# 自定义机器人（基于 webhook）
# 使用 加签 的安全方式
def sign_cos_dingtalk_restrict():
    secret = '钉钉机器人加签秘钥'
    timestamp = str(round(time.time() * 1000))
    secret_enc = secret.encode('utf-8')
    string_to_sign = '{}\n{}'.format(timestamp, secret)
    string_to_sign_enc = string_to_sign.encode('utf-8')
    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return timestamp, sign


# 北京时间
def beijing_time():
    utc_now = datetime.utcnow().replace(tzinfo=timezone.utc)
    sh_timezone = timezone(timedelta(hours=8), name='Asia/Shanghai', )
    beijing_now = utc_now.astimezone(sh_timezone)
    return beijing_now.strftime('%d/%m %H:%M:%S')


# 发送钉钉消息
def post_dingtalk_msg(markdown_text: str):
    if markdown_text is None:
        return
    ts, sign = sign_cos_dingtalk_restrict()
    token = '钉钉机器人访问的accesstoken'
    url = 'https://oapi.dingtalk.com/robot/send?access_token={}&sign={}&timestamp={}'.format(token, sign, ts)
    data = {
        'msgtype': 'markdown',
        'markdown': {
            'title': '一条转发消息',
            'text': markdown_text
        }
    }
    headers = {"Content-Type": "application/json"}
    rsp = requests.request(url=url, method='POST', headers=headers, data=json.dumps(data))
    print(beijing_time(), 'post dingtalk rsp status: ', rsp.status_code, ', rsp text: ', rsp.text)


# 消息处理器回调。参数必须是俩
async def fetch_updates(update: Update, context):
    try:
        markdown_text = update.message.caption_markdown_v2_urled
        post_dingtalk_msg(markdown_text)
    except Exception as ex:
        print('提取消息或发送消息失败', ex)


# telegram 机器人 access token
bot_access_token = "电报机器人accesstoken"


def main():
    # 使用消息处理器持续监听机器人收到的消息
    application = Application.builder().token(bot_access_token).build()
    application.add_handler(MessageHandler(filters.ALL, callback=fetch_updates))
    application.run_polling()


if __name__ == '__main__':
    main()

文档：python-telegram-bot v20.x