tests/audit_edges_now.py hinzugefügt

tests/audit_edges_now.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+tests/audit_edges_now.py — Zählt Edges nach kind & scope und zeigt Chunk- und Note-Anzahl.
+"""
+import os, sys, json
+PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+if PROJECT_ROOT not in sys.path:
+    sys.path.insert(0, PROJECT_ROOT)
+
+from qdrant_client.http import models as rest
+from app.core.qdrant import QdrantConfig, get_client
+
+def scroll_all(c, col, flt=None):
+    out, nextp = [], None
+    while True:
+        pts, nextp = c.scroll(collection_name=col, with_payload=True, with_vectors=False, limit=256, scroll_filter=flt, offset=nextp)
+        if not pts: break
+        out.extend(pts)
+        if nextp is None: break
+    return out
+
+def main():
+    cfg = QdrantConfig.from_env()
+    c = get_client(cfg)
+    notes = f"{cfg.prefix}_notes"
+    chunks = f"{cfg.prefix}_chunks"
+    edges = f"{cfg.prefix}_edges"
+
+    es = scroll_all(c, edges)
+    counts = {}
+    by_scope = {}
+    for p in es:
+        pl = p.payload or {}
+        k = pl.get("kind") or "?"
+        s = pl.get("scope") or "?"
+        counts[k] = counts.get(k, 0) + 1
+        key = f"{k}:{s}"
+        by_scope[key] = by_scope.get(key, 0) + 1
+
+    print(json.dumps({
+        "collections": {"notes": notes, "chunks": chunks, "edges": edges},
+        "total_edges": len(es),
+        "counts": counts,
+        "counts_by_scope": by_scope
+    }, ensure_ascii=False, indent=2))
+
+if __name__ == "__main__":
+    main()