backend/app/services/external_download.py

"""外部下载（家用 worker）路由策略 + worker 心跳。

某些平台（默认 YouTube）在机房 IP 上会被风控，改由跑在住宅 IP 上的家用 worker
拉取下载。这里集中：是否该走外部下载、worker 是否在线。

配置存 app_config（key="external_downloader"）：{enabled: bool, platforms: [str]}。
默认开启、只对 youtube 生效——启动了 worker 就能用；没启动则入口会提示先开 worker。
"""
import time
from typing import Dict, Any

from app.db import app_config_dao

_KEY = "external_downloader"
_DEFAULT = {"enabled": True, "platforms": ["youtube"]}

# worker 最近一次轮询的时间戳（进程内存；HF 重启清零，worker 几秒一轮很快回填）
_last_seen = {"ts": 0.0}


def get_config() -> Dict[str, Any]:
    cfg = app_config_dao.get_value(_KEY) or {}
    return {
        "enabled": bool(cfg.get("enabled", _DEFAULT["enabled"])),
        "platforms": cfg.get("platforms") or _DEFAULT["platforms"],
    }


def set_config(enabled: bool, platforms) -> Dict[str, Any]:
    app_config_dao.set_value(_KEY, {"enabled": bool(enabled), "platforms": list(platforms or [])})
    return get_config()


def should_external(platform: str) -> bool:
    cfg = get_config()
    return cfg["enabled"] and platform in cfg["platforms"]


def mark_worker_seen() -> None:
    _last_seen["ts"] = time.time()


def worker_alive(within: float = 90.0) -> bool:
    return (time.time() - _last_seen["ts"]) < within