Spaces:
Paused
Paused
File size: 5,070 Bytes
37c8161 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | """
任务历史持久化：按 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
|