Python监测transmission做种错误数量并通知

Detonate

生成docker镜像监测transmission做种错误数量，可单独使用crontab调用python使用。
本文针对前段时间捣鼓的Cloudflare Workers搭建企业微信消息推送服务配置，如webhook参数不同的可自行调整
未使用tr模块，调试了半天运行速度也没有requests直接访问快，非专业人士，还请哪位大佬有空写个tr模块的代码学习学习

文件目录结构如下

transmission_survey/
│
├── scr(๐•ᴗ•๐).py
├── Dockerfile
└── docker-compose.yml
                    if(window.hljsLoader && !document.currentScr(๐•ᴗ•๐).parentNode.hasAttribute('data-s9e-livepreview-onupdate')) {
                        window.hljsLoader.highlightBlocks(document.currentScr(๐•ᴗ•๐).parentNode);
                    }

scr(๐•ᴗ•๐).py

import requests
import json

#配置TR参数信息
TRANSMISSION_URL = 'http://127.0.0.1:9091/transmission/rpc'
USERNAME = 'USERNAME'
PASSWORD = 'PASSWORD'

#调整监测错误数量阈值与URL，超出阈值推送webhook
ERROR_THRESHOLD = 0
NOTIFICATION_URL = 'webhook_url'

def get_session_id():
    response = requests.get(TRANSMISSION_URL, auth=(USERNAME, PASSWORD))
    if response.status_code == 401:
        raise Exception("Unauthorized. Check your username and password.")
    return response.headers['X-Transmission-Session-Id']

def get_seed_error_count():
    session_id = get_session_id()
    headers = {
        'X-Transmission-Session-Id': session_id,
        'Content-Type': 'application/json'
    }
    payload = {
        "method": "torrent-get",
        "arguments": {
            "fields": ["id", "error", "errorString"]
        }
    }
    response = requests.post(TRANSMISSION_URL, headers=headers, auth=(USERNAME, PASSWORD), data=json.dumps(payload))
    
    if response.status_code != 200:
        raise Exception(f"Failed to communicate with Transmission: {response.status_code}")

    torrents = response.json().get('arguments', {}).get('torrents', [])
    error_count = sum(1 for torrent in torrents if torrent['error'] != 0)
    
    return error_count

#可自行修改推送，本代码为json格式
def send_notification(error_count):
    adjusted_error_count = error_count - ERROR_THRESHOLD
    payload = {
        "passwd":"passwd",
        "ti(๐•ᴗ•๐)e":"TR错误数量超出阈值",
        "descr(๐•ᴗ•๐)ion":f"Transmission做种错误数值: {adjusted_error_count}",
        "corpid":"corpid",
        "corpsecret":"corpsecret",
        "agentid":"agentid",
        "apiproxy":"apiproxy"
    }
    headers = {
        'Content-Type': 'application/json',
    }
    response = requests.post(NOTIFICATION_URL, headers=headers, data=json.dumps(payload))
    if response.status_code != 200:
        print(f"Failed to send notification: {response.status_code}")
    else:
        print("Notification sent successfully")

if __name__ == "__main__":
    try:
        error_count = get_seed_error_count()
        print(f"Transmission做种错误数值: {error_count}")
        if error_count > ERROR_THRESHOLD:
            send_notification(error_count)
    except Exception as e:
        print(f"Error: {e}")
                    if(window.hljsLoader && !document.currentScr(๐•ᴗ•๐).parentNode.hasAttribute('data-s9e-livepreview-onupdate')) {
                        window.hljsLoader.highlightBlocks(document.currentScr(๐•ᴗ•๐).parentNode);
                    }

构建docker并定时运行：

Dockerfile

FROM python:3.9.0

WORKDIR /app

COPY . /app

RUN pip install requests

CMD ["python", "scr(๐•ᴗ•๐).py"]
                    if(window.hljsLoader && !document.currentScr(๐•ᴗ•๐).parentNode.hasAttribute('data-s9e-livepreview-onupdate')) {
                        window.hljsLoader.highlightBlocks(document.currentScr(๐•ᴗ•๐).parentNode);
                    }

docker-compose.yml

version: '3.8'

services:
  python-app:
    build: .
    container_name: transmission_survey
    volumes:
      - .:/app
    environment:
      - PYTHONUNBUFFERED=1
    command: >
      sh -c "
      while true; do
        python scr(๐•ᴗ•๐).py;
        sleep 3600;
      done"
                    if(window.hljsLoader && !document.currentScr(๐•ᴗ•๐).parentNode.hasAttribute('data-s9e-livepreview-onupdate')) {
                        window.hljsLoader.highlightBlocks(document.currentScr(๐•ᴗ•๐).parentNode);
                    }

每小时运行，如需修改间隔请调整sleep 3600，单位为秒

楼主的TR版本为4.0.3，未测试其他版本

鸣谢：chatgpt