63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import typer
|
|
import yaml
|
|
|
|
from tht.cli.config_cmd import CONFIG_OPT
|
|
from tht.config import ConfigError, load_config
|
|
|
|
|
|
def _emit(payload: dict[str, Any], as_json: bool) -> None:
|
|
if as_json:
|
|
# A single serializer call keeps stdout valid for machine consumers.
|
|
typer.echo(json.dumps(payload, sort_keys=True))
|
|
return
|
|
for component, result in payload["components"].items():
|
|
detail = ""
|
|
if result["status"] == "error":
|
|
detail = f" - {result['message']}"
|
|
elif component == "data_root" and result["status"] == "warning":
|
|
detail = " - set THT_DATA_ROOT to enable portable storage"
|
|
elif component == "workspace_paths" and result["status"] == "warning":
|
|
detail = " - absolute legacy roots: " + ", ".join(result["legacy_absolute"])
|
|
typer.echo(f"{component}: {result['status']}{detail}")
|
|
|
|
|
|
def doctor(
|
|
config: Path = CONFIG_OPT,
|
|
as_json: bool = typer.Option(False, "--json", help="Emette diagnostica JSON."),
|
|
) -> None:
|
|
"""Validate portable storage configuration without contacting external services."""
|
|
data_root = os.environ.get("THT_DATA_ROOT")
|
|
components: dict[str, dict[str, Any]] = {
|
|
"config": {"status": "ok"},
|
|
"data_root": {"status": "ok" if data_root else "warning"},
|
|
}
|
|
try:
|
|
cfg = load_config(config)
|
|
except (ConfigError, yaml.YAMLError, OSError) as exc:
|
|
path_error = isinstance(exc, ConfigError) and "outside workspace" in str(exc)
|
|
target = "workspace_paths" if path_error else "config"
|
|
# Validation errors can contain Pydantic input excerpts, including credentials.
|
|
message = str(exc) if path_error else "configuration is invalid or unreadable"
|
|
components[target] = {"status": "error", "message": message}
|
|
payload = {"ok": False, "components": components}
|
|
_emit(payload, as_json)
|
|
raise typer.Exit(code=1)
|
|
|
|
legacy_absolute = [
|
|
name
|
|
for name in ("sessions", "artifacts", "indexes")
|
|
if getattr(cfg.roots, name).is_absolute()
|
|
]
|
|
components["workspace_paths"] = {
|
|
"status": "warning" if legacy_absolute else "ok",
|
|
"legacy_absolute": legacy_absolute,
|
|
}
|
|
_emit({"ok": True, "components": components}, as_json)
|