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, }