128 lines
4.2 KiB
Python
128 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""verify_projects.py — multi-project integrity verifier (Phase D1).
|
|
|
|
Idempotent, safe to run anytime. For every project under PROJECT_BASE it:
|
|
- lists slug, DB path, project id, chapter count, speaker count, voice count
|
|
- asserts each project.db has exactly ONE project row
|
|
- flags any cross-project references (utters/chapters pointing at a DB file
|
|
outside their own project)
|
|
|
|
Exit code 0 = all checks passed; 1 = a project failed a check.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from umbraaudio.config import PROJECT_BASE, list_projects
|
|
|
|
|
|
def _q(path: Path, sql: str, params: tuple = ()) -> list[tuple]:
|
|
conn = sqlite3.connect(str(path))
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
return [tuple(r) for r in conn.execute(sql, params).fetchall()]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def verify_project(project: dict) -> list[str]:
|
|
slug = str(project["slug"])
|
|
db_path = PROJECT_BASE / slug / ".umbra" / "project.db"
|
|
errors: list[str] = []
|
|
|
|
if not db_path.exists():
|
|
return [f"{slug}: missing project.db at {db_path}"]
|
|
|
|
try:
|
|
conn = sqlite3.connect(str(db_path))
|
|
conn.row_factory = sqlite3.Row
|
|
except sqlite3.Error as e:
|
|
return [f"{slug}: cannot open project.db: {e}"]
|
|
|
|
try:
|
|
project_rows = conn.execute("SELECT id, name FROM project").fetchall()
|
|
if len(project_rows) != 1:
|
|
errors.append(f"{slug}: expected 1 project row, found {len(project_rows)}")
|
|
|
|
chapters = conn.execute("SELECT COUNT(*) AS c FROM chapters").fetchone()["c"]
|
|
speakers = conn.execute(
|
|
"SELECT COUNT(DISTINCT speaker_name) AS c FROM utterances"
|
|
).fetchone()["c"]
|
|
voices = conn.execute("SELECT COUNT(*) AS c FROM utterances").fetchone()["c"]
|
|
except sqlite3.Error as e:
|
|
errors.append(f"{slug}: DB query failed: {e}")
|
|
conn.close()
|
|
return errors
|
|
conn.close()
|
|
|
|
print(
|
|
f" {slug:<40} id={project_rows[0][1] if project_rows else '?':<8} "
|
|
f"db={db_path.name:<12} chapters={chapters:<4} speakers={speakers:<4} "
|
|
f"voice_utterances={voices}"
|
|
)
|
|
if len(project_rows) != 1:
|
|
errors.append(f"{slug}: exactly one project row required, got {len(project_rows)}")
|
|
return errors
|
|
|
|
|
|
def check_cross_project_references() -> list[str]:
|
|
"""Flag projects whose DB rows reference an out-of-project data path."""
|
|
errors: list[str] = []
|
|
projects = list_projects()
|
|
if len(projects) < 2:
|
|
return errors # nothing to cross-reference with a single project
|
|
|
|
slugmap = {str(p["slug"]): str(p["slug"]) for p in projects}
|
|
for p in projects:
|
|
slug = str(p["slug"])
|
|
db_path = PROJECT_BASE / slug / ".umbra" / "project.db"
|
|
if not db_path.exists():
|
|
continue
|
|
try:
|
|
for (src,) in _q(db_path, "SELECT source_path FROM project"):
|
|
if not src:
|
|
continue
|
|
s = str(src)
|
|
# A source_path pointing inside ANOTHER project dir is a leak.
|
|
for other in slugmap:
|
|
if other != slug and (f"/{other}/" in s or s.startswith(other + "/")):
|
|
errors.append(
|
|
f"{slug}: source_path references another project: {s}"
|
|
)
|
|
except sqlite3.Error:
|
|
pass
|
|
return errors
|
|
|
|
|
|
def main() -> int:
|
|
projects = list_projects()
|
|
if not projects:
|
|
print("No projects found under", PROJECT_BASE)
|
|
return 0
|
|
|
|
print(f"Verifying {len(projects)} project(s) under {PROJECT_BASE}\n")
|
|
errors: list[str] = []
|
|
for project in projects:
|
|
errors.extend(verify_project(project))
|
|
|
|
print("\nCross-project reference check:")
|
|
cross = check_cross_project_references()
|
|
errors.extend(cross)
|
|
print(" " + (", ".join(cross) if cross else "OK — no cross-project references"))
|
|
|
|
print(f"\n{'FAILED' if errors else 'PASSED'} "
|
|
f"({len(errors)} issue(s))")
|
|
if errors:
|
|
for e in errors:
|
|
print(" [x]", e)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|