#!/usr/bin/env python3 """Prüft den Vault auf alles, was die Referenzschicht unbrauchbar macht. Standard ist ein reiner Bericht. `--fix-missing --apply` ergänzt ausschließlich fehlende Frontmatter-Felder (uid, type, created, updated, status) und fasst nie bestehende Werte oder Fließtext an. Beispiele: python3 scripts/vault_doctor.py --vault ~/Vault python3 scripts/vault_doctor.py --stale-days 180 python3 scripts/vault_doctor.py --fix-missing # Trockenlauf python3 scripts/vault_doctor.py --fix-missing --apply """ from __future__ import annotations import argparse import os import sys from collections import defaultdict sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import vaultlib as V # noqa: E402 REQUIRED = ["type", "title", "status", "summary", "updated"] def collect(vault: str): notes = [] for ap, rp in V.walk_vault(vault): if rp.endswith(".md"): with open(ap, "r", encoding="utf-8", errors="replace") as fh: text = fh.read() fm, body = V.parse_frontmatter(text) notes.append({"abs": ap, "rel": rp, "fm": fm, "body": body, "raw": text}) return notes def section(title: str, items, formatter=str, limit=25): if not items: return [] out = [f"\n## {title} ({len(items)})"] for it in items[:limit]: out.append(" - " + formatter(it)) if len(items) > limit: out.append(f" … {len(items) - limit} weitere") return out def main() -> int: ap = argparse.ArgumentParser(description="Vault-Gesundheitscheck") ap.add_argument("--vault", default=".") ap.add_argument("--stale-days", type=int, default=365) ap.add_argument("--inbox-days", type=int, default=14) ap.add_argument("--fix-missing", action="store_true", help="fehlende Frontmatter-Felder ergänzen (Trockenlauf ohne --apply)") ap.add_argument("--apply", action="store_true", help="Änderungen wirklich schreiben") args = ap.parse_args() vault = os.path.abspath(os.path.expanduser(args.vault)) notes = collect(vault) if not notes: print("Keine Markdown-Dateien gefunden.") return 1 lookup = set() for n in notes: rel = n["rel"] lookup |= {rel, os.path.splitext(rel)[0], os.path.basename(rel), os.path.splitext(os.path.basename(rel))[0]} if n["fm"].get("title"): lookup.add(str(n["fm"]["title"])) if n["fm"].get("uid"): lookup.add(str(n["fm"]["uid"])) no_fm, missing_fields, bad_type, bad_status = [], [], [], [] broken, dup_uid, dup_title, stale, old_inbox = [], defaultdict(list), defaultdict(list), [], [] inbound = defaultdict(int) import datetime as dt today = dt.date.today() for n in notes: fm, rel = n["fm"], n["rel"] if not fm: no_fm.append(rel) else: miss = [f for f in REQUIRED if not str(fm.get(f, "")).strip()] if miss: missing_fields.append((rel, miss)) t = str(fm.get("type", "")).strip() if t and t not in V.KNOWN_TYPES: bad_type.append((rel, t)) s = str(fm.get("status", "")).strip() if s and s not in V.KNOWN_STATUS: bad_status.append((rel, s)) if fm.get("uid"): dup_uid[str(fm["uid"])].append(rel) title = V.derive_title(fm, n["body"], rel) dup_title[title.lower()].append(rel) for raw in V.extract_links(n["body"]): t = raw.strip().lstrip("./") if t in lookup or t + ".md" in lookup or os.path.basename(t) in lookup: inbound[t] += 1 else: broken.append((rel, raw)) mtime = dt.date.fromtimestamp(os.path.getmtime(n["abs"])) upd = str(fm.get("updated", "")).strip() try: ref = dt.date.fromisoformat(upd[:10]) if upd else mtime except ValueError: ref = mtime age = (today - ref).days if age > args.stale_days: stale.append((rel, age)) if rel.startswith("00_Inbox/") and age > args.inbox_days: old_inbox.append((rel, age)) linked = set() for n in notes: for raw in V.extract_links(n["body"]): t = raw.strip().lstrip("./") for cand in (t, t + ".md", os.path.basename(t), os.path.basename(t) + ".md"): linked.add(cand) orphans = [ n["rel"] for n in notes if not (set([n["rel"], os.path.splitext(n["rel"])[0], os.path.basename(n["rel"]), os.path.splitext(os.path.basename(n["rel"]))[0]]) & linked) and not n["rel"].startswith(("_index/", "40_Templates/")) ] report = [f"# Vault-Doctor — {len(notes)} Notizen in {vault}"] report += section("Ohne Frontmatter", no_fm) report += section("Pflichtfelder fehlen", missing_fields, lambda x: f"{x[0]} → fehlt: {', '.join(x[1])}") report += section("Unbekannter type", bad_type, lambda x: f"{x[0]} → '{x[1]}'") report += section("Unbekannter status", bad_status, lambda x: f"{x[0]} → '{x[1]}'") report += section("Tote Wikilinks", broken, lambda x: f"{x[0]} → [[{x[1]}]]") report += section("Doppelte uid", [(k, v) for k, v in dup_uid.items() if len(v) > 1], lambda x: f"{x[0]}: {', '.join(x[1])}") report += section("Doppelte Titel", [(k, v) for k, v in dup_title.items() if len(v) > 1], lambda x: f"{x[0]}: {', '.join(x[1])}") report += section("Verwaist (keine eingehenden Links)", orphans) report += section(f"Liegt länger als {args.inbox_days} Tage in der Inbox", old_inbox, lambda x: f"{x[0]} ({x[1]} Tage)") report += section(f"Seit über {args.stale_days} Tagen unangetastet", stale, lambda x: f"{x[0]} ({x[1]} Tage)") problems = (len(no_fm) + len(missing_fields) + len(bad_type) + len(bad_status) + len(broken) + len(orphans)) if problems == 0: report.append("\nKeine strukturellen Probleme gefunden.") else: report.append(f"\nBefunde gesamt: {problems}") print("\n".join(report)) if args.fix_missing: changed = [] for n in notes: fm, rel = dict(n["fm"]), n["rel"] before = dict(fm) if not str(fm.get("type", "")).strip(): fm["type"] = V.infer_type(rel) if not str(fm.get("title", "")).strip(): fm["title"] = V.derive_title(n["fm"], n["body"], rel) if not str(fm.get("uid", "")).strip(): import datetime as _d stamp = _d.date.fromtimestamp(os.path.getmtime(n["abs"])).strftime("%Y%m%d") fm["uid"] = f"{stamp}-{V.slugify(fm['title'])[:40]}" if not str(fm.get("status", "")).strip(): if rel.startswith("90_Archive/"): fm["status"] = "archived" elif rel.startswith("00_Inbox/"): fm["status"] = "idea" else: fm["status"] = "active" if not str(fm.get("summary", "")).strip(): s = V.derive_summary(n["body"]) if s: fm["summary"] = s if not str(fm.get("created", "")).strip(): import datetime as _d fm["created"] = _d.date.fromtimestamp(os.path.getmtime(n["abs"])).isoformat() if not str(fm.get("updated", "")).strip(): fm["updated"] = fm["created"] ordered = {} for k in ("uid", "type", "title", "status", "summary", "tags", "project", "created", "updated"): if k in fm: ordered[k] = fm[k] for k, v in fm.items(): ordered.setdefault(k, v) if ordered != before: changed.append((n, ordered)) print(f"\n## Frontmatter-Ergänzungen: {len(changed)} Datei(en)") for n, fm in changed[:15]: added = [k for k in fm if k not in n["fm"]] print(f" - {n['rel']}: + {', '.join(added) or 'Feldreihenfolge'}") if len(changed) > 15: print(f" … {len(changed) - 15} weitere") if not args.apply: print("\n(Trockenlauf — mit --apply schreiben)") else: for n, fm in changed: body = n["body"] if n["fm"] else n["raw"] with open(n["abs"], "w", encoding="utf-8") as fh: fh.write(V.dump_frontmatter(fm) + "\n" + body.lstrip("\n")) print(f"\n{len(changed)} Datei(en) geschrieben. " "Jetzt build_index.py erneut laufen lassen.") return 0 if __name__ == "__main__": raise SystemExit(main())