NeteaseUrl/services/tasks.py

# -*- coding: utf-8 -*-
"""任务监控循环与设备心跳。

监控循环原是一个 170 行 / 圈复杂度 58 / 嵌套 8 层的单函数，内含 7 组互不相关
的周期性作业。此处按「一组作业一个具名函数」拆开：改超时取消时不必通读账本
对账与缓存清扫，每项作业的异常兜底粒度也一目了然（见各自 docstring）。
"""

import os
import shutil
import threading
import time
import traceback

import auth_system
from core.config import (
    CANCEL_AFTER_SECONDS,
    HEARTBEAT_INTERVAL,
    PAUSE_AFTER_SECONDS,
    RESULT_RETENTION_SECONDS,
)
from core.state import (
    DEVICE_LAST_SEEN,
    DEVICE_LAST_SEEN_LOCK,
    DOWNLOAD_TASKS,
    DOWNLOAD_TASKS_LOCK,
)

from ._shared import (
    _LEDGER_LAST_CLEANUP,
    _LEDGER_LAST_RECOVER,
    _MONITOR_ERR_TS,
    _TASKS_SWEEP_LAST,
    _sweep_rate_buckets,
)
from ._taskcore import (
    _task_files,
)
from .api_service import (
    api_service,
)

# ---- 监控循环各项作业的周期与阈值（集中一处，避免魔数散落在循环体里）----

# 任务硬超时：运行/排队/暂停超过该时长仍未结束 → 置取消
_TASK_HARD_TIMEOUT = 4 * 3600

# 终态任务结果文件清理的最大重试轮数（文件被占用时下轮再试）
_CLEANUP_MAX_TRIES = 8

# 账本周期兜底：孤儿对账每 60s、settled 行清理每 24h
_LEDGER_RECOVER_INTERVAL = 60
_LEDGER_CLEANUP_INTERVAL = 24 * 3600

# 孤儿账退回阈值：严格大于任务 4h 硬超时 + 监控 tick 抖动——账本行 created_at
# 整秒截断使账龄恒 ≥ 任务龄，旧的「4h 相等」口径存在抢退合法 4h+ 长任务的窗口
_LEDGER_ORPHAN_MAX_AGE = 5 * 3600

# settled 账本行保留 90 天后清理（防长运行进程下表无界增长）
_LEDGER_SETTLED_RETENTION = 90 * 86400

# tasks/ 孤儿目录补扫：每小时一次，仅回收 >24h 且无任务记录的目录
_TASKS_SWEEP_INTERVAL = 3600
_TASKS_ORPHAN_MAX_AGE = 24 * 3600

# 监控异常限频窗口（秒）：持续异常不静默，也不每 tick 刷屏
_MONITOR_ERR_WINDOW = 600


def _quiesce_tasks_for_restore() -> None:
    """恢复数据库前取消全部在途任务（worker 不再引用旧库用户结算）"""
    with DOWNLOAD_TASKS_LOCK:
        for task in DOWNLOAD_TASKS.values():
            if task.get('status') in ('queued', 'running', 'paused'):
                task['status'] = 'canceled'
                task['msg'] = '数据库恢复中，任务已取消（费用按实际交付结算）'
                task['completed_at'] = time.time()
                try:
                    task['cancel'].set()
                except Exception:
                    pass


# 数据库恢复前静默在途任务
auth_system.register_before_restore_hook(_quiesce_tasks_for_restore)


def _sweep_download_cache(max_age_seconds: int = RESULT_RETENTION_SECONDS) -> None:
    """下载缓存清扫（配套：批量任务不再即时删除共享缓存，改由这里按年龄回收）。
    只清扫 downloads 根目录的普通文件，不碰 tasks/ 子目录（那是任务专属交付物）。"""
    try:
        root = api_service.downloads_path
        now = time.time()
        for p in root.iterdir():
            try:
                if p.is_file() and now - p.stat().st_mtime > max_age_seconds:
                    p.unlink()
            except OSError:
                pass
    except Exception:
        pass


def _reap_finished_tasks(now: float) -> None:
    """终态任务回收：按 completed_at 保留 RESULT_RETENTION_SECONDS 后删记录与文件。

    锁内只收集过期清单，文件删除与 rmtree 移到锁外——GB 级任务目录的删除曾
    阻塞任务锁数十秒，期间全站进度上报/提交/取消/领取全部停摆。

    单条删除失败保留记录下轮重试，最多 _CLEANUP_MAX_TRIES 轮；超限后放弃记录
    但大声记日志，文件留给 _sweep_download_cache 按年龄兜底。
    """
    expired: list[tuple] = []
    with DOWNLOAD_TASKS_LOCK:
        for tid, task in DOWNLOAD_TASKS.items():
            if task.get('status') not in ('done', 'error', 'canceled'):
                continue
            done_at = task.get('completed_at') or 0
            # 兼容无 completed_at 的旧记录：回退到 start
            if not done_at:
                done_at = task.get('start', 0)
            if now - done_at <= RESULT_RETENTION_SECONDS:
                continue
            expired.append((tid, _task_files(task),
                            api_service.downloads_path / 'tasks' / tid))
    if not expired:
        return

    cleanup_results: list[tuple] = []
    for tid, files, task_dir in expired:
        all_removed = True
        for f in files:
            try:
                os.remove(f)
            except FileNotFoundError:
                pass
            except OSError:
                all_removed = False   # 文件暂时不可删（被占用等）
        if task_dir.exists():
            shutil.rmtree(str(task_dir), ignore_errors=True)
            if task_dir.exists():
                all_removed = False
        cleanup_results.append((tid, all_removed))

    finished_done: list[str] = []
    finished_retry: list[str] = []
    with DOWNLOAD_TASKS_LOCK:
        for tid, all_removed in cleanup_results:
            task = DOWNLOAD_TASKS.get(tid)
            if task is None:
                finished_done.append(tid)   # 锁外期间已被其他路径移除
                continue
            if all_removed:
                finished_done.append(tid)
            else:
                task['_cleanup_tries'] = int(task.get('_cleanup_tries') or 0) + 1
                if task['_cleanup_tries'] >= _CLEANUP_MAX_TRIES:
                    # 重试多次仍失败：放弃记录但大声记录日志
                    api_service.logger.warning(
                        f"任务 {tid} 结果文件多次清理失败，已放弃（文件可能被占用）")
                    finished_done.append(tid)
                else:
                    finished_retry.append(tid)
        for tid in finished_done:
            DOWNLOAD_TASKS.pop(tid, None)
    if finished_done:
        api_service.logger.info(f"监控线程清理过期终态任务: {finished_done}")
    if finished_retry:
        api_service.logger.info(f"监控线程保留清理失败任务待重试: {finished_retry}")


def _cancel_overtime_tasks(now: float) -> None:
    """运行/排队/暂停超过硬上限仍未结束的任务 → 置取消并唤醒 cancel Event。

    worker 已结束则无人再走终态结算，监控代为按账本退款——ledger_settle 幂等，
    worker 若已结算过此处返回 0，无双退。结算移到任务锁外（同 _mark_task_terminal
    口径）：库忙 15s 不再随任务锁波及全站任务端点。
    """
    timeout_settle: list[str] = []
    with DOWNLOAD_TASKS_LOCK:
        for tid, task in DOWNLOAD_TASKS.items():
            if (task.get('status') in ('queued', 'running', 'paused')
                    and now - task.get('start', 0) > _TASK_HARD_TIMEOUT):
                task['status'] = 'canceled'
                task['msg'] = '任务超时未完成，已自动取消（费用按实际交付结算）'
                task['completed_at'] = time.time()
                try:
                    task['cancel'].set()
                except Exception:
                    pass
                # worker 已结束则无人再走终态结算——监控代为按账本结算退款
                wt = task.get('worker_task')
                if wt is not None and wt.done():
                    timeout_settle.append(tid)
                api_service.logger.warning(f"任务超时自动取消: {tid}")
    for tid in timeout_settle:
        try:
            auth_system.ledger_settle(tid, 0, 'timeout')
        except Exception:
            pass


def _run_periodic_ledger_jobs(now: float) -> None:
    """账本周期兜底（原实现仅启动执行一次）。

    - 孤儿对账：open 且账龄 > _LEDGER_ORPHAN_MAX_AGE 的行每分钟检查并退回
    - settled 行清理：每小时低频执行，防长运行进程下表无界增长

    用「时间戳门槛」而非「tick 取模」——单轮工作段超过周期时取模落点会被整段
    跳过，时间戳门槛只延迟不漏跑。

    自带 try/except：账本库忙不该让本轮后续作业（孤儿目录补扫、心跳离线处理）
    一起落空。
    """
    try:
        if now - _LEDGER_LAST_RECOVER['ts'] > _LEDGER_RECOVER_INTERVAL:
            _LEDGER_LAST_RECOVER['ts'] = now
            auth_system.ledger_recover_orphans(max_age_seconds=_LEDGER_ORPHAN_MAX_AGE)
        if now - _LEDGER_LAST_CLEANUP['ts'] > _LEDGER_CLEANUP_INTERVAL:
            _LEDGER_LAST_CLEANUP['ts'] = now
            auth_system.ledger_cleanup_settled(_LEDGER_SETTLED_RETENTION)
    except Exception as e:
        api_service.logger.warning(f"账本周期兜底异常（下轮重试）: {e}")


def _sweep_orphan_task_dirs(now: float) -> None:
    """tasks/ 孤儿目录每小时低频补扫：>24h 且无任务记录才移除。

    与 _reap_finished_tasks 互补——那条路径由任务记录驱动，记录已丢的目录
    （进程曾在删除前崩溃等）只能靠这里按年龄兜底。

    自带 try/except：同 _run_periodic_ledger_jobs，单条竞态与整体异常都不该
    打断本轮后续作业。
    """
    try:
        if now - _TASKS_SWEEP_LAST['ts'] <= _TASKS_SWEEP_INTERVAL:
            return
        _TASKS_SWEEP_LAST['ts'] = now
        tasks_root = api_service.downloads_path / 'tasks'
        if not tasks_root.exists():
            return
        for d in tasks_root.iterdir():
            try:
                if (d.is_dir() and d.name not in DOWNLOAD_TASKS
                        and now - d.stat().st_mtime > _TASKS_ORPHAN_MAX_AGE):
                    shutil.rmtree(str(d), ignore_errors=True)
                    api_service.logger.info(f"tasks 孤儿目录补扫移除: {d.name}")
            except FileNotFoundError:
                pass   # 单条竞态不中止整轮扫描
            except OSError:
                pass
    except Exception as e:
        api_service.logger.warning(f"tasks 孤儿目录补扫异常（下轮重试）: {e}")


def _handle_offline_devices(now: float) -> None:
    """心跳离线处理（两档，阈值见 core.config）。

    - 离线 > CANCEL_AFTER_SECONDS（30 分钟）：取消该设备所有未完成任务。文件删除
      与退款统一交给 worker 的取消收尾/终态清理；worker 已不在时由启动对账按账本兜底。
    - 离线 > PAUSE_AFTER_SECONDS：批量任务先置暂停，让出排队名额给下一位
      （「断线不占坑」）。
    """
    with DEVICE_LAST_SEEN_LOCK:
        offline = {d: now - ts for d, ts in DEVICE_LAST_SEEN.items()
                   if now - ts > PAUSE_AFTER_SECONDS}
    if not offline:
        return
    with DOWNLOAD_TASKS_LOCK:
        for tid, task in DOWNLOAD_TASKS.items():
            dev = task.get('device_id')
            gap = offline.get(dev) if dev else None
            if not gap:
                continue
            st = task.get('status')
            if gap > CANCEL_AFTER_SECONDS and st in ('queued', 'running', 'paused'):
                # 离线超 30 分钟：请求取消（文件删除与退款统一交给 worker 的
                # 取消收尾/终态清理；worker 已不在时由启动对账按账本兜底）
                task['status'] = 'canceled'
                task['msg'] = '设备离线超过 30 分钟，任务已自动取消'
                task['completed_at'] = time.time()
                try:
                    task['cancel'].set()
                except Exception:
                    pass
                api_service.logger.info(f"设备离线超 30 分钟自动请求取消: {tid}")
            elif task.get('type') == 'batch' and st in ('queued', 'running'):
                # 心跳超时：先暂停，让出排队名额给下一位
                task['status'] = 'paused'
                task['msg'] = '设备离线，任务暂停（重新打开页面自动恢复）'
                task['pause'].set()


def _log_monitor_error() -> None:
    """监控循环异常限频记录。

    持续异常不能完全静默（否则监控失能无人知晓），也不能每 tick 刷屏：每
    _MONITOR_ERR_WINDOW 秒最多一条，附堆栈尾部。

    必须在 except 块内**直接**调用——traceback.format_exc() 取的是当前活跃异常，
    转手排队/异步回调后再调用会拿到 NoneType 堆栈。
    """
    now = time.time()
    _MONITOR_ERR_TS['n'] += 1
    if now - _MONITOR_ERR_TS['ts'] > _MONITOR_ERR_WINDOW:
        api_service.logger.warning(
            f"任务监控循环异常（近 10 分钟累计 {_MONITOR_ERR_TS['n']} 次，"
            f"最后异常见堆栈）: {traceback.format_exc()[-600:]}")
        _MONITOR_ERR_TS['ts'] = now
        _MONITOR_ERR_TS['n'] = 0


def _run_monitor_tick(now: float) -> None:
    """单轮监控作业（7 步，顺序即优先级）。

    先回收已终态任务腾出名额，再处理超时，最后处理离线设备——顺序与拆分前一致。

    ⚠️ 参数 ``now`` 由调用方传入（其余作业内部各自取 time.time()）：
    _reap_finished_tasks 与 _cancel_overtime_tasks 共用同一轮的时间基准，避免
    同一轮里两次取时导致边界任务在两处判断不一致。
    """
    _reap_finished_tasks(now)
    _cancel_overtime_tasks(now)
    _sweep_download_cache(RESULT_RETENTION_SECONDS)
    _sweep_rate_buckets()
    _run_periodic_ledger_jobs(now)
    _sweep_orphan_task_dirs(now)
    _handle_offline_devices(now)


def _task_monitor_loop() -> None:
    """任务监控循环：每 HEARTBEAT_INTERVAL 跑一轮 _run_monitor_tick。

    单轮里任何一步抛异常都由 _log_monitor_error 限频记录，循环本身不退出
    （保持拆分前的兜底粒度：拆分前整个函数体共用一个 try/except）。
    注意 _run_periodic_ledger_jobs 与 _sweep_orphan_task_dirs 自带局部
    try/except，它们失败不会中断本轮其余作业。

    本函数曾因迁移丢失启动入口（全项目零引用、一次都没跑过），现由
    start_task_monitor() 在应用 lifespan 中拉起——见该函数 docstring。
    """
    while True:
        time.sleep(HEARTBEAT_INTERVAL)
        try:
            _run_monitor_tick(time.time())
        except Exception:
            _log_monitor_error()


_MONITOR_STARTED = False
_MONITOR_START_LOCK = threading.Lock()


def start_task_monitor() -> bool:
    """启动任务监控线程（幂等）。返回本次是否真正启动了线程。

    为什么需要显式启动：Flask 单体时代由模块级 `threading.Thread(target=
    _task_monitor_loop, daemon=True).start()` 拉起（基线 main.py:4594），
    8446166「移除 Flask」把实现搬进 services 包时**启动点丢失**，此后
    `_task_monitor_loop` 全项目零引用，监控循环一次都没跑过。后果：
      心跳超时不再暂停离线设备的批量任务；离线 >30 分钟 / 运行 >4 小时的任务
      不再自动取消；终态任务记录与 tasks/ 目录、downloads 缓存不再回收；
      `ledger_recover_orphans` 不再执行——预扣未退款的账永久悬挂在 open 状态。
    现改由应用 lifespan 调用（见 web.app._make_lifespan），恢复设计意图。

    幂等：多 worker 或测试多次构建应用时只启动一个线程。
    """
    global _MONITOR_STARTED
    with _MONITOR_START_LOCK:
        if _MONITOR_STARTED:
            return False
        threading.Thread(target=_task_monitor_loop, name='task-monitor', daemon=True).start()
        _MONITOR_STARTED = True
        api_service.logger.info('任务监控线程已启动')
        return True


def _touch_device(device_id: str) -> None:
    """记录设备心跳时间（设备表容量上限——device_id 是客户端任意串，
    登录用户可提交海量随机 ID 使字典无界增长，超限清最旧；
    入口已统一钳 64，此处兜底再钳一次防新调用点绕过）"""
    device_id = str(device_id or '')[:64]
    if not device_id:
        return
    with DEVICE_LAST_SEEN_LOCK:
        if len(DEVICE_LAST_SEEN) >= 5000:
            # 保留最近心跳的设备：按最后心跳时间排序，砍掉最旧的一半
            for d in sorted(DEVICE_LAST_SEEN, key=lambda k: DEVICE_LAST_SEEN.get(k, 0.0))[:2500]:
                DEVICE_LAST_SEEN.pop(d, None)
        DEVICE_LAST_SEEN[device_id] = time.time()