#!/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()