# -*- coding: utf-8 -*- """任务生命周期原语(启动 / 终态 / 文件收尾 / 结算)。 (自 services/runtime.py 拆出,行为与原实现一致。) """ import threading import time import traceback from typing import Any import auth_system from core.state import ( DOWNLOAD_TASKS, DOWNLOAD_TASKS_LOCK, ) from .api_service import ( api_service, ) def _start_task_thread(task_id: str, target, args=()) -> str | None: """启动任务 worker(启动失败时收敛任务终态并按账本退费,不留僵尸 queued 记录)。 退款只走账本(本处 settle 后装饰器看到账已关,不会重复退)。 线程句柄存入任务字典——监控循环对 4h 超时任务若发现 worker 线程已死 (无法自行走到终态结算),可代为按账本结算退款(settle 幂等,无双退)。""" try: th = threading.Thread(target=target, args=args, daemon=True) with DOWNLOAD_TASKS_LOCK: t = DOWNLOAD_TASKS.get(task_id) if t is not None: t['worker_thread'] = th th.start() return None except Exception as e: with DOWNLOAD_TASKS_LOCK: t = DOWNLOAD_TASKS.pop(task_id, None) if t: try: auth_system.ledger_settle(task_id, 0, 'submit_failed') except Exception: pass api_service.logger.error(f"任务线程启动失败: {task_id} {e}") return f'任务启动失败:{e}' def _mark_task_terminal(task_id: str, status: str, msg: str = '') -> None: """任务进入终态的统一入口(记录 completed_at,保留期从完成时刻起算, 而不是从提交时刻起算)。 /F-E:置 error/canceled 终态时若 worker 线程已死(无人再走正常结算), 代为按账本结算退款——ledger_settle 幂等,worker 已自行结算过则此处为 0。 终态先到先得——已终态(done/error/canceled)的任务 不再被覆盖(堵住 canceled→error 的非法迁移边:取消后解析阶段的自身失败 不得把"已取消"改写成"失败",completed_at 也不再刷新)。""" need_settle = False with DOWNLOAD_TASKS_LOCK: t = DOWNLOAD_TASKS.get(task_id) if t and t.get('status') in ('done', 'error', 'canceled'): return if t: t['status'] = status t['msg'] = msg or t.get('msg', '') t['completed_at'] = time.time() if status in ('canceled', 'error'): th = t.get('worker_thread') if th is not None and not th.is_alive(): need_settle = True # settle 移到任务锁外——ledger_settle 内部要抢 _DB_LOCK 写库 # (busy 最长 15s),持任务锁调用会让全站进度/提交/取消/领取随库停摆; # settle 幂等,锁外调用无双退 if need_settle: try: auth_system.ledger_settle(task_id, 0, status) except Exception: pass def _task_files(task: dict[str, Any]) -> list[str]: """任务当前关联的可清理文件路径(交付 ZIP/交付副本;缓存文件不在此列)""" result = task.get('result') or {} files = [] for f in (result.get('zip_path'), result.get('file_path')): if f: files.append(f) return files def _settle_batch_billing(task_id: str, ok_count: int, final_status: str = 'settled') -> int: """批量任务结算(走持久账本,按实际预扣-实际交付幂等退款; 重复调用/重启后调用都只会退一次)""" try: refund = auth_system.ledger_settle(task_id, ok_count, final_status) except Exception: # 本函数多在异常处理器内被调用——结算再抛错(库忙超时) # 会杀死 worker 收尾路径,跳过终态标记;失败留周期对账兜底(幂等) api_service.logger.error( f"批量结算异常(留对账兜底): {task_id}\n{traceback.format_exc()}") return 0 if refund: try: with DOWNLOAD_TASKS_LOCK: t = DOWNLOAD_TASKS.get(task_id) or {} uid = int(t.get('billing_user_id') or 0) u = auth_system.get_user_by_id(uid) if uid else None uname = u['username'] if u else 'system' remaining = u['duration_seconds'] if u else '?' auth_system.add_log('success', uname, 'system', '批量退费', f'失败部分 {refund} 次已退回,剩余 {remaining} 次') except Exception: pass return refund