File size: 3,096 Bytes
e5d8d3a 033070f e5d8d3a 033070f e5d8d3a 033070f e5d8d3a | 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 | import json
import os
from pathlib import Path
from typing import Any, Dict, List, Optional
DEFAULT_DATA_DIR = Path(os.path.abspath("data/demo/public"))
def resolve_data_dir(dir_arg: Optional[str]) -> Path:
"""
Resolve the base data directory from CLI args or fall back to demo/public.
"""
if dir_arg:
return Path(dir_arg).expanduser().absolute()
return DEFAULT_DATA_DIR
def get_demo_dir(data_dir: Path, create: bool = False) -> Path:
"""Return the demo directory under the given data dir, optionally creating it."""
# data_dir 此时默认就是 data/demo/public 的绝对路径
demo_dir = data_dir
if create:
demo_dir.mkdir(parents=True, exist_ok=True)
return demo_dir
def list_demo_files(demo_dir: Path) -> List[Dict[str, str]]:
"""Return sorted demo metadata from a directory. Missing dirs result in empty list."""
if not demo_dir.exists():
return []
demo_list = []
for file_path in demo_dir.glob("*.json"):
demo_list.append(
{
"name": file_path.stem,
"file": file_path.name,
}
)
demo_list.sort(key=lambda item: item["name"])
return demo_list
def sanitize_demo_name(name: str) -> str:
"""Remove unsafe characters from a demo name to create a safe filename."""
unsafe_chars = ['/', '\\', ':', '*', '?', '"', '<', '>', '|']
safe_name = name or ""
for char in unsafe_chars:
safe_name = safe_name.replace(char, '_')
safe_name = safe_name.strip(' .')
return safe_name[:200]
def save_demo_payload(demo_dir: Path, name: str, data: Dict[str, Any], path: str = "", overwrite: bool = False) -> Dict[str, Any]:
"""
Persist an AnalyzeResponse payload as a demo JSON file.
Args:
demo_dir: demo目录的绝对路径
name: demo文件名（不含扩展名）
data: 要保存的数据
path: 保存路径，可以是 ""、"/" 或以 "/" 开头的路径，默认为根目录
overwrite: 是否覆盖已存在的文件，默认为False
"""
from backend.path_utils import resolve_demo_path
safe_name = sanitize_demo_name(name)
if not safe_name:
return {"success": False, "message": "文件名无效"}
# 解析目标路径
target_dir = resolve_demo_path(demo_dir, path)
if target_dir is None:
return {"success": False, "message": f"无效的保存路径: {path}"}
# 确保目标目录存在
target_dir.mkdir(parents=True, exist_ok=True)
file_path = target_dir / f"{safe_name}.json"
# 检查文件是否存在
if file_path.exists() and not overwrite:
return {
"success": False,
"exists": True,
"message": f'文件 "{safe_name}.json" 已存在',
"file": file_path.name,
}
with open(file_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
return {
"success": True,
"message": f'Demo "{name}" 保存成功',
"file": file_path.name,
}
|