"""数据根、技能目录、兄弟技能根路径与用户数据子目录解析。""" from __future__ import annotations import os from jiangchang_skill_core.runtime_env import get_data_root, get_sibling_skills_root, get_user_id from util.constants import SKILL_SLUG _SCRIPTS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) STANDARD_SUBDIR_DOWNLOADS = "downloads" STANDARD_SUBDIR_IMPORTS = "imports" STANDARD_SUBDIR_EXPORTS = "exports" STANDARD_SUBDIR_UPLOADS = "uploads" STANDARD_SUBDIR_CACHE = "cache" STANDARD_SUBDIR_TEMP = "temp" STANDARD_SUBDIR_RPA_ARTIFACTS = "rpa-artifacts" STANDARD_SUBDIR_VIDEOS = "videos" CONFIG_KEY_DOWNLOAD_DIR = "SKILL_DOWNLOAD_DIR" CONFIG_KEY_IMPORT_DIR = "SKILL_IMPORT_DIR" CONFIG_KEY_EXPORT_DIR = "SKILL_EXPORT_DIR" CONFIG_KEY_UPLOAD_DIR = "SKILL_UPLOAD_DIR" CONFIG_KEY_CACHE_DIR = "SKILL_CACHE_DIR" CONFIG_KEY_TEMP_DIR = "SKILL_TEMP_DIR" def get_skill_root() -> str: return os.path.dirname(_SCRIPTS_DIR) def get_openclaw_root() -> str: return get_sibling_skills_root(_SCRIPTS_DIR) def get_skills_root() -> str: return get_sibling_skills_root(_SCRIPTS_DIR) def get_skill_data_dir() -> str: path = os.path.join(get_data_root(), get_user_id(), SKILL_SLUG) os.makedirs(path, exist_ok=True) return path def get_db_path(filename: str | None = None) -> str: name = filename or f"{SKILL_SLUG}.db" return os.path.join(get_skill_data_dir(), name) def _normalize_subdir(value: str) -> str: cleaned = value.strip().strip("/\\") if not cleaned or cleaned.startswith(".."): raise ValueError(f"invalid skill data subdir: {value!r}") return cleaned.replace("/", os.sep).replace("\\", os.sep) def resolve_data_path( config_key: str | None, default_subdir: str, *, create: bool = True, ) -> str: """解析技能拥有路径:空配置 → 数据目录下 default_subdir;相对 → 相对数据目录;绝对 → 原样。""" from jiangchang_skill_core import config base = get_skill_data_dir() raw = (config.get(config_key) or "").strip() if config_key else "" if not raw: path = os.path.join(base, _normalize_subdir(default_subdir)) elif os.path.isabs(raw): path = os.path.abspath(os.path.expanduser(raw)) else: path = os.path.join(base, _normalize_subdir(raw)) if create: os.makedirs(path, exist_ok=True) return path def resolve_input_path(raw_path: str, *, create_parent: bool = False) -> str: """解析 CLI 输入路径:绝对路径原样;相对路径相对 {skill_data_dir}/imports/。""" text = (raw_path or "").strip() if not text: return text expanded = os.path.expanduser(text) if os.path.isabs(expanded): return os.path.abspath(expanded) imports_dir = get_imports_dir(create=create_parent) return os.path.abspath(os.path.join(imports_dir, os.path.normpath(expanded))) def get_downloads_dir(*, create: bool = True) -> str: return resolve_data_path(CONFIG_KEY_DOWNLOAD_DIR, STANDARD_SUBDIR_DOWNLOADS, create=create) def get_imports_dir(*, create: bool = True) -> str: return resolve_data_path(CONFIG_KEY_IMPORT_DIR, STANDARD_SUBDIR_IMPORTS, create=create) def get_exports_dir(*, create: bool = True) -> str: return resolve_data_path(CONFIG_KEY_EXPORT_DIR, STANDARD_SUBDIR_EXPORTS, create=create) def get_uploads_dir(*, create: bool = True) -> str: return resolve_data_path(CONFIG_KEY_UPLOAD_DIR, STANDARD_SUBDIR_UPLOADS, create=create) def get_cache_dir(*, create: bool = True) -> str: return resolve_data_path(CONFIG_KEY_CACHE_DIR, STANDARD_SUBDIR_CACHE, create=create) def get_temp_dir(*, create: bool = True) -> str: return resolve_data_path(CONFIG_KEY_TEMP_DIR, STANDARD_SUBDIR_TEMP, create=create) def get_rpa_artifacts_dir(batch_id: str, *, create: bool = True) -> str: batch = (batch_id or "").strip() or "default" path = os.path.join(get_skill_data_dir(), STANDARD_SUBDIR_RPA_ARTIFACTS, batch) if create: os.makedirs(path, exist_ok=True) return path def get_videos_dir(*, create: bool = True) -> str: path = os.path.join(get_skill_data_dir(), STANDARD_SUBDIR_VIDEOS) if create: os.makedirs(path, exist_ok=True) return path def list_resolved_data_paths() -> dict[str, str]: """供 health / config-path 输出已解析的数据子目录(不创建目录)。""" return { "skill_data_dir": get_skill_data_dir(), "downloads_dir": get_downloads_dir(create=False), "imports_dir": get_imports_dir(create=False), "exports_dir": get_exports_dir(create=False), "uploads_dir": get_uploads_dir(create=False), "cache_dir": get_cache_dir(create=False), "temp_dir": get_temp_dir(create=False), "videos_dir": get_videos_dir(create=False), }