156 lines
5.2 KiB
Python
156 lines
5.2 KiB
Python
"""数据根、技能目录、兄弟技能根路径与用户数据子目录解析。"""
|
||
|
||
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_INPUT = "input"
|
||
STANDARD_SUBDIR_OUTPUT = "output"
|
||
|
||
# 宿主扫描的 HTML 报告页（数据管理侧栏 htmlPages，非 input/output 文件浏览器）
|
||
STANDARD_SUBDIR_HTML = "html"
|
||
|
||
# 内部/可选目录（不上侧栏）
|
||
STANDARD_SUBDIR_UPLOADS = "uploads"
|
||
STANDARD_SUBDIR_CACHE = "cache"
|
||
STANDARD_SUBDIR_TEMP = "temp"
|
||
STANDARD_SUBDIR_RPA_ARTIFACTS = "rpa-artifacts"
|
||
STANDARD_SUBDIR_VIDEOS = "videos"
|
||
|
||
CONFIG_KEY_INPUT_DIR = "SKILL_INPUT_DIR"
|
||
CONFIG_KEY_OUTPUT_DIR = "SKILL_OUTPUT_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}/input/。"""
|
||
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)
|
||
input_dir = get_input_dir(create=create_parent)
|
||
return os.path.abspath(os.path.join(input_dir, os.path.normpath(expanded)))
|
||
|
||
|
||
def get_input_dir(*, create: bool = True) -> str:
|
||
"""上行：用户/Agent 放入的待处理文件（清单、CSV、产品表等）。"""
|
||
return resolve_data_path(CONFIG_KEY_INPUT_DIR, STANDARD_SUBDIR_INPUT, create=create)
|
||
|
||
|
||
def get_output_dir(*, create: bool = True) -> str:
|
||
"""下行：技能/宿主交给用户拿走的文件（视频、报表、导出 CSV 等）。"""
|
||
return resolve_data_path(CONFIG_KEY_OUTPUT_DIR, STANDARD_SUBDIR_OUTPUT, 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 get_html_dir(*, create: bool = True) -> str:
|
||
"""HTML 报告页目录：宿主扫描 `{skill_data}/html/*.html`（一层、非递归）。"""
|
||
path = os.path.join(get_skill_data_dir(), STANDARD_SUBDIR_HTML)
|
||
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(),
|
||
"input_dir": get_input_dir(create=False),
|
||
"output_dir": get_output_dir(create=False),
|
||
"html_dir": get_html_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),
|
||
}
|