app/services/user_files_sync.py

"""用户文件 Hub 同步后台任务。

定时扫描本地缓存，把未同步到 Hub 的文件推上去。
作为 asyncio.create_task(push_to_hub) 的兜底：如果异步推送失败或进程崩溃，
下次定时任务会重新扫描并补推。

仅在 Hub 启用时实际工作（is_hub_enabled()=True）。
"""
from __future__ import annotations

import asyncio
import logging
from pathlib import Path

from ..hf_storage import (
    _local_path,
    _local_root,
    is_hub_enabled,
    list_hub_keys,
    push_to_hub,
)

logger = logging.getLogger(__name__)

# 同步间隔（秒）
SYNC_INTERVAL_SEC = 60  # 1 分钟

# 需要定时扫描同步的 namespace 前缀
SYNC_NS_PREFIXES = ["user_accounts", "user_files/", "user_files_meta/"]

_sync_task: asyncio.Task | None = None


def _collect_local_files() -> list[tuple[str, str]]:
    """扫描本地缓存，返回 (namespace, key) 列表。

    namespace 从路径前缀推断：
    - user_accounts/<username>           → ns="user_accounts", key=<username>
    - user_files/<user_id>/<file_key>    → ns="user_files/<user_id>", key=<file_key>
    - user_files_meta/<user_id>/<key>    → ns="user_files_meta/<user_id>", key=<key>
    """
    root = _local_root()
    out: list[tuple[str, str]] = []
    if not root.exists():
        return out

    for ns_prefix in SYNC_NS_PREFIXES:
        parts = ns_prefix.rstrip("/").split("/")
        ns_dir = root.joinpath(*parts)
        if not ns_dir.exists() or not ns_dir.is_dir():
            continue
        for p in ns_dir.rglob("*"):
            if not p.is_file():
                continue
            try:
                rel = p.relative_to(ns_dir).as_posix()
                if not rel:
                    continue
                if ns_prefix == "user_accounts/":
                    ns = "user_accounts"
                else:
                    ns = ns_prefix.rstrip("/")
                out.append((ns, rel))
            except Exception:
                continue
    return out


async def _sync_once() -> int:
    """执行一轮同步：扫描本地 → 对比 Hub → 补推缺失文件。返回补推数量。"""
    if not is_hub_enabled():
        return 0

    local_files = await asyncio.to_thread(_collect_local_files)
    if not local_files:
        return 0

    # 按 namespace 分组，批量 list_hub_keys 减少调用
    ns_set: dict[str, list[str]] = {}
    for ns, key in local_files:
        ns_set.setdefault(ns, []).append(key)

    pushed = 0
    for ns, keys in ns_set.items():
        try:
            hub_keys = set(await asyncio.to_thread(list_hub_keys, ns))
        except Exception as e:
            logger.warning("[user_files_sync] list_hub_keys failed ns=%s: %s", ns, e)
            continue

        for key in keys:
            if key in hub_keys:
                continue
            # 本地有但 Hub 没有，补推
            try:
                ok = await push_to_hub(ns, key)
                if ok:
                    pushed += 1
                    logger.info("[user_files_sync] pushed ns=%s key=%s", ns, key)
            except Exception as e:
                logger.warning("[user_files_sync] push failed ns=%s key=%s: %s", ns, key, e)

    if pushed > 0:
        logger.info("[user_files_sync] sync round done, pushed %d files", pushed)
    return pushed


async def sync_loop():
    """定时同步循环。"""
    logger.info(
        "[user_files_sync] started, interval=%ds, hub_enabled=%s",
        SYNC_INTERVAL_SEC, is_hub_enabled(),
    )
    # 启动后先等一小段时间，让应用初始化完成
    await asyncio.sleep(10)
    while True:
        try:
            await _sync_once()
        except Exception as e:
            logger.warning("[user_files_sync] round failed: %s", e)
        try:
            await asyncio.sleep(SYNC_INTERVAL_SEC)
        except asyncio.CancelledError:
            break


def start_sync_task() -> asyncio.Task:
    """启动同步后台任务。"""
    global _sync_task
    if _sync_task is not None and not _sync_task.done():
        return _sync_task
    _sync_task = asyncio.create_task(sync_loop())
    return _sync_task


async def stop_sync_task() -> None:
    """停止同步后台任务。"""
    global _sync_task
    if _sync_task is None:
        return
    _sync_task.cancel()
    try:
        await _sync_task
    except asyncio.CancelledError:
        pass
    _sync_task = None