# coding: utf-8 """ process.py —— torrent 下载中央监控 / 控制服务（固定机器上运行） ================================================================ 部署：在固定机器（IP 21.196.89.120）上执行 `python process.py` 起一个 HTTP 服务， 汇总所有 downloader 上报的状态并展示，同时可通过接口控制 downloader 任务的 暂停 / 开始 / 删除。 持久化 ------ 所有任务快照写入本机 SQLite（Approach B：downloader 只负责 HTTP 上报，process 独占 DB 单写入端）。进程重启时从 DB 加载全部记录，看板不因重启丢历史。DB 路径由 环境变量 TORRENT_PROCESS_DB 指定（默认与本文件同目录 process_state.db）。 HTTP 接口 --------- POST /api/report downloader 周期性上报本机某个种子的状态快照（JSON）。 响应体返回该任务当前“期望状态” {"desired": "running|paused|deleted"}。 POST /api/control 控制某个任务：body {"task_key": "rid:info_hash", "action": "pause|start|delete|force_ingest"} POST /api/retry 重试：body {"task_key": "..."} 或 {"task_keys": [...]}，把原始种子入队到原机本机重试队列（复用本地文件重跑，不走远端 job_id） GET /api/pending_retries 原机 downloader 轮询取回本机待重试的原始种子：?machine=X -> {"seeds": [...]} GET /api/summary 过滤后的分类计数 / 机器磁盘 / 速度统计 / requirement_id 候选。 GET /api/list 某分类的一页轻量行（不含 files/bt，支持 offset/limit 无限下拉）。 GET /api/task 单个 task_key 的完整明细（files/bt，展开时按需拉取）。 GET /api/export 按当前筛选（requirement_id + 时间）导出某分类为 CSV。 GET /api/status 全量状态（兼容旧接口 / 控制台）。 GET /api/dispatch 重复投递决策（completed/ignore/proceed + mode）。 GET / Web 看板（分类无限下拉懒加载 + 按 requirement_id/添加时间筛选 + 分类导出 + 静默刷新防卡顿）。 筛选：requirement_id 为必填字段（downloader 必须上报，看板按其 + 添加时间联合筛选）。 """ from __future__ import annotations import csv import datetime import io import json import os import sqlite3 import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Dict, List, Optional, Tuple from urllib.parse import urlparse, parse_qs import requests from loguru import logger # ============================================================================ # 常量 # ============================================================================ # 固定监控机器 PROCESS_SERVER_HOST = "21.196.89.120" PROCESS_SERVER_PORT = 8686 PROCESS_SERVER_URL = f"http://{PROCESS_SERVER_HOST}:{PROCESS_SERVER_PORT}" # 持久化 DB 路径（process 独占单写） DB_PATH = os.environ.get( "TORRENT_PROCESS_DB", os.path.join(os.path.dirname(os.path.abspath(__file__)), "process_state.db")) # 后台刷盘间隔：progress-only 变更攒批写；状态/期望态/新任务立即写 FLUSH_INTERVAL = float(os.environ.get("TORRENT_PROCESS_FLUSH_INTERVAL", "5")) # 文件 / 种子状态 ST_PENDING = "pending" ST_METADATA = "metadata" # 正在获取种子元数据（尚未开始下载文件） ST_CHECKING = "checking" # 正在校验磁盘上已存在的数据（本地哈希校验，非下载） ST_DOWNLOADING = "downloading" ST_UPLOADING = "uploading" ST_COMPLETED = "已完成" ST_ERROR = "error" ST_SKIPPED = "skipped" ST_PAUSED = "paused" ST_DELETED = "deleted" ST_STALLED = "stalled" # 下载停滞/未完成被放弃 ST_FORCE_INGESTED = "force_ingested" # 已强制入库（只处理了已下载完成的文件） ST_SEEDING = "seeding" # 下载完成后持续做种/分享（模仿 qB） ST_CANCELED_NO_SPACE = "canceled_no_space" # 种子过大、整机都塞不下，取消下载 ST_DELETED_NO_SPACE = "deleted_no_space" # 已下载入库后因存储不足被 FIFO 删除 ST_REQUEUED = "requeued" # 已点“重试”：原始种子已入队到原机本机重试队列，等原机本地重跑 ST_INGESTED_DONE = "ingested_done" # 下载+做种结束、进程已释放：数据已全部入库(COS)，本地已清 # 下载器自身 job_id：无进度(无本地数据)的重试可直接重投到这里，由调度分配到任意机器 # 重新下载（有进度的则按原机路由复用本地文件，见 MonitorStore.retry）。 RETRY_JOB_ID = os.environ.get("TORRENT_RETRY_JOB_ID", "16441f134ef940cc89305d78e4e9bb04") # 状态 -> 中文展示文案（看板用） STATE_LABELS = { ST_PENDING: "等待中", ST_METADATA: "获取元数据中", ST_CHECKING: "校验中", ST_DOWNLOADING: "下载中", ST_UPLOADING: "上传中", ST_COMPLETED: "已完成", ST_ERROR: "错误", ST_SKIPPED: "已跳过", ST_PAUSED: "已暂停", ST_DELETED: "已删除", ST_STALLED: "下载停滞放弃", ST_FORCE_INGESTED: "已强制入库", ST_SEEDING: "做种中", ST_CANCELED_NO_SPACE: "因存储不足取消下载", ST_DELETED_NO_SPACE: "因存储不足已下载入库后删除", ST_REQUEUED: "重投队列", ST_INGESTED_DONE: "已全部入库完成", } # 期望状态（控制指令的落地态） DESIRED_RUNNING = "running" DESIRED_PAUSED = "paused" DESIRED_DELETED = "deleted" # 强制入库：不再等待整个种子下载完成，直接把“已下载完成的文件”处理并入库后结束 DESIRED_FORCE_INGEST = "force_ingest" # 任务快照超过该秒数无上报视为离线 OFFLINE_SECONDS = 60 # 看板分类顺序（key）。做种中放最后（页面最底部）；重投队列在其上。 GROUP_KEYS = ["downloading", "paused", "stalled", "force_ingested", "canceled_no_space", "deleted_no_space", "deleted", "other", "requeued", "seeding", "ingested_done"] def _fmt_size(n: float) -> str: n = float(n or 0) for unit in ("B", "KB", "MB", "GB", "TB"): if n < 1024 or unit == "TB": return f"{n:.1f}{unit}" n /= 1024 return f"{n:.1f}TB" def _fmt_speed(bps: float) -> str: return _fmt_size(bps) + "/s" def _category_of(state: str, desired: str, offline: bool = False) -> str: """把 (state, desired) 归到看板分类（与前端保持一致，供 list/summary/export 共用）。 未识别的状态一律落入 "other"（其他特殊情况），保证任何种子都能在前端显示、不丢失。 离线判定：下载中/元数据/校验/上传/等待 等“进行中”状态若掉线（长时间无上报）， 归入 "other"（其他特殊情况）——机器/进程异常导致的中断，需人工关注/重试。 """ if state == ST_SEEDING: return "seeding" if state == ST_INGESTED_DONE: return "ingested_done" if state == ST_REQUEUED: return "requeued" if state == ST_FORCE_INGESTED: return "force_ingested" if state == ST_CANCELED_NO_SPACE: return "canceled_no_space" if state == ST_DELETED_NO_SPACE: return "deleted_no_space" if state in (ST_STALLED, ST_ERROR): return "stalled" if state == ST_DELETED or desired == DESIRED_DELETED: return "deleted" if desired == DESIRED_PAUSED or state == ST_PAUSED: return "paused" if state in (ST_DOWNLOADING, ST_METADATA, ST_CHECKING, ST_UPLOADING, ST_PENDING, ST_COMPLETED, ST_SKIPPED): # 进行中/已完成待交接的任务若掉线（长时间无上报），一律归入 "other" # （其他特殊情况）——机器/进程异常导致的中断，需人工关注/重试；否则会顶着 # “离线”徽标仍堆在“正在下载”分组里。 return "other" if offline else "downloading" return "other" def _task_key(rid, info_hash: str) -> str: """任务主键：同 info_hash 不同 requirement_id 属不同任务，避免互相覆盖/漏显示。""" try: r = int(rid or 0) except (TypeError, ValueError): r = 0 return f"{r}:{info_hash}" def _to_float(v) -> Optional[float]: try: if v is None or v == "": return None return float(v) except (TypeError, ValueError): return None # ============================================================================ # 服务端状态存储（内存为读源 + SQLite 持久化） # ============================================================================ class MonitorStore: """线程安全的任务状态 + 控制期望态存储，写穿到 SQLite。""" def __init__(self, db_path: str = DB_PATH): self._lock = threading.RLock() # task_key -> snapshot(dict)（downloader 上报的最新快照，含 machine/files 等） self._status: Dict[str, dict] = {} # task_key -> desired 状态 self._desired: Dict[str, str] = {} # info_hash -> 首次出现（添加）时间戳（用于按添加时间筛选，不被覆盖） self._added_ts: Dict[str, float] = {} # 待刷盘（progress-only 变更攒批） self._dirty: set = set() # 本机重试队列：machine -> [raw_seed,...]。重试改为“原机本地重投”（复用本地已下载 # 文件、尝试解压入库），不走远端 job_id（避免被调度到别的机器重新下载一遍）。 self._retry_queue: Dict[str, list] = {} self.db_path = db_path self._db = sqlite3.connect(db_path, check_same_thread=False) self._db.execute("PRAGMA journal_mode=WAL") self._db.execute("PRAGMA synchronous=NORMAL") self._init_schema() self._load_all() self._load_retry_queue() self._stop = threading.Event() threading.Thread(target=self._flush_loop, daemon=True, name="store-flush").start() # -------------------------------------------------- 持久化底层 def _migrate_legacy_tasks(self): """旧版以 info_hash 为主键的 tasks 表迁移到 task_key 复合主键。""" logger.info("[process] 迁移 tasks 表：info_hash 主键 -> task_key 复合主键") self._db.execute("ALTER TABLE tasks RENAME TO tasks_legacy") self._db.execute(""" CREATE TABLE tasks ( task_key TEXT PRIMARY KEY, info_hash TEXT, requirement_id INTEGER, name TEXT, machine TEXT, state TEXT, desired TEXT, category TEXT, progress REAL, download_rate REAL, upload_rate REAL, num_seeds INTEGER, num_peers INTEGER, num_files INTEGER, completed_files INTEGER, total_size INTEGER, added_ts REAL, updated_ts REAL, snapshot_json TEXT )""") legacy_cols = {r[1] for r in self._db.execute( "PRAGMA table_info(tasks_legacy)")} if "snapshot_json" in legacy_cols: self._db.execute(""" INSERT INTO tasks(task_key,info_hash,requirement_id,name,machine,state, desired,category,progress,download_rate,upload_rate,num_seeds, num_peers,num_files,completed_files,total_size,added_ts,updated_ts, snapshot_json) SELECT printf('%s:%s', COALESCE(requirement_id, 0), info_hash), info_hash, requirement_id, name, machine, state, desired, category, progress, download_rate, upload_rate, num_seeds, num_peers, num_files, completed_files, total_size, added_ts, updated_ts, snapshot_json FROM tasks_legacy""") self._db.execute("DROP TABLE tasks_legacy") self._db.commit() def _init_schema(self): cur = self._db.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name='tasks'") if cur.fetchone(): cols = {r[1] for r in self._db.execute("PRAGMA table_info(tasks)")} if "task_key" not in cols: self._migrate_legacy_tasks() self._db.execute(""" CREATE TABLE IF NOT EXISTS tasks ( task_key TEXT PRIMARY KEY, info_hash TEXT, requirement_id INTEGER, name TEXT, machine TEXT, state TEXT, desired TEXT, category TEXT, progress REAL, download_rate REAL, upload_rate REAL, num_seeds INTEGER, num_peers INTEGER, num_files INTEGER, completed_files INTEGER, total_size INTEGER, added_ts REAL, updated_ts REAL, snapshot_json TEXT )""") self._db.execute( "CREATE INDEX IF NOT EXISTS idx_rid_added ON tasks(requirement_id, added_ts)") self._db.execute("CREATE INDEX IF NOT EXISTS idx_state ON tasks(state)") # 错误/告警事件表（解压失败、存储不足跳过、入库失败、停滞放弃等），持久化供 # 前端“消息列表”查询历史。 self._db.execute(""" CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, info_hash TEXT, requirement_id INTEGER, machine TEXT, name TEXT, level TEXT, message TEXT )""") self._db.execute( "CREATE INDEX IF NOT EXISTS idx_ev_ts ON events(ts)") self._db.execute( "CREATE INDEX IF NOT EXISTS idx_ev_rid_ts ON events(requirement_id, ts)") # 本机重试队列（持久化：process 重启后原机仍能取回本机待重试种子）。 self._db.execute(""" CREATE TABLE IF NOT EXISTS retry_queue ( id INTEGER PRIMARY KEY AUTOINCREMENT, machine TEXT, raw_seed TEXT, ts REAL )""") self._db.execute( "CREATE INDEX IF NOT EXISTS idx_retry_machine ON retry_queue(machine)") self._db.commit() def _load_all(self): cur = self._db.execute( "SELECT task_key, info_hash, requirement_id, desired, added_ts, " "updated_ts, snapshot_json FROM tasks") n = 0 for key, ih, rid, desired, added_ts, updated_ts, snap_json in cur.fetchall(): try: snap = json.loads(snap_json) if snap_json else {} except Exception: snap = {} snap["info_hash"] = ih snap["requirement_id"] = int(rid or 0) snap["_added_ts"] = float(added_ts or 0) # 重启后按最后一次更新时间计算离线（未再上报即显示离线） snap["_report_ts"] = float(updated_ts or 0) self._status[key] = snap self._desired[key] = desired or DESIRED_RUNNING self._added_ts[key] = float(added_ts or 0) n += 1 logger.info(f"[process] 已从 DB 加载 {n} 条历史任务记录 ({self.db_path})") def _load_retry_queue(self): """加载持久化的本机重试队列（process 重启后原机仍能取回）。""" n = 0 try: cur = self._db.execute( "SELECT machine, raw_seed FROM retry_queue ORDER BY id") for machine, raw in cur.fetchall(): if machine and raw: self._retry_queue.setdefault(machine, []).append(raw) n += 1 except Exception as e: logger.warning(f"[process] 加载重试队列失败: {e}") if n: logger.info(f"[process] 已从 DB 加载 {n} 条待重试种子") def _persist_locked(self, key: str): snap = self._status.get(key) if snap is None: return desired = self._desired.get(key, DESIRED_RUNNING) state = snap.get("state", "") cat = _category_of(state, desired) try: self._db.execute( "INSERT OR REPLACE INTO tasks(task_key,info_hash,requirement_id,name," "machine,state,desired,category,progress,download_rate,upload_rate," "num_seeds,num_peers,num_files,completed_files,total_size,added_ts," "updated_ts,snapshot_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (key, str(snap.get("info_hash") or ""), int(snap.get("requirement_id") or 0), str(snap.get("name") or ""), str(snap.get("machine") or ""), state, desired, cat, float(snap.get("progress") or 0), float(snap.get("download_rate") or 0), float(snap.get("upload_rate") or 0), int(snap.get("num_seeds") or 0), int(snap.get("num_peers") or 0), int(snap.get("num_files") or 0), int(snap.get("completed_files") or 0), int(snap.get("total_size") or 0), float(self._added_ts.get(key, 0)), float(snap.get("_report_ts", 0)), json.dumps(snap, ensure_ascii=False))) self._db.commit() self._dirty.discard(key) except Exception as e: logger.warning(f"[process] 持久化失败 key={key}: {e}") def _delete_locked(self, key: str): try: self._db.execute("DELETE FROM tasks WHERE task_key=?", (key,)) self._db.commit() except Exception as e: logger.warning(f"[process] 删除持久化记录失败 key={key}: {e}") def _insert_events_locked(self, events: list, info_hash: str, rid: int, machine: str, name: str): """把 downloader 本轮上报的错误/告警事件写入 events 表（持久化）。""" try: rows = [] for ev in events: if not isinstance(ev, dict): continue rows.append(( float(ev.get("ts") or time.time()), info_hash, int(rid or 0), machine, name, str(ev.get("level") or "info"), str(ev.get("message") or "")[:1000])) if rows: self._db.executemany( "INSERT INTO events(ts,info_hash,requirement_id,machine,name," "level,message) VALUES(?,?,?,?,?,?,?)", rows) self._db.commit() except Exception as e: logger.warning(f"[process] 事件落库失败 info_hash={info_hash}: {e}") def list_events(self, rid=None, start: Optional[float] = None, end: Optional[float] = None, level: str = "", offset: int = 0, limit: int = 30, machine=None, q=None) -> dict: """按 requirement_id + 时间范围（事件发生时间）+ 级别 + 机器 + 搜索 分页查询历史事件。 搜索 q：匹配 info_hash 或 种子名称（name），用于“搜索某种子后只看它的报错消息”。 """ where = [] args: list = [] if rid not in (None, "", "all"): try: where.append("requirement_id=?"); args.append(int(rid)) except (TypeError, ValueError): pass if start is not None: where.append("ts>=?"); args.append(float(start)) if end is not None: where.append("ts<=?"); args.append(float(end)) if level: where.append("level=?"); args.append(str(level)) if machine not in (None, "", "all"): where.append("machine=?"); args.append(str(machine)) if q: ql = str(q).strip() if ql: like = f"%{ql}%" where.append("(info_hash LIKE ? OR name LIKE ? OR info_hash=?)") args.extend([like, like, ql]) wsql = (" WHERE " + " AND ".join(where)) if where else "" rows = [] total = 0 with self._lock: try: total = self._db.execute( "SELECT COUNT(*) FROM events" + wsql, args).fetchone()[0] cur = self._db.execute( "SELECT ts,info_hash,requirement_id,machine,name,level,message " "FROM events" + wsql + " ORDER BY ts DESC, id DESC LIMIT ? OFFSET ?", args + [int(limit) if limit else -1, int(offset)]) for ts, ih, r, mac, nm, lv, msg in cur.fetchall(): rows.append({ "ts": ts, "info_hash": ih, "requirement_id": r, "machine": mac, "name": nm, "level": lv, "message": msg, }) except Exception as e: logger.warning(f"[process] 查询事件失败: {e}") return {"rows": rows, "total": total, "offset": offset, "limit": limit} def _flush_loop(self): while not self._stop.wait(FLUSH_INTERVAL): try: with self._lock: dirty = list(self._dirty) for ih in dirty: with self._lock: self._persist_locked(ih) except Exception as e: logger.warning(f"[process] 刷盘循环异常: {e}") # -------------------------------------------------- 上报 / 控制 def report(self, snapshot: dict) -> str: info_hash = str(snapshot.get("info_hash") or "") if not info_hash: return DESIRED_RUNNING try: rid = int(snapshot.get("requirement_id") or 0) except (TypeError, ValueError): rid = 0 if not rid: logger.warning(f"[process] 上报缺少 requirement_id（必填）info_hash={info_hash} " f"machine={snapshot.get('machine')}") snapshot["requirement_id"] = rid key = _task_key(rid, info_hash) # 取出本轮事件（不随 snapshot_json 落库，单独进 events 表持久化） events = snapshot.pop("events", None) or [] now = time.time() with self._lock: if events: self._insert_events_locked( events, info_hash, rid, str(snapshot.get("machine") or ""), str(snapshot.get("name") or "")) prev = self._status.get(key) if key not in self._added_ts: self._added_ts[key] = now snapshot["_report_ts"] = now snapshot["_added_ts"] = self._added_ts[key] self._status[key] = snapshot desired = self._desired.get(key, DESIRED_RUNNING) # 状态变化 / 新任务立即落盘；纯进度变更攒批（后台刷盘） state_changed = (prev is None) or \ (prev.get("state") != snapshot.get("state")) self._dirty.add(key) if state_changed: self._persist_locked(key) return desired def set_control(self, task_key: str, action: str) -> bool: action = (action or "").lower() mapping = { "pause": DESIRED_PAUSED, "start": DESIRED_RUNNING, "resume": DESIRED_RUNNING, "delete": DESIRED_DELETED, "force_ingest": DESIRED_FORCE_INGEST, "force": DESIRED_FORCE_INGEST, } desired = mapping.get(action) if desired is None: return False with self._lock: self._desired[task_key] = desired if task_key in self._status: self._persist_locked(task_key) logger.info(f"[process] 控制指令 key={task_key} action={action} " f"-> desired={desired}") return True def dispatch_decision(self, info_hash: str, rid=0) -> dict: """重复投递同一 (requirement_id, info_hash) 时的处理决策（下载端开始前询问）。 去重按 (rid, info_hash)：同一需求下重复投递才 ignore/completed；不同需求复用 同一种子各自独立处理（本机已有完整数据则 downloader 走本地直接入库快路径）。 规则： - 本 rid 已完成 -> completed（跳过） - 本 rid 在线下载中/元数据/上传/等待 -> ignore（存活实例在跑） - 本 rid 在线已暂停（存活实例） -> ignore + desired=running（让其继续） - 停滞放弃 / 错误 -> proceed, mode=restart - 掉线的下载中/元数据/上传/等待 -> proceed, mode=continue（续传已下载部分） - 已删除 / 掉线-已完成残留 / 未见过 -> proceed """ if not info_hash: return {"decision": "proceed", "reason": "no-hash", "mode": "new"} key = _task_key(rid, info_hash) now = time.time() with self._lock: snap = self._status.get(key) mode = "new" if snap is not None: state = snap.get("state", "") offline = (now - snap.get("_report_ts", 0)) > OFFLINE_SECONDS if state == ST_COMPLETED and not offline: return {"decision": "completed", "reason": "completed", "mode": "skip"} if not offline and state in (ST_DOWNLOADING, ST_METADATA, ST_CHECKING, ST_UPLOADING, ST_PENDING): return {"decision": "ignore", "reason": f"downloading:{state}", "mode": "continue"} if not offline and state == ST_PAUSED: self._desired[key] = DESIRED_RUNNING self._persist_locked(key) return {"decision": "ignore", "reason": "resume-paused", "mode": "continue"} # 到这里都会 proceed，仅区分 continue / restart if state in (ST_STALLED, ST_ERROR): mode = "restart" elif state in (ST_DOWNLOADING, ST_METADATA, ST_CHECKING, ST_UPLOADING, ST_PENDING) and offline: mode = "continue" # 曾在下载但掉线：续传本地已有部分 else: mode = "restart" # proceed：占位 + 复位 desired if key not in self._added_ts: self._added_ts[key] = now self._desired[key] = DESIRED_RUNNING self._status[key] = { "info_hash": info_hash, "state": ST_PENDING, "name": "", "machine": "", "requirement_id": int(rid or 0), "_report_ts": now, "_added_ts": self._added_ts[key], "files": [], } self._persist_locked(key) reason = "restart" if snap is not None else "new" return {"decision": "proceed", "reason": reason, "mode": mode} def retry(self, task_keys: List[str]) -> dict: """重试，按“是否有下载进度”决定路由： - **有进度**（progress>0 或已完成部分文件）：原机本地有部分下载文件，入队到**原 下载机器**的本机重试队列，让原机复用本地文件续传/解压入库，避免同一种子在两台 机器各下一份。 - **无进度**（没下过、无本地数据）：直接重投到下载器 job_id(RETRY_JOB_ID)，由调度 分配到任意机器重新下载（不必绑原机）。 原始种子(raw_seed) 来自上报快照（被 write_seed_json_file 包装前的 seed 原文）。 """ now = time.time() queued_local = 0 missing = 0 remote_items: List[Tuple[str, str]] = [] # (task_key, raw_seed) with self._lock: for tk in task_keys: snap = self._status.get(tk) if not snap: continue raw = snap.get("raw_seed") or "" if not raw: missing += 1 continue machine = snap.get("machine") or "" has_progress = (float(snap.get("progress") or 0) > 0 or int(snap.get("completed_files") or 0) > 0) if has_progress and machine: # 有本地下载数据 -> 原机本地重跑（复用文件） self._retry_queue.setdefault(machine, []).append(str(raw)) try: self._db.execute( "INSERT INTO retry_queue(machine, raw_seed, ts) VALUES(?,?,?)", (machine, str(raw), now)) except Exception as e: logger.warning(f"[process] 重试队列持久化失败: {e}") snap["state"] = ST_REQUEUED snap["error"] = "" snap["_report_ts"] = now self._persist_locked(tk) queued_local += 1 else: # 无进度（或无机器记录）-> 重投远端 job_id，任意机器重新下载 remote_items.append((tk, str(raw))) if queued_local: try: self._db.commit() except Exception: pass # 无进度的：批量重投到下载器 job_id remote_n = 0 if remote_items: try: try: from .common import XunLiuClient, split_by_size except ImportError: from common import XunLiuClient, split_by_size client = XunLiuClient() seeds = [raw for _tk, raw in remote_items] for chunk in split_by_size(seeds, 40 * 1024 * 1024): client.upload_seed(RETRY_JOB_ID, chunk) remote_n = len(seeds) with self._lock: for tk, _raw in remote_items: snap = self._status.get(tk) if snap: snap["state"] = ST_REQUEUED snap["error"] = "" snap["_report_ts"] = now self._persist_locked(tk) except Exception as e: logger.error(f"[process] 无进度重试重投 job_id={RETRY_JOB_ID} 失败: {e}") if queued_local == 0: return {"ok": False, "requeued": 0, "missing": missing, "error": f"重投远端失败: {e}"} total = queued_local + remote_n if total == 0: reason = "缺少原始种子 raw_seed" if missing else "无匹配任务" return {"ok": False, "requeued": 0, "missing": missing, "error": f"无可重试任务（{reason}）"} logger.info(f"[process] 重试 {total} 个：有进度原机重跑 {queued_local}，" f"无进度重投 job_id {remote_n}（缺原始种子 {missing}）") return {"ok": True, "requeued": total, "local": queued_local, "remote": remote_n, "missing": missing} def claim_retries(self, machine: str) -> List[str]: """原机 downloader 轮询取回属于本机的待重试原始种子（取出即清空，避免重复）。""" if not machine: return [] with self._lock: seeds = self._retry_queue.pop(machine, []) if seeds: try: # 取出即从 DB 清除，避免重复重跑 self._db.execute( "DELETE FROM retry_queue WHERE machine=?", (machine,)) self._db.commit() except Exception as e: logger.warning(f"[process] 清除重试队列失败: {e}") if seeds: logger.info(f"[process] 机器 {machine} 取回 {len(seeds)} 个本机重试种子") return list(seeds) def remove(self, task_key: str) -> None: with self._lock: self._status.pop(task_key, None) self._desired.pop(task_key, None) self._added_ts.pop(task_key, None) self._dirty.discard(task_key) self._delete_locked(task_key) # -------------------------------------------------- 查询（供接口） def _passes(self, snap: dict, rid, start: Optional[float], end: Optional[float], machine=None, q=None) -> bool: if rid not in (None, "", "all"): try: if int(snap.get("requirement_id") or 0) != int(rid): return False except (TypeError, ValueError): return False ats = float(snap.get("_added_ts", 0) or 0) if start is not None and ats < start: return False if end is not None and ats > end: return False if machine not in (None, "", "all"): if (snap.get("machine") or "") != machine: return False if q: ql = str(q).strip().lower() if ql: hay = " ".join(( str(snap.get("info_hash") or ""), str(snap.get("info_hash_v2") or ""), str(snap.get("name") or ""), )).lower() if ql not in hay: return False return True def _light_row(self, key: str, snap: dict, desired: str, now: float) -> dict: state = snap.get("state", "") offline = (now - snap.get("_report_ts", 0)) > OFFLINE_SECONDS return { "task_key": key, "info_hash": snap.get("info_hash", "") or "", "requirement_id": int(snap.get("requirement_id") or 0), "name": snap.get("name", "") or "", "machine": snap.get("machine", "") or "", "state": state, "state_label": STATE_LABELS.get(state, state), "desired": desired, "offline": offline, "progress": float(snap.get("progress") or 0), "download_rate": float(snap.get("download_rate") or 0), "upload_rate": float(snap.get("upload_rate") or 0), "num_seeds": int(snap.get("num_seeds") or 0), "num_peers": int(snap.get("num_peers") or 0), "num_files": int(snap.get("num_files") or 0), "completed_files": int(snap.get("completed_files") or 0), "total_size": int(snap.get("total_size") or 0), "downloaded": int(snap.get("downloaded") or 0), "uploaded": int(snap.get("uploaded") or 0), "added_ts": float(snap.get("_added_ts", 0) or 0), "updated_ts": float(snap.get("_report_ts", 0) or 0), } def summary(self, rid=None, start: Optional[float] = None, end: Optional[float] = None, machine=None, q=None) -> dict: counts = {k: 0 for k in GROUP_KEYS} total_rate = 0.0 seeding_count = 0 seed_up_rate = 0.0 success_dl_bytes = 0 # 已成功下载的数据总量（下载完成的种子体量，按 info_hash 去重） _counted_ih: set = set() # 去重：多 rid 复用同一 info_hash 只计一次物理下载量 machines: Dict[str, dict] = {} rids: set = set() now = time.time() with self._lock: for key, snap in self._status.items(): r = int(snap.get("requirement_id") or 0) if r: rids.add(r) # 候选下拉：全量（不受筛选影响） # 机器卡片：按 rid/时间/搜索 过滤，但**不含机器过滤**，这样切换机器时 # 所有机器卡片仍在、可点击切换。 if not self._passes(snap, rid, start, end, machine=None, q=q): continue mn = snap.get("machine") or "(未知)" m = machines.setdefault(mn, {"name": mn, "tasks": 0, "disk": None, "cpu": None, "mem": None, "_stat_ts": 0}) dk = snap.get("disk") if dk and (m["disk"] is None or float(dk.get("percent") or 0) > float(m["disk"].get("percent") or 0)): m["disk"] = dk # CPU/内存为整机指标：取该机最近一次上报的值 rts = float(snap.get("_report_ts", 0) or 0) if rts >= float(m["_stat_ts"] or 0) and \ (snap.get("cpu_percent") is not None or snap.get("mem_percent") is not None): m["cpu"] = snap.get("cpu_percent") m["mem"] = snap.get("mem_percent") m["_stat_ts"] = rts # 计数 / 速度 / 流量：叠加机器过滤（选中机器时只统计该机） if machine not in (None, "", "all") and mn != machine: continue m["tasks"] += 1 desired = self._desired.get(key, DESIRED_RUNNING) state = snap.get("state", "") offline = (now - snap.get("_report_ts", 0)) > OFFLINE_SECONDS cat = _category_of(state, desired, offline) counts[cat] = counts.get(cat, 0) + 1 # 已成功下载的数据总量：整种已下完的状态才计入（做种/已入库完成/已完成/ # 存储不足删除——均为已下完再入库）；按 info_hash 去重避免多 rid 复用重复计。 if state in (ST_COMPLETED, ST_SEEDING, ST_INGESTED_DONE, ST_DELETED_NO_SPACE): ih = str(snap.get("info_hash") or "") if ih and ih not in _counted_ih: _counted_ih.add(ih) success_dl_bytes += int(snap.get("total_size") or 0) if state == ST_DOWNLOADING and not offline: total_rate += float(snap.get("download_rate") or 0) if state == ST_SEEDING and not offline: seeding_count += 1 seed_up_rate += float(snap.get("upload_rate") or 0) return { "counts": counts, "total_rate": total_rate, "seeding_count": seeding_count, "seed_up_rate": seed_up_rate, "success_dl_bytes": success_dl_bytes, "machines": machines, "rids": sorted(rids), "updated": now, } def list_page(self, category: str, rid=None, start: Optional[float] = None, end: Optional[float] = None, offset: int = 0, limit: int = 20, machine=None, q=None) -> dict: now = time.time() items: List[Tuple[float, str, dict, str]] = [] with self._lock: for key, snap in self._status.items(): if not self._passes(snap, rid, start, end, machine=machine, q=q): continue desired = self._desired.get(key, DESIRED_RUNNING) offline = (now - snap.get("_report_ts", 0)) > OFFLINE_SECONDS cat = _category_of(snap.get("state", ""), desired, offline) if category and category != "all" and cat != category: continue items.append((float(snap.get("_added_ts", 0) or 0), key, snap, desired)) items.sort(key=lambda x: x[0], reverse=True) # 添加时间倒序 total = len(items) page = items[offset:offset + limit] if limit else items[offset:] rows = [self._light_row(key, snap, desired, now) for _ats, key, snap, desired in page] return {"rows": rows, "total": total, "offset": offset, "limit": limit} def export_rows(self, category: str, rid=None, start: Optional[float] = None, end: Optional[float] = None, machine=None, q=None) -> List[dict]: return self.list_page(category, rid, start, end, offset=0, limit=0, machine=machine, q=q)["rows"] def get_task(self, task_key: str) -> Optional[dict]: now = time.time() with self._lock: snap = self._status.get(task_key) if snap is None: return None s = dict(snap) s["task_key"] = task_key s["desired"] = self._desired.get(task_key, DESIRED_RUNNING) s["offline"] = (now - snap.get("_report_ts", 0)) > OFFLINE_SECONDS state = snap.get("state", "") s["state_label"] = STATE_LABELS.get(state, state) return s def all_status(self) -> Dict[str, dict]: """全量状态（兼容旧 /api/status 与控制台展示）。""" with self._lock: out = {} now = time.time() for key, snap in self._status.items(): s = dict(snap) s["task_key"] = key s["desired"] = self._desired.get(key, DESIRED_RUNNING) s["offline"] = (now - snap.get("_report_ts", 0)) > OFFLINE_SECONDS state = s.get("state", "") s["state_label"] = STATE_LABELS.get(state, state) out[key] = s return out STORE = MonitorStore() # ============================================================================ # HTTP 服务端 # ============================================================================ def _render_dashboard() -> str: """看板静态页：数据由前端按分类分页 fetch，无限下拉懒加载，静默增量刷新。""" return """ torrent 下载监控

torrent 下载监控

~
总下载速度
-
已成功下载数据总量
-
下载中
-
做种中
-
已全部入库完成
-
已暂停
-
停滞放弃
-
强制入库
-
重投队列
-
其他特殊情况
-
存储不足取消
-
存储不足删除
-
已删除
-
更新时间
-
加载中…
""" class _Handler(BaseHTTPRequestHandler): def log_message(self, format, *args): # 静默默认访问日志 pass def _send_json(self, code: int, obj: dict): data = json.dumps(obj, ensure_ascii=False).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _send_html(self, html: str): data = html.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _send_csv(self, filename: str, text: str): data = text.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/csv; charset=utf-8") self.send_header("Content-Disposition", f'attachment; filename="{filename}"') self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _read_body(self) -> dict: length = int(self.headers.get("Content-Length", 0) or 0) if length <= 0: return {} try: return json.loads(self.rfile.read(length).decode("utf-8")) except Exception: return {} @staticmethod def _q1(qs: dict, key: str, default=None): v = qs.get(key, [default]) return v[0] if v else default def _handle_export(self, qs: dict): category = str(self._q1(qs, "category", "all") or "all") rid = self._q1(qs, "requirement_id", None) start = _to_float(self._q1(qs, "start", None)) end = _to_float(self._q1(qs, "end", None)) machine = self._q1(qs, "machine", None) q = self._q1(qs, "q", None) rows = STORE.export_rows(category, rid, start, end, machine=machine, q=q) buf = io.StringIO() buf.write("\ufeff") # BOM，便于 Excel 正确识别中文 w = csv.writer(buf) w.writerow(["info_hash", "requirement_id", "name", "machine", "state", "progress", "download_rate_Bps", "upload_rate_Bps", "num_seeds", "num_peers", "completed_files", "num_files", "total_size", "added_time", "updated_time"]) def _ts(t): try: return datetime.datetime.fromtimestamp(float(t)).strftime( "%Y-%m-%d %H:%M:%S") if t else "" except Exception: return "" for r in rows: w.writerow([ r["info_hash"], r["requirement_id"], r["name"], r["machine"], r["state_label"], f'{r["progress"]*100:.1f}%', int(r["download_rate"]), int(r["upload_rate"]), r["num_seeds"], r["num_peers"], r["completed_files"], r["num_files"], r["total_size"], _ts(r["added_ts"]), _ts(r["updated_ts"])]) fname = f"torrent_{category}_{int(time.time())}.csv" self._send_csv(fname, buf.getvalue()) def do_GET(self): parsed = urlparse(self.path) qs = parse_qs(parsed.query) path = parsed.path if path in ("/", "/index.html"): self._send_html(_render_dashboard()) elif path == "/api/summary": self._send_json(200, STORE.summary( self._q1(qs, "requirement_id", None), _to_float(self._q1(qs, "start", None)), _to_float(self._q1(qs, "end", None)), machine=self._q1(qs, "machine", None), q=self._q1(qs, "q", None))) elif path == "/api/list": try: offset = int(self._q1(qs, "offset", 0) or 0) except (TypeError, ValueError): offset = 0 try: limit = int(self._q1(qs, "limit", 20) or 20) except (TypeError, ValueError): limit = 20 limit = max(0, min(limit, 1000000)) self._send_json(200, STORE.list_page( str(self._q1(qs, "category", "all") or "all"), self._q1(qs, "requirement_id", None), _to_float(self._q1(qs, "start", None)), _to_float(self._q1(qs, "end", None)), offset, limit, machine=self._q1(qs, "machine", None), q=self._q1(qs, "q", None))) elif path == "/api/task": task = STORE.get_task((self._q1(qs, "task_key", "") or "").strip()) if task is None: self._send_json(404, {"error": "not found"}) else: self._send_json(200, {"task": task}) elif path == "/api/events": try: offset = int(self._q1(qs, "offset", 0) or 0) except (TypeError, ValueError): offset = 0 try: limit = int(self._q1(qs, "limit", 30) or 30) except (TypeError, ValueError): limit = 30 limit = max(0, min(limit, 1000)) self._send_json(200, STORE.list_events( self._q1(qs, "requirement_id", None), _to_float(self._q1(qs, "start", None)), _to_float(self._q1(qs, "end", None)), str(self._q1(qs, "level", "") or ""), offset, limit, machine=self._q1(qs, "machine", None), q=self._q1(qs, "q", None))) elif path == "/api/export": self._handle_export(qs) elif path == "/api/status": self._send_json(200, {"status": STORE.all_status()}) elif path == "/api/dispatch": info_hash = (self._q1(qs, "info_hash", "") or "").strip() try: rid_q = int(self._q1(qs, "requirement_id", 0) or 0) except (TypeError, ValueError): rid_q = 0 self._send_json(200, STORE.dispatch_decision(info_hash, rid_q)) elif path == "/api/pending_retries": machine = (self._q1(qs, "machine", "") or "").strip() self._send_json(200, {"seeds": STORE.claim_retries(machine)}) else: self._send_json(404, {"error": "not found"}) def do_POST(self): parsed = urlparse(self.path) body = self._read_body() if parsed.path == "/api/report": desired = STORE.report(body) self._send_json(200, {"desired": desired}) elif parsed.path == "/api/control": ok = STORE.set_control(str(body.get("task_key") or ""), str(body.get("action") or "")) self._send_json(200 if ok else 400, {"ok": ok}) elif parsed.path == "/api/retry": keys = body.get("task_keys") if not keys: one = str(body.get("task_key") or "") keys = [one] if one else [] keys = [str(k) for k in keys if k] res = STORE.retry(keys) self._send_json(200 if res.get("ok") else 400, res) else: self._send_json(404, {"error": "not found"}) def serve(host: str = "0.0.0.0", port: int = PROCESS_SERVER_PORT, console_display: bool = True): """启动 HTTP 监控/控制服务（阻塞）。""" if console_display: threading.Thread(target=_console_display_loop, daemon=True).start() server = ThreadingHTTPServer((host, port), _Handler) logger.info(f"[process] 监控服务启动: http://{host}:{port} " f"（对外 {PROCESS_SERVER_URL}），DB={DB_PATH}") try: server.serve_forever() except KeyboardInterrupt: logger.info("[process] 收到中断，关闭服务") finally: server.shutdown() def _console_display_loop(interval: float = 10.0): while True: time.sleep(interval) try: status = STORE.all_status() if not status: continue total_rate = sum(float(s.get("download_rate") or 0) for s in status.values() if s.get("state") == ST_DOWNLOADING and not s.get("offline")) lines = [f"===== 监控总览 任务={len(status)} 总速度={_fmt_speed(total_rate)} " f"{time.strftime('%H:%M:%S')} ====="] for _key, s in sorted(status.items()): files = s.get("files") or [] completed = sum(1 for f in files if f.get("status") == ST_COMPLETED) st_txt = "离线" if s.get("offline") else s.get("state_label") or s.get("state", "") ih = str(s.get("info_hash") or "") lines.append( f" {ih[:12]} rid={s.get('requirement_id',0)} " f"{(s.get('name') or '')[:24]} " f"{float(s.get('progress') or 0)*100:.1f}% " f"{_fmt_speed(s.get('download_rate') or 0)} " f"seeds={s.get('num_seeds',0)} peers={s.get('num_peers',0)} " f"files={completed}/{s.get('num_files', len(files))} " f"{st_txt}/{s.get('desired','')}") logger.info("\n".join(lines)) except Exception as e: logger.warning(f"[process] 控制台展示异常: {e}") # ============================================================================ # 客户端（downloader 侧使用） # ============================================================================ class ProcessReporterClient: """downloader 用来向 process 服务上报状态并获取控制期望态。""" def __init__(self, server_url: str = PROCESS_SERVER_URL, timeout: float = 8.0): self.server_url = server_url.rstrip("/") self.timeout = timeout self._session = requests.Session() self._warned = False def report(self, snapshot: dict) -> str: """上报快照，返回该任务的 desired 状态；服务不可达时返回 running。""" try: resp = self._session.post( self.server_url + "/api/report", json=snapshot, timeout=self.timeout) if resp.status_code == 200: return str(resp.json().get("desired") or DESIRED_RUNNING) except Exception as e: if not self._warned: logger.warning(f"[reporter] 无法连接监控服务 {self.server_url}: {e}" f"（后续将静默重试）") self._warned = True return DESIRED_RUNNING def control(self, info_hash: str, action: str, rid=0) -> bool: try: resp = self._session.post( self.server_url + "/api/control", json={"task_key": _task_key(rid, info_hash), "action": action}, timeout=self.timeout) return resp.status_code == 200 except Exception: return False def dispatch_decision(self, info_hash: str, rid=0) -> str: """询问该 (rid, info_hash) 是否应处理：返回 'proceed' / 'ignore' / 'completed'。 服务不可达时返回 'proceed'（不因监控缺失而阻塞下载）。响应里还带 mode （continue/restart/skip/new）作为附加信息，此处仅回传 decision 字符串保持 与既有 downloader 调用兼容。 """ try: resp = self._session.get( self.server_url + "/api/dispatch", params={"info_hash": info_hash, "requirement_id": rid}, timeout=self.timeout) if resp.status_code == 200: return str(resp.json().get("decision") or "proceed") except Exception: pass return "proceed" def claim_retries(self, machine: str) -> list: """取回属于本机的待重试原始种子（原机本地重跑用）；服务不可达返回空列表。""" try: resp = self._session.get( self.server_url + "/api/pending_retries", params={"machine": machine}, timeout=self.timeout) if resp.status_code == 200: return list(resp.json().get("seeds") or []) except Exception: pass return [] if __name__ == "__main__": serve()