diff --git a/tests/diag_edges_for_note.py b/tests/diag_edges_for_note.py new file mode 100644 index 0000000..8aa2494 --- /dev/null +++ b/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 _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()