""" 任务历史持久化:按 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