task_history.py

"""
任务历史持久化：按 task_id 记录 image_url、audio_url、result_video_url、status、created_at 等，媒体以 URL 保存。
"""
import json
from loguru import logger
import os
import threading
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional


# 默认历史文件路径：/mnt/workspace/task_history.json（创空间持久卷）。
# 本地开发等无 /mnt/workspace 的环境，请通过环境变量 TASK_HISTORY_PATH 覆盖，
# 例如 TASK_HISTORY_PATH=./task_history.json。
_DEFAULT_PATH = Path("/mnt/workspace/task_history.json")
_lock = threading.Lock()


def _path() -> Path:
    return Path(os.environ.get("TASK_HISTORY_PATH", str(_DEFAULT_PATH)))


def _load_raw() -> list[dict[str, Any]]:
    """加载原始列表（调用方需已持有 _lock）。"""
    p = _path()
    if not p.exists():
        return []
    try:
        with open(p, "r", encoding="utf-8") as f:
            data = json.load(f)
        return data.get("tasks", [])
    except (json.JSONDecodeError, OSError) as e:
        logger.warning("加载任务历史失败 path={} error={}", p, e)
        return []


def _save_raw(tasks: list[dict[str, Any]]) -> None:
    """写入列表（调用方需已持有 _lock）。"""
    p = _path()
    p.parent.mkdir(parents=True, exist_ok=True)
    with open(p, "w", encoding="utf-8") as f:
        json.dump({"tasks": tasks}, f, ensure_ascii=False, indent=2)


def load_history(owner_id: Optional[str] = None) -> list[dict[str, Any]]:
    """加载历史记录，返回按创建时间倒序的列表（最近在前）。

    :param owner_id:
        - None: 不过滤，返回全部记录（管理员视角，本地调试或脚本里使用）。
        - 非 None（含空字符串）: 仅返回 owner_id 严格相等的记录。
          老记录（无 owner_id 字段，等价于 None）会被规范成空字符串后再比对，
          所以匿名访客（无 router_id，owner_id=""）只会看到老记录或同样匿名的记录。
    """
    with _lock:
        tasks = _load_raw()
        tasks.sort(key=lambda t: t.get("created_at", ""), reverse=True)
        if owner_id is not None:
            tasks = [t for t in tasks if (t.get("owner_id") or "") == owner_id]
        return tasks


def add_task(
    task_id: str,
    image_url: str,
    audio_url: str,
    status: str = "PENDING",
    result_video_url: Optional[str] = None,
    message: Optional[str] = None,
    audio_duration: Optional[float] = None,
    dance_genres: Optional[str] = None,
    image_local_path: Optional[str] = None,
    audio_local_path: Optional[str] = None,
    owner_id: Optional[str] = None,
) -> None:
    """追加一条任务记录。

    :param owner_id: 用户身份标识（创空间下为 X-Modelscope-Router-Id 的值）。
        历史页按此字段过滤展示，确保用户只看到自己的任务。
    """
    created_at = datetime.now(timezone.utc).isoformat()
    entry = {
        "task_id": task_id,
        "image_url": image_url,
        "audio_url": audio_url,
        "result_video_url": result_video_url,
        "status": status,
        "created_at": created_at,
        "message": message,
        "audio_duration": audio_duration,
        "dance_genres": dance_genres,
        "image_local_path": image_local_path,
        "audio_local_path": audio_local_path,
        "video_local_path": None,
        "owner_id": owner_id or "",
    }
    with _lock:
        tasks = _load_raw()
        tasks = [t for t in tasks if t.get("task_id") != task_id]
        tasks.insert(0, entry)
        _save_raw(tasks)
    logger.debug("历史已添加 task_id={}", task_id)


def update_task(
    task_id: str,
    *,
    result_video_url: Optional[str] = None,
    status: Optional[str] = None,
    message: Optional[str] = None,
    video_local_path: Optional[str] = None,
) -> None:
    """按 task_id 更新任务记录（只更新传入的字段）。"""
    with _lock:
        tasks = _load_raw()
        for t in tasks:
            if t.get("task_id") == task_id:
                if result_video_url is not None:
                    t["result_video_url"] = result_video_url
                if status is not None:
                    t["status"] = status
                if message is not None:
                    t["message"] = message
                if video_local_path is not None:
                    t["video_local_path"] = video_local_path
                _save_raw(tasks)
                logger.debug("历史已更新 task_id={} status={}", task_id, status)
                return
    logger.warning("更新历史时未找到 task_id={}", task_id)


def get_task(task_id: str) -> Optional[dict[str, Any]]:
    """按 task_id 返回一条任务记录，不存在则返回 None。"""
    tasks = load_history()
    for t in tasks:
        if t.get("task_id") == task_id:
            return t
    return None