tests/diag_edges_for_note.py hinzugefügt

tests/diag_edges_for_note.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+Kleine Diagnosehilfe für mindnet_edges.
+
+Nutzung (aus Projektwurzel):
+
+    (.venv) python tests/diag_edges_for_note.py \
+        --note-id 20251130-relshow-embeddings-101
+
+Optional:
+
+    --prefix mindnet
+
+Ziel:
+    - Zeigt alle Edges aus der Collection <prefix>_edges,
+      die sich auf die angegebene note_id beziehen:
+        * note_id == NOTE
+        * oder source_id == NOTE
+        * oder target_id == NOTE
+"""
+
+from __future__ import annotations
+
+import argparse
+from typing import List
+
+from qdrant_client.http import models as rest
+
+from app.core.qdrant import QdrantConfig, get_client, collection_names
+
+
+def fetch_edges_for_note(note_id: str, prefix: str | None, limit: int = 512) -> List[dict]:
+    cfg = QdrantConfig.from_env()
+    if prefix:
+        cfg.prefix = prefix
+
+    client = get_client(cfg)
+    _, _, edges_col = collection_names(cfg.prefix)
+
+    conditions = []
+    for field in ("note_id", "source_id", "target_id"):
+        conditions.append(
+            rest.FieldCondition(
+                key=field,
+                match=rest.MatchValue(value=str(note_id)),
+            )
+        )
+
+    flt = rest.Filter(
+        should=conditions,
+    )
+
+    points, _ = client.scroll(
+        collection_name=edges_col,
+        scroll_filter=flt,
+        with_payload=True,
+        with_vectors=False,
+        limit=limit,
+    )
+
+    out: List[dict] = []
+    for p in points or []:
+        pl = dict(p.payload or {})
+        if pl:
+            out.append(pl)
+    return out
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--note-id", required=True, help="Note-ID, z. B. 20251130-relshow-embeddings-101")
+    parser.add_argument("--prefix", default=None, help="Qdrant-Prefix (Default: aus ENV/QdrantConfig)")
+    args = parser.parse_args()
+
+    note_id = args.note_id
+    prefix = args.prefix
+
+    edges = fetch_edges_for_note(note_id, prefix)
+    print(f"Prefix: {prefix or '(aus ENV)'}")
+    print(f"Note-ID: {note_id}")
+    print(f"Gefundene Edges: {len(edges)}")
+    print("-" * 60)
+
+    for i, e in enumerate(edges, start=1):
+        print(f"[{i}] kind={e.get('kind')}  confidence={e.get('confidence')}")
+        print(f"     note_id  = {e.get('note_id')}")
+        print(f"     source_id= {e.get('source_id')}")
+        print(f"     target_id= {e.get('target_id')}")
+        print()
+
+
+if __name__ == "__main__":
+    main()