264 lines
8.7 KiB
Python
264 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Script: export_markdown.py
|
||
Version: 1.3.0
|
||
Datum: 2025-09-09
|
||
|
||
Kurzbeschreibung
|
||
---------------
|
||
Exportiert Markdown-Notizen aus Qdrant in einen Obsidian-kompatiblen Vault-Ordner.
|
||
Für jede Note wird die YAML-Frontmatter + Body rekonstruiert.
|
||
|
||
Body-Rekonstruktion (Priorität):
|
||
1) Aus notes.payload.fulltext, falls vorhanden (verlustfreie Rückführung).
|
||
2) Andernfalls werden alle Chunks der Note geladen und deren Textfelder
|
||
(text -> content -> raw) in stabiler Reihenfolge zusammengefügt.
|
||
|
||
Wichtige Hinweise
|
||
-----------------
|
||
- Qdrant-Zugriff wird über Umgebungsvariablen konfiguriert:
|
||
* QDRANT_URL (z. B. http://127.0.0.1:6333)
|
||
* QDRANT_API_KEY (optional)
|
||
* COLLECTION_PREFIX (z. B. mindnet)
|
||
* VECTOR_DIM (wird hier nur zum Collection-Setup benötigt; Standard 384)
|
||
- Es wird **kein** --prefix Parameter erwartet; das Präfix kommt aus den Umgebungsvariablen.
|
||
- Exportiert nach --out. Unterordner gemäß payload['path'] werden automatisch angelegt.
|
||
- Standard: bestehende Dateien werden NICHT überschrieben; mit --overwrite schon.
|
||
|
||
Aufrufparameter
|
||
---------------
|
||
--out PATH Ziel-Ordner (Pfad zum Export-Vault) [erforderlich]
|
||
--note-id ID Nur eine einzelne Note exportieren (optional)
|
||
--overwrite Ziel-Dateien überschreiben, falls vorhanden (optional)
|
||
|
||
Beispiele
|
||
---------
|
||
COLLECTION_PREFIX=mindnet QDRANT_URL=http://127.0.0.1:6333 \\
|
||
python3 -m scripts.export_markdown --out ./_exportVault
|
||
|
||
Nur eine Note:
|
||
COLLECTION_PREFIX=mindnet python3 -m scripts.export_markdown \\
|
||
--out ./_exportVault --note-id 20250821-architektur-ki-trainerassistent-761cfe
|
||
|
||
Mit Überschreiben:
|
||
COLLECTION_PREFIX=mindnet python3 -m scripts.export_markdown \\
|
||
--out ./_exportVault --overwrite
|
||
|
||
Änderungen (1.3.0)
|
||
------------------
|
||
- Body-Rekonstruktion robust gemacht:
|
||
* nutzt 'fulltext' aus Notes-Payload, falls vorhanden
|
||
* sonst Zusammenbau aus Chunks; Fallbacks für Textfelder: 'text' -> 'content' -> 'raw'
|
||
* stabile Sortierung der Chunks: seq -> chunk_index -> Nummer in chunk_id
|
||
- Entfernt veralteten --prefix Parameter; nutzt QdrantConfig.from_env()
|
||
- Verbesserte YAML-Ausgabe (Strings bleiben Strings), saubere Trennlinie '---'
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
from typing import List, Optional, Tuple
|
||
|
||
import yaml
|
||
from qdrant_client.http import models as rest
|
||
|
||
from app.core.qdrant import QdrantConfig, get_client, ensure_collections
|
||
from qdrant_client import QdrantClient
|
||
|
||
|
||
def _names(prefix: str) -> Tuple[str, str, str]:
|
||
return f"{prefix}_notes", f"{prefix}_chunks", f"{prefix}_edges"
|
||
|
||
|
||
def _ensure_dir(path: str) -> None:
|
||
d = os.path.dirname(path)
|
||
if d and not os.path.isdir(d):
|
||
os.makedirs(d, exist_ok=True)
|
||
|
||
|
||
def _to_md(frontmatter: dict, body: str) -> str:
|
||
fm = yaml.safe_dump(frontmatter, sort_keys=False, allow_unicode=True).strip()
|
||
# Frontmatter, dann eine leere Zeile, danach der Body
|
||
return f"---\n{fm}\n---\n{body.rstrip()}\n"
|
||
|
||
|
||
def _scroll_all(
|
||
client: QdrantClient,
|
||
collection: str,
|
||
flt: Optional[rest.Filter] = None,
|
||
with_payload: bool = True,
|
||
with_vectors: bool = False,
|
||
limit: int = 256,
|
||
) -> List:
|
||
"""Holt *alle* Punkte via Scroll (QdrantClient.scroll liefert (points, next_offset))."""
|
||
pts_all = []
|
||
next_offset = None
|
||
while True:
|
||
points, next_offset = client.scroll(
|
||
collection_name=collection,
|
||
scroll_filter=flt,
|
||
with_payload=with_payload,
|
||
with_vectors=with_vectors,
|
||
limit=limit,
|
||
offset=next_offset,
|
||
)
|
||
pts_all.extend(points or [])
|
||
if not next_offset:
|
||
break
|
||
return pts_all
|
||
|
||
|
||
_NUM_IN_CHUNK_ID = re.compile(r"#(?:c)?(\d+)$")
|
||
|
||
|
||
def _chunk_sort_key(pl: dict) -> Tuple[int, int, str]:
|
||
"""
|
||
Stabile Reihenfolge:
|
||
1) 'seq' (falls vorhanden),
|
||
2) 'chunk_index' (falls vorhanden),
|
||
3) Nummer aus 'chunk_id' Suffix (#c02 -> 2),
|
||
4) als letzter Fallback: gesamte 'chunk_id' als string.
|
||
"""
|
||
seq = pl.get("seq")
|
||
if isinstance(seq, int):
|
||
return (0, seq, "")
|
||
idx = pl.get("chunk_index")
|
||
if isinstance(idx, int):
|
||
return (1, idx, "")
|
||
cid = str(pl.get("chunk_id") or "")
|
||
m = _NUM_IN_CHUNK_ID.search(cid)
|
||
if m:
|
||
try:
|
||
return (2, int(m.group(1)), "")
|
||
except ValueError:
|
||
pass
|
||
return (3, 0, cid)
|
||
|
||
|
||
def _join_chunk_texts(chunks_payloads: List[dict]) -> str:
|
||
"""Nimmt die sortierten Chunk-Payloads und baut den Body zusammen."""
|
||
parts: List[str] = []
|
||
for pl in chunks_payloads:
|
||
txt = pl.get("text") or pl.get("content") or pl.get("raw") or ""
|
||
if txt:
|
||
parts.append(txt.rstrip())
|
||
# Doppelte Leerzeile zwischen Chunks – in Markdown meist ein guter Standard
|
||
return ("\n\n".join(parts)).rstrip() + ("\n" if parts else "")
|
||
|
||
|
||
def _export_one_note(
|
||
client: QdrantClient,
|
||
prefix: str,
|
||
note_pl: dict,
|
||
out_root: str,
|
||
overwrite: bool,
|
||
) -> dict:
|
||
notes_col, chunks_col, _ = _names(prefix)
|
||
note_id = note_pl.get("note_id") or note_pl.get("id")
|
||
path = note_pl.get("path") or f"{note_id}.md"
|
||
|
||
# Zielpfad relativ zu out_root
|
||
out_path = os.path.join(out_root, path).replace("\\", "/")
|
||
|
||
# Frontmatter aus Payload zurückführen (nur bekannte Felder)
|
||
fm = {}
|
||
# Bewährte Felder zurückschreiben – unbekannte Keys nicht in YAML aufnehmen
|
||
for k in [
|
||
"title", "id", "type", "status", "created", "updated", "tags",
|
||
"priority", "effort_min", "due", "people", "aliases",
|
||
"depends_on", "assigned_to", "lang"
|
||
]:
|
||
v = note_pl.get(k) if k in note_pl else note_pl.get(f"note_{k}") # Toleranz für evtl. Namensvarianten
|
||
if v not in (None, [], ""):
|
||
fm[k] = v
|
||
|
||
# Pflichtfelder sicherstellen
|
||
fm["id"] = fm.get("id") or note_id
|
||
fm["title"] = fm.get("title") or note_pl.get("title") or note_id
|
||
fm["type"] = fm.get("type") or "concept"
|
||
fm["status"] = fm.get("status") or "draft"
|
||
|
||
# Body-Rekonstruktion
|
||
body = ""
|
||
fulltext = note_pl.get("fulltext")
|
||
if isinstance(fulltext, str) and fulltext.strip():
|
||
body = fulltext
|
||
else:
|
||
# Chunks zur Note holen und sortieren
|
||
flt = rest.Filter(must=[rest.FieldCondition(
|
||
key="note_id",
|
||
match=rest.MatchValue(value=note_id)
|
||
)])
|
||
chunk_pts = _scroll_all(client, chunks_col, flt, with_payload=True, with_vectors=False)
|
||
# Sortierung
|
||
chunk_payloads = [p.payload or {} for p in chunk_pts]
|
||
chunk_payloads.sort(key=_chunk_sort_key)
|
||
body = _join_chunk_texts(chunk_payloads)
|
||
|
||
# Ziel schreiben
|
||
if (not overwrite) and os.path.exists(out_path):
|
||
return {"note_id": note_id, "path": out_path, "status": "skip_exists"}
|
||
|
||
_ensure_dir(out_path)
|
||
with open(out_path, "w", encoding="utf-8") as f:
|
||
f.write(_to_md(fm, body))
|
||
|
||
return {
|
||
"note_id": note_id,
|
||
"path": out_path,
|
||
"status": "written",
|
||
"body_from": "fulltext" if isinstance(fulltext, str) and fulltext.strip() else "chunks",
|
||
"chunks_used": None if (isinstance(fulltext, str) and fulltext.strip()) else len(chunk_payloads),
|
||
}
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--out", required=True, help="Zielordner für den Export-Vault")
|
||
ap.add_argument("--note-id", help="Nur eine Note exportieren (Note-ID)")
|
||
ap.add_argument("--overwrite", action="store_true", help="Bestehende Dateien überschreiben")
|
||
args = ap.parse_args()
|
||
|
||
# Qdrant-Konfiguration
|
||
cfg = QdrantConfig.from_env()
|
||
client = get_client(cfg)
|
||
ensure_collections(client, cfg.prefix, cfg.dim)
|
||
|
||
notes_col, _, _ = _names(cfg.prefix)
|
||
|
||
# Notes holen
|
||
flt = None
|
||
if args.note_id:
|
||
flt = rest.Filter(must=[rest.FieldCondition(
|
||
key="note_id",
|
||
match=rest.MatchValue(value=args.note_id)
|
||
)])
|
||
|
||
note_pts = _scroll_all(client, notes_col, flt, with_payload=True, with_vectors=False)
|
||
|
||
if not note_pts:
|
||
print(json.dumps({"exported": 0, "out": args.out, "message": "Keine Notes gefunden."}, ensure_ascii=False))
|
||
return
|
||
|
||
results = []
|
||
for p in note_pts:
|
||
pl = p.payload or {}
|
||
try:
|
||
res = _export_one_note(client, cfg.prefix, pl, args.out, args.overwrite)
|
||
except Exception as e:
|
||
res = {"note_id": pl.get("note_id") or pl.get("id"), "error": str(e)}
|
||
results.append(res)
|
||
|
||
print(json.dumps({"exported": len([r for r in results if r.get('status') == 'written']),
|
||
"skipped": len([r for r in results if r.get('status') == 'skip_exists']),
|
||
"out": args.out,
|
||
"details": results}, ensure_ascii=False))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|