Lars
e93bab6ea7
All checks were successful
Deploy mindnet to llm-node / deploy (push) Successful in 4s
72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
scripts/show_edges_for_note.py
|
|
|
|
Zeigt Kanten einer Note (gefiltert nach kind/scope) in einer kompakten Form.
|
|
Aufrufbeispiele:
|
|
python3 -m scripts.show_edges_for_note --note-id 20251110-ollama-llm-9f0a12 --kinds references,next,prev --limit 10
|
|
python3 -m scripts.show_edges_for_note --title "Qdrant Vektordatenbank" --scope note
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
import argparse, json, os, sys
|
|
from typing import Dict, Any, List, Tuple
|
|
from qdrant_client.http import models as rest
|
|
|
|
from app.core.database.qdrant import QdrantConfig, get_client
|
|
|
|
def collections(prefix: str) -> Tuple[str, str, str]:
|
|
return f"{prefix}_notes", f"{prefix}_chunks", f"{prefix}_edges"
|
|
|
|
def find_note_by_title(client, prefix: str, title: str) -> str | None:
|
|
notes_col, _, _ = collections(prefix)
|
|
f = rest.Filter(must=[rest.FieldCondition(key="title", match=rest.MatchText(text=title))])
|
|
pts, _ = client.scroll(collection_name=notes_col, scroll_filter=f, with_payload=True, with_vectors=False, limit=1)
|
|
if not pts:
|
|
return None
|
|
return pts[0].payload.get("note_id")
|
|
|
|
def fetch_edges_for_note(client, prefix: str, note_id: str, kinds: List[str] | None, scope: str | None, limit: int) -> List[Dict[str, Any]]:
|
|
_, _, edges_col = collections(prefix)
|
|
must = [rest.FieldCondition(key="note_id", match=rest.MatchValue(value=note_id))]
|
|
if scope:
|
|
must.append(rest.FieldCondition(key="scope", match=rest.MatchValue(value=scope)))
|
|
if kinds:
|
|
must.append(rest.FieldCondition(key="kind", match=rest.MatchAny(any=kinds)))
|
|
f = rest.Filter(must=must)
|
|
pts, _ = client.scroll(collection_name=edges_col, scroll_filter=f, with_payload=True, with_vectors=False, limit=limit)
|
|
return [p.payload for p in pts]
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--note-id")
|
|
ap.add_argument("--title")
|
|
ap.add_argument("--kinds", help="CSV: references,next,prev,belongs_to,backlink")
|
|
ap.add_argument("--scope", choices=["note","chunk"])
|
|
ap.add_argument("--limit", type=int, default=25)
|
|
args = ap.parse_args()
|
|
|
|
cfg = QdrantConfig.from_env()
|
|
client = get_client(cfg)
|
|
|
|
nid = args.note_id
|
|
if not nid and args.title:
|
|
nid = find_note_by_title(client, cfg.prefix, args.title)
|
|
if not nid:
|
|
print(json.dumps({"error": f"note with title '{args.title}' not found"}))
|
|
sys.exit(2)
|
|
if not nid:
|
|
print(json.dumps({"error": "please provide --note-id or --title"}))
|
|
sys.exit(2)
|
|
|
|
kinds = None
|
|
if args.kinds:
|
|
kinds = [s.strip() for s in args.kinds.split(",") if s.strip()]
|
|
|
|
edges = fetch_edges_for_note(client, cfg.prefix, nid, kinds, args.scope, args.limit)
|
|
print(json.dumps({"note_id": nid, "count": len(edges), "edges": edges}, ensure_ascii=False, indent=2))
|
|
|
|
if __name__ == "__main__":
|
|
main()
|