132 lines
4.4 KiB
Python
132 lines
4.4 KiB
Python
|
"""Rules for required project artifacts (DB + repo hybrid)."""
|
|||
|
|
import os
|
|||
|
|
from typing import Dict, List
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# DB-backed artifacts (5 mandatory: worklog, todo, project_state,
|
|||
|
|
# current_status, next_steps) – checked via db_state_store
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
DB_ARTIFACTS: List[str] = [
|
|||
|
|
"worklog", # orch_worklog (mindestens 1 Eintrag)
|
|||
|
|
"todo", # orch_todos (mindestens 1 Eintrag)
|
|||
|
|
"project_state", # orch_kv mit key='project_state' ODER 'phase'
|
|||
|
|
"current_status", # orch_status (gesetzt)
|
|||
|
|
"next_steps", # orch_next_steps (mindestens 1 pending)
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
# Repo-backed artifacts (15 specs/docs/deploy files) – checked as file paths
|
|||
|
|
# unter project_root
|
|||
|
|
# ---------------------------------------------------------------------------
|
|||
|
|
|
|||
|
|
REPO_ARTIFACTS: List[str] = [
|
|||
|
|
"specs/current/requirements.md",
|
|||
|
|
"specs/current/design.md",
|
|||
|
|
"specs/current/tasks.md",
|
|||
|
|
"docs/architecture.md",
|
|||
|
|
"docs/test_report.md",
|
|||
|
|
"docs/runtime_report.md",
|
|||
|
|
"deploy/env.md",
|
|||
|
|
"deploy/healthcheck.md",
|
|||
|
|
"deploy/runbook.md",
|
|||
|
|
"deploy/rollback.md",
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
# All artifacts (kept for backward compat with the original list)
|
|||
|
|
CORE_ARTIFACTS = [f".a0/{a}.md" for a in DB_ARTIFACTS if a != "project_state"] + [
|
|||
|
|
".a0/project_state.json"
|
|||
|
|
] + REPO_ARTIFACTS
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _check_db_artifacts(project_name: str) -> Dict[str, bool]:
|
|||
|
|
"""Prüft die 5 DB-backed Artefakte via db_state_store."""
|
|||
|
|
from .db_state_store import (
|
|||
|
|
get_project_id,
|
|||
|
|
get_kv,
|
|||
|
|
get_status,
|
|||
|
|
list_worklog,
|
|||
|
|
list_todos,
|
|||
|
|
list_next_steps,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
pid = get_project_id(project_name)
|
|||
|
|
if pid is None:
|
|||
|
|
return {
|
|||
|
|
"worklog": False, "todo": False, "project_state": False,
|
|||
|
|
"current_status": False, "next_steps": False,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return {
|
|||
|
|
"worklog": len(list_worklog(pid, limit=1)) > 0,
|
|||
|
|
"todo": len(list_todos(pid, status="open", limit=1)) > 0,
|
|||
|
|
"project_state": (
|
|||
|
|
get_kv(pid, "project_state") is not None
|
|||
|
|
or get_kv(pid, "phase") is not None
|
|||
|
|
),
|
|||
|
|
"current_status": get_status(pid) is not None,
|
|||
|
|
"next_steps": len(list_next_steps(pid, status="pending", limit=1)) > 0,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _check_repo_artifacts(project_root: str) -> Dict[str, bool]:
|
|||
|
|
"""Prüft die 15 Repo-Artefakte (specs/, docs/, deploy/) per File-Existenz."""
|
|||
|
|
result: Dict[str, bool] = {}
|
|||
|
|
for rel_path in REPO_ARTIFACTS:
|
|||
|
|
result[rel_path] = os.path.exists(os.path.join(project_root, rel_path))
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
|
|||
|
|
def check_required_artifacts(
|
|||
|
|
project_name: str = "",
|
|||
|
|
project_root: str = "",
|
|||
|
|
) -> dict:
|
|||
|
|
"""Prüft alle Pflicht-Artefakte (DB + Repo).
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
project_name: Projektname (für DB-Checks). Default: aus project_root
|
|||
|
|
abgeleitet (basename).
|
|||
|
|
project_root: Projekt-Wurzelverzeichnis (für Repo-Checks). Default: cwd.
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
{
|
|||
|
|
"db": {artifact_name: bool, …},
|
|||
|
|
"repo": {relative_path: bool, …},
|
|||
|
|
"present": [alle existierenden], # kombiniert
|
|||
|
|
"missing": [alle fehlenden], # kombiniert
|
|||
|
|
"total": int,
|
|||
|
|
}
|
|||
|
|
"""
|
|||
|
|
if not project_name:
|
|||
|
|
project_name = os.path.basename(project_root or os.getcwd())
|
|||
|
|
if not project_root:
|
|||
|
|
project_root = os.getcwd()
|
|||
|
|
|
|||
|
|
db = _check_db_artifacts(project_name)
|
|||
|
|
repo = _check_repo_artifacts(project_root)
|
|||
|
|
|
|||
|
|
present = [a for a, ok in db.items() if ok] + [a for a, ok in repo.items() if ok]
|
|||
|
|
missing = [a for a, ok in db.items() if not ok] + [a for a, ok in repo.items() if not ok]
|
|||
|
|
|
|||
|
|
return {
|
|||
|
|
"db": db,
|
|||
|
|
"repo": repo,
|
|||
|
|
"present": present,
|
|||
|
|
"missing": missing,
|
|||
|
|
"total": len(db) + len(repo),
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
# Backward-compat: altes Format (nur Repo-Files) für externe Caller
|
|||
|
|
def check_legacy_artifacts(project_root: str) -> dict:
|
|||
|
|
"""Alte CORE_ARTIFACTS-Liste als File-Existenz-Check (nur Repo)."""
|
|||
|
|
missing = []
|
|||
|
|
present = []
|
|||
|
|
for artifact in CORE_ARTIFACTS:
|
|||
|
|
path = os.path.join(project_root, artifact)
|
|||
|
|
if os.path.exists(path):
|
|||
|
|
present.append(artifact)
|
|||
|
|
else:
|
|||
|
|
missing.append(artifact)
|
|||
|
|
return {"present": present, "missing": missing, "total": len(CORE_ARTIFACTS)}
|