#!/usr/bin/env python3 """aria-schema-stats — Schema-Coverage-Report für den Aria-Brain-Vault. Liest den Vault (Standard: /root/aria/brain) und berichtet: - Typed vs. untyped Notes - Per-type Counts mit Vergleich zu erlaubten Types aus schema-pack.yaml - Orphan-Notes (weder eingehende noch ausgehende Wikilinks) - Dangling Wikilinks (Ziel existiert nicht im Vault) - Coverage-Thresholds aus schema-pack.yaml Usage: aria-schema-stats.py aria-schema-stats.py --vault /path/to/brain aria-schema-stats.py --json aria-schema-stats.py --json --vault /path/to/brain Exit-Codes: 0 = pass (alle thresholds OK), 1 = warnings/errors über threshold """ import argparse import json import os import re import sys from pathlib import Path from typing import Any # ─── Defaults ───────────────────────────────────────────────────────────────── VAULT_DEFAULT = Path("/root/aria/brain") SCHEMA_PACK_DEFAULT = Path("/root/aria/brain/schema-pack.yaml") WIKILINK_RE = re.compile(r"!?\[\[([^\]]+?)\]\]") ASSET_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp", ".pdf", ".canvas", ".excalidraw", ".mp4", ".mov", ".mp3"} # Fallback thresholds wenn schema-pack.yaml nicht parsebar DEFAULT_THRESHOLDS = { "typed_ratio_warn": 0.70, "typed_ratio_error": 0.50, "orphan_ratio_warn": 0.20, "dangling_warn": 50, } # Erlaubte Types aus CLAUDE.md (Fallback wenn kein Pack geladen) CLAUDE_TYPE_ENUM = { "project", "reference", "research", "daily", "identity", "system", "learnings", "template", "inbox", "audit", "schema", } # ─── Schema Pack laden ──────────────────────────────────────────────────────── def load_schema_pack(path: Path) -> dict[str, Any]: """Laedt schema-pack.yaml; gibt leeres dict bei Fehler zurueck.""" if not path.exists(): return {} try: import yaml # type: ignore data = yaml.safe_load(path.read_text(encoding="utf-8")) return data if isinstance(data, dict) else {} except Exception: return {} def get_allowed_types(pack: dict) -> set[str]: if pack and "page_types" in pack: return set(pack["page_types"].keys()) return CLAUDE_TYPE_ENUM def get_thresholds(pack: dict) -> dict: if pack and "coverage_thresholds" in pack: t = pack["coverage_thresholds"] return { "typed_ratio_warn": float(t.get("typed_ratio_warn", DEFAULT_THRESHOLDS["typed_ratio_warn"])), "typed_ratio_error": float(t.get("typed_ratio_error", DEFAULT_THRESHOLDS["typed_ratio_error"])), "orphan_ratio_warn": float(t.get("orphan_ratio_warn", DEFAULT_THRESHOLDS["orphan_ratio_warn"])), "dangling_warn": int(t.get("dangling_warn", DEFAULT_THRESHOLDS["dangling_warn"])), } return DEFAULT_THRESHOLDS # ─── Frontmatter Parser ─────────────────────────────────────────────────────── def parse_frontmatter(text: str) -> dict | None: """Extrahiert YAML-Frontmatter; None wenn nicht vorhanden oder Parse-Error.""" parts = text.split("---", 2) if len(parts) < 3 or parts[0].strip(): return None fm_raw = parts[1] # Obsidian-wikilinks in frontmatter quoten fm_raw = re.sub( r"^(\s*(?:related|tags|crossref|references|superseded_by|supersedes|" r"derived_from|source_for|part_of_project|has_part|implements|challenges):\s*)(\[\[.+?)$", lambda m: f'{m.group(1)}"{m.group(2).replace(chr(34), chr(39))}"', fm_raw, flags=re.MULTILINE, ) try: import yaml # type: ignore result = yaml.safe_load(fm_raw) return result if isinstance(result, dict) else None except Exception: return None # ─── Vault Index ───────────────────────────────────────────────────────────── def build_vault_index(root: Path) -> tuple[dict[str, list[str]], set[str]]: """stem(lower) -> [relpaths], by_relpath (ohne .md, lower).""" by_stem: dict[str, list[str]] = {} by_relpath: set[str] = set() for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [d for d in dirnames if not d.startswith(".")] for fn in filenames: if not fn.endswith(".md"): continue full = Path(dirpath) / fn rel = full.relative_to(root).as_posix() stem = fn[:-3] by_stem.setdefault(stem.lower(), []).append(rel) by_relpath.add(rel[:-3].lower()) return by_stem, by_relpath def normalize_wikilink_target(raw: str) -> str: t = raw.strip().split("#", 1)[0].split("|", 1)[0].strip() return t def resolve_wikilink(target: str, by_stem: dict, by_relpath: set) -> bool: t = target if t.endswith(".md"): t = t[:-3] key = t.lower() if "/" in t: return key in by_relpath return key in by_stem # ─── Scan ──────────────────────────────────────────────────────────────────── def scan_vault(root: Path, allowed_types: set[str]) -> dict[str, Any]: by_stem, by_relpath = build_vault_index(root) total = 0 no_frontmatter: list[str] = [] no_type: list[str] = [] invalid_type: dict[str, str] = {} # rel -> actual type type_counts: dict[str, int] = {} folder_type_counts: dict[str, dict[str, int]] = {} # folder -> type -> count # Link tracking outbound_links: dict[str, set[str]] = {} # rel -> {targets} inbound_links: dict[str, set[str]] = {} # stem_lower -> {files linking to it} dangling: list[tuple[str, str]] = [] # [(source_rel, target)] for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [d for d in dirnames if not d.startswith(".")] for fn in filenames: if not fn.endswith(".md"): continue total += 1 full = Path(dirpath) / fn rel = full.relative_to(root).as_posix() folder = rel.rsplit("/", 1)[0] if "/" in rel else "root" try: text = full.read_text(encoding="utf-8", errors="replace") except Exception: continue # Frontmatter fm = parse_frontmatter(text) if fm is None: no_frontmatter.append(rel) elif "type" not in fm or fm.get("type") in (None, ""): no_type.append(rel) else: note_type = str(fm["type"]) if note_type not in allowed_types: invalid_type[rel] = note_type else: type_counts[note_type] = type_counts.get(note_type, 0) + 1 # Für stats: trotzdem zählen (auch non-standard) if note_type in allowed_types: folder_type_counts.setdefault(folder, {}) folder_type_counts[folder][note_type] = folder_type_counts[folder].get(note_type, 0) + 1 # Wikilinks seen: set[str] = set() for m in WIKILINK_RE.finditer(text): is_embed = m.group(0).startswith("!") target = normalize_wikilink_target(m.group(1)) if not target or target in seen: continue ext = os.path.splitext(target)[1].lower() if is_embed or ext in ASSET_EXTS: continue seen.add(target) # Inbound tracking t_lower = target.lower().split("#")[0].split("|")[0].strip() if "/" not in t_lower: inbound_links.setdefault(t_lower, set()).add(rel) else: inbound_links.setdefault(t_lower, set()).add(rel) # Dangling check if not resolve_wikilink(target, by_stem, by_relpath): dangling.append((rel, target)) outbound_links[rel] = seen # Orphans: kein eingehender UND kein ausgehender Link orphans: list[str] = [] for rel, outbound in outbound_links.items(): stem = rel.rsplit("/", 1)[-1][:-3].lower() rel_no_ext = rel[:-3].lower() has_inbound = (stem in inbound_links and inbound_links[stem]) or \ (rel_no_ext in inbound_links and inbound_links[rel_no_ext]) has_outbound = bool(outbound) if not has_inbound and not has_outbound: orphans.append(rel) typed_valid = sum(type_counts.values()) typed_all = typed_valid + len(invalid_type) # hat type-feld, auch wenn non-standard return { "total_notes": total, "no_frontmatter": len(no_frontmatter), "no_frontmatter_list": sorted(no_frontmatter)[:20], # limit für output "no_type": len(no_type), "no_type_list": sorted(no_type)[:20], "invalid_type_count": len(invalid_type), "invalid_types": dict(sorted(invalid_type.items())[:20]), "typed_valid_count": typed_valid, "type_counts": dict(sorted(type_counts.items(), key=lambda x: -x[1])), "folder_type_counts": folder_type_counts, "orphan_count": len(orphans), "orphan_list": sorted(orphans)[:20], "dangling_count": len(dangling), "dangling_list": [(s, t) for s, t in dangling[:20]], "allowed_types": sorted(allowed_types), } # ─── Report ─────────────────────────────────────────────────────────────────── def evaluate_thresholds(stats: dict, thresholds: dict) -> list[str]: issues: list[str] = [] total = stats["total_notes"] if total == 0: return issues typed_ratio = stats["typed_valid_count"] / total orphan_ratio = stats["orphan_count"] / total if typed_ratio < thresholds["typed_ratio_error"]: issues.append( f"ERROR: Typed-Ratio {typed_ratio:.0%} unter Error-Schwelle " f"{thresholds['typed_ratio_error']:.0%}" ) elif typed_ratio < thresholds["typed_ratio_warn"]: issues.append( f"WARN: Typed-Ratio {typed_ratio:.0%} unter Warn-Schwelle " f"{thresholds['typed_ratio_warn']:.0%}" ) if orphan_ratio > thresholds["orphan_ratio_warn"]: issues.append( f"WARN: Orphan-Ratio {orphan_ratio:.0%} über Warn-Schwelle " f"{thresholds['orphan_ratio_warn']:.0%}" ) if stats["dangling_count"] > thresholds["dangling_warn"]: issues.append( f"WARN: {stats['dangling_count']} dangling Wikilinks über Schwelle " f"{thresholds['dangling_warn']}" ) return issues def print_human_report(stats: dict, thresholds: dict, vault_path: Path, pack_id: str) -> None: total = stats["total_notes"] typed = stats["typed_valid_count"] untyped = stats["no_frontmatter"] + stats["no_type"] + stats["invalid_type_count"] typed_ratio = typed / total if total else 0 orphan_ratio = stats["orphan_count"] / total if total else 0 print("=" * 60) print(f"aria-schema-stats — Pack: {pack_id}") print(f"Vault: {vault_path}") print("=" * 60) print(f"\n── Coverage ────────────────────────────────────────────────") print(f" Gesamt Notes : {total:>5}") print(f" Typed (valide) : {typed:>5} ({typed_ratio:.0%})") print(f" Kein Frontmatter : {stats['no_frontmatter']:>5}") print(f" Frontmatter, kein type: {stats['no_type']:>5}") print(f" Non-Standard type : {stats['invalid_type_count']:>5}") print(f"\n── Types (valide, nach Häufigkeit) ─────────────────────────") for t, c in stats["type_counts"].items(): bar = "█" * (c * 30 // max(stats["type_counts"].values(), default=1)) print(f" {t:<20} {c:>4} {bar}") if stats["invalid_types"]: print(f"\n── Non-Standard Types (nicht in schema-pack) ───────────────") # Aggregate non-standard types ns_counts: dict[str, int] = {} for rel, t in stats["invalid_types"].items(): ns_counts[t] = ns_counts.get(t, 0) + 1 for t, c in sorted(ns_counts.items(), key=lambda x: -x[1]): print(f" {t:<20} {c:>4} (→ migrieren oder Pack erweitern)") print(f"\n── Links ────────────────────────────────────────────────────") print(f" Orphan Notes : {stats['orphan_count']:>5} ({orphan_ratio:.0%})") print(f" Dangling Wikilinks : {stats['dangling_count']:>5}") issues = evaluate_thresholds(stats, thresholds) print(f"\n── Threshold-Check ─────────────────────────────────────────") if issues: for issue in issues: print(f" {issue}") else: print(" ✓ Alle Thresholds OK") if stats["orphan_list"]: print(f"\n── Beispiel-Orphans (erste {min(10, len(stats['orphan_list']))}) ──────────────────────────────") for o in stats["orphan_list"][:10]: print(f" {o}") if stats["no_type_list"]: print(f"\n── Beispiel: fehlende type (erste {min(10, len(stats['no_type_list']))}) ────────────────────────") for n in stats["no_type_list"][:10]: print(f" {n}") print("=" * 60) # ─── Main ───────────────────────────────────────────────────────────────────── def main() -> int: ap = argparse.ArgumentParser( description="Schema-Coverage-Report für den Aria-Brain-Vault" ) ap.add_argument("--vault", type=Path, default=VAULT_DEFAULT, help=f"Brain-Vault-Root (default: {VAULT_DEFAULT})") ap.add_argument("--schema-pack", type=Path, default=SCHEMA_PACK_DEFAULT, help="Pfad zur schema-pack.yaml") ap.add_argument("--json", action="store_true", help="JSON-Output statt Human-Readable") args = ap.parse_args() try: import yaml # noqa: F401 except ImportError: sys.stderr.write("pyyaml nicht installiert: pip install pyyaml\n") return 3 pack = load_schema_pack(args.schema_pack) pack_id = pack.get("pack_id", "aria-base") if pack else "fallback" allowed_types = get_allowed_types(pack) thresholds = get_thresholds(pack) stats = scan_vault(args.vault, allowed_types) issues = evaluate_thresholds(stats, thresholds) if args.json: output = { "pack_id": pack_id, "vault": str(args.vault), "stats": stats, "thresholds": thresholds, "threshold_issues": issues, } print(json.dumps(output, ensure_ascii=False, indent=2)) else: print_human_report(stats, thresholds, args.vault, pack_id) return 1 if issues else 0 if __name__ == "__main__": sys.exit(main())