240 lines
7.3 KiB
Python
240 lines
7.3 KiB
Python
"""宿主机模式工作区配置管理。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Tuple, Union
|
|
|
|
from config import DEFAULT_PROJECT_PATH, HOST_WORKSPACES_FILE
|
|
|
|
_WORKSPACE_ID_RE = re.compile(r"^[a-zA-Z0-9._-]{1,64}$")
|
|
_REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _resolve_config_path(config_path: Optional[Union[str, Path]] = None) -> Path:
|
|
raw = str(config_path or HOST_WORKSPACES_FILE or "").strip()
|
|
if not raw:
|
|
raw = "./config/host_workspaces.json"
|
|
path = Path(raw).expanduser()
|
|
if not path.is_absolute():
|
|
path = (_REPO_ROOT / path).resolve()
|
|
return path
|
|
|
|
|
|
def _default_workspace_path() -> Path:
|
|
p = Path(DEFAULT_PROJECT_PATH).expanduser()
|
|
if not p.is_absolute():
|
|
p = (_REPO_ROOT / p).resolve()
|
|
return p
|
|
|
|
|
|
def _default_payload() -> Dict[str, Any]:
|
|
return {
|
|
"default_workspace_id": "default",
|
|
"workspaces": [
|
|
{
|
|
"workspace_id": "default",
|
|
"label": "默认工作区",
|
|
"path": str(_default_workspace_path()),
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def _ensure_config_file(path: Path) -> Dict[str, Any]:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
if path.exists():
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
if isinstance(data, dict):
|
|
return data
|
|
except Exception:
|
|
pass
|
|
data = _default_payload()
|
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return data
|
|
|
|
|
|
def _normalize_workspace_id(raw: Any, index: int) -> str:
|
|
value = str(raw or "").strip()
|
|
if _WORKSPACE_ID_RE.match(value):
|
|
return value
|
|
return f"workspace_{index + 1}"
|
|
|
|
|
|
def _slugify_workspace_id(raw: Any) -> str:
|
|
value = str(raw or "").strip().lower()
|
|
value = re.sub(r"[^a-z0-9._-]+", "-", value)
|
|
value = re.sub(r"-{2,}", "-", value).strip("-._")
|
|
if not value:
|
|
return "workspace"
|
|
if _WORKSPACE_ID_RE.match(value):
|
|
return value
|
|
return "workspace"
|
|
|
|
|
|
def _normalize_workspace_path(raw: Any) -> Path:
|
|
value = str(raw or "").strip()
|
|
if not value:
|
|
return _default_workspace_path()
|
|
path = Path(value).expanduser()
|
|
if not path.is_absolute():
|
|
path = (_REPO_ROOT / path).resolve()
|
|
else:
|
|
path = path.resolve()
|
|
return path
|
|
|
|
|
|
def _normalize_entry(raw: Any, index: int) -> Optional[Dict[str, str]]:
|
|
if not isinstance(raw, dict):
|
|
return None
|
|
workspace_id = _normalize_workspace_id(
|
|
raw.get("workspace_id") or raw.get("id") or raw.get("name"), index
|
|
)
|
|
label = str(raw.get("label") or raw.get("name") or workspace_id).strip() or workspace_id
|
|
path = _normalize_workspace_path(raw.get("path"))
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return {
|
|
"workspace_id": workspace_id,
|
|
"label": label,
|
|
"path": str(path),
|
|
}
|
|
|
|
|
|
def load_host_workspace_catalog(
|
|
config_path: Optional[Union[str, Path]] = None,
|
|
) -> Dict[str, Any]:
|
|
cfg_path = _resolve_config_path(config_path)
|
|
payload = _ensure_config_file(cfg_path)
|
|
raw_workspaces = payload.get("workspaces")
|
|
if not isinstance(raw_workspaces, list):
|
|
raw_workspaces = []
|
|
|
|
seen_ids = set()
|
|
workspaces: List[Dict[str, str]] = []
|
|
for idx, item in enumerate(raw_workspaces):
|
|
normalized = _normalize_entry(item, idx)
|
|
if not normalized:
|
|
continue
|
|
ws_id = normalized["workspace_id"]
|
|
if ws_id in seen_ids:
|
|
continue
|
|
seen_ids.add(ws_id)
|
|
workspaces.append(normalized)
|
|
|
|
if not workspaces:
|
|
fallback = _default_payload()["workspaces"][0]
|
|
path = _normalize_workspace_path(fallback.get("path"))
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
workspaces = [
|
|
{
|
|
"workspace_id": "default",
|
|
"label": "默认工作区",
|
|
"path": str(path),
|
|
}
|
|
]
|
|
seen_ids.add("default")
|
|
|
|
default_workspace_id = str(payload.get("default_workspace_id") or "").strip()
|
|
if default_workspace_id not in seen_ids:
|
|
default_workspace_id = workspaces[0]["workspace_id"]
|
|
|
|
return {
|
|
"source_path": str(cfg_path),
|
|
"default_workspace_id": default_workspace_id,
|
|
"workspaces": workspaces,
|
|
}
|
|
|
|
|
|
def resolve_host_workspace(
|
|
selected_workspace_id: Optional[str] = None,
|
|
config_path: Optional[Union[str, Path]] = None,
|
|
) -> Tuple[Dict[str, Any], Dict[str, str]]:
|
|
catalog = load_host_workspace_catalog(config_path=config_path)
|
|
candidates = catalog.get("workspaces") or []
|
|
selected_id = str(selected_workspace_id or "").strip()
|
|
current = None
|
|
if selected_id:
|
|
current = next((ws for ws in candidates if ws.get("workspace_id") == selected_id), None)
|
|
if not current:
|
|
default_id = catalog.get("default_workspace_id")
|
|
current = next((ws for ws in candidates if ws.get("workspace_id") == default_id), None)
|
|
if not current:
|
|
current = candidates[0]
|
|
return catalog, current
|
|
|
|
|
|
def create_host_workspace(
|
|
path: str,
|
|
label: Optional[str] = None,
|
|
*,
|
|
set_default: bool = False,
|
|
config_path: Optional[Union[str, Path]] = None,
|
|
) -> Dict[str, Any]:
|
|
cfg_path = _resolve_config_path(config_path)
|
|
payload = _ensure_config_file(cfg_path)
|
|
|
|
raw_workspaces = payload.get("workspaces")
|
|
if not isinstance(raw_workspaces, list):
|
|
raw_workspaces = []
|
|
|
|
normalized_path = _normalize_workspace_path(path)
|
|
normalized_path.mkdir(parents=True, exist_ok=True)
|
|
target_path_str = str(normalized_path)
|
|
|
|
for idx, item in enumerate(raw_workspaces):
|
|
existing = _normalize_entry(item, idx)
|
|
if not existing:
|
|
continue
|
|
if existing.get("path") == target_path_str:
|
|
# 已存在相同路径则直接返回
|
|
return {
|
|
"created": False,
|
|
"workspace": existing,
|
|
"catalog": load_host_workspace_catalog(config_path=cfg_path),
|
|
}
|
|
|
|
clean_label = str(label or "").strip()
|
|
base_id_seed = clean_label or normalized_path.name or "workspace"
|
|
base_id = _slugify_workspace_id(base_id_seed)
|
|
existing_ids = {
|
|
_normalize_workspace_id(item.get("workspace_id") or item.get("id"), i)
|
|
for i, item in enumerate(raw_workspaces)
|
|
if isinstance(item, dict)
|
|
}
|
|
workspace_id = base_id
|
|
suffix = 2
|
|
while workspace_id in existing_ids:
|
|
workspace_id = f"{base_id}-{suffix}"
|
|
suffix += 1
|
|
|
|
workspace = {
|
|
"workspace_id": workspace_id,
|
|
"label": clean_label or workspace_id,
|
|
"path": target_path_str,
|
|
}
|
|
raw_workspaces.append(workspace)
|
|
payload["workspaces"] = raw_workspaces
|
|
|
|
default_id = str(payload.get("default_workspace_id") or "").strip()
|
|
if set_default or not default_id:
|
|
payload["default_workspace_id"] = workspace_id
|
|
|
|
cfg_path.parent.mkdir(parents=True, exist_ok=True)
|
|
cfg_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
return {
|
|
"created": True,
|
|
"workspace": workspace,
|
|
"catalog": load_host_workspace_catalog(config_path=cfg_path),
|
|
}
|
|
|
|
|
|
__all__ = [
|
|
"load_host_workspace_catalog",
|
|
"resolve_host_workspace",
|
|
"create_host_workspace",
|
|
]
|