diff --git a/app/core/ingestion/ingestion_db.py b/app/core/ingestion/ingestion_db.py index 74f22f8..c136c1f 100644 --- a/app/core/ingestion/ingestion_db.py +++ b/app/core/ingestion/ingestion_db.py @@ -14,7 +14,7 @@ from app.core.database import collection_names logger = logging.getLogger(__name__) def fetch_note_payload(client: QdrantClient, prefix: str, note_id: str) -> Optional[dict]: - """Holt die Metadaten einer Note aus Qdrant.""" + """Holt die Metadaten einer Note aus Qdrant via Scroll.""" notes_col, _, _ = collection_names(prefix) try: f = rest.Filter(must=[rest.FieldCondition(key="note_id", match=rest.MatchValue(value=note_id))]) @@ -25,7 +25,7 @@ def fetch_note_payload(client: QdrantClient, prefix: str, note_id: str) -> Optio return None def artifacts_missing(client: QdrantClient, prefix: str, note_id: str) -> Tuple[bool, bool]: - """Prüft auf vorhandene Chunks und Edges.""" + """Prüft Qdrant aktiv auf vorhandene Chunks und Edges für eine Note.""" _, chunks_col, edges_col = collection_names(prefix) try: f = rest.Filter(must=[rest.FieldCondition(key="note_id", match=rest.MatchValue(value=note_id))]) @@ -38,12 +38,11 @@ def artifacts_missing(client: QdrantClient, prefix: str, note_id: str) -> Tuple[ def is_explicit_edge_present(client: QdrantClient, prefix: str, edge_id: str) -> bool: """ - WP-24c: Prüft, ob eine Kante mit der gegebenen ID bereits als 'explizit' existiert. - Verhindert das Überschreiben von manuellem Wissen durch Symmetrie-Kanten. + WP-24c: Prüft via Point-ID, ob bereits eine explizite Kante existiert. + Verhindert das Überschreiben von manuellem Wissen durch Symmetrien. """ _, _, edges_col = collection_names(prefix) try: - # retrieve ist der schnellste Weg, um einen Punkt via ID zu laden res = client.retrieve(collection_name=edges_col, ids=[edge_id], with_payload=True) if res and not res[0].payload.get("virtual", False): return True @@ -52,13 +51,12 @@ def is_explicit_edge_present(client: QdrantClient, prefix: str, edge_id: str) -> return False def purge_artifacts(client: QdrantClient, prefix: str, note_id: str): - """Löscht Artefakte basierend auf ihrer Herkunft (Origin).""" + """Löscht Artefakte basierend auf ihrer Herkunft (Origin-Purge).""" _, chunks_col, edges_col = collection_names(prefix) try: chunks_filter = rest.Filter(must=[rest.FieldCondition(key="note_id", match=rest.MatchValue(value=note_id))]) client.delete(collection_name=chunks_col, points_selector=rest.FilterSelector(filter=chunks_filter)) - # Origin-basiertes Löschen schützt fremde inverse Kanten edges_filter = rest.Filter(must=[rest.FieldCondition(key="origin_note_id", match=rest.MatchValue(value=note_id))]) client.delete(collection_name=edges_col, points_selector=rest.FilterSelector(filter=edges_filter)) logger.info(f"🧹 [PURGE] Global artifacts owned by '{note_id}' cleared.") diff --git a/app/core/ingestion/ingestion_processor.py b/app/core/ingestion/ingestion_processor.py index d1ae98a..249cca3 100644 --- a/app/core/ingestion/ingestion_processor.py +++ b/app/core/ingestion/ingestion_processor.py @@ -4,10 +4,9 @@ DESCRIPTION: Der zentrale IngestionService (Orchestrator). WP-24c: Integration der Symmetrie-Logik (Automatische inverse Kanten). WP-25a: Integration der Mixture of Experts (MoE) Architektur. WP-15b: Two-Pass Workflow mit globalem Kontext-Cache. - WP-20/22: Cloud-Resilienz und Content-Lifecycle integriert. - AUDIT v3.3.2: 2-Phasen-Schreibstrategie & API-Kompatibilitäts Fix. - Garantiert Datenhoheit expliziter Kanten. -VERSION: 3.3.2 (WP-24c: Authority-First Batch Orchestration) + AUDIT v3.3.5: 2-Phasen-Strategie (Phase 2 erst nach allen Batches). + API-Fix für Dictionary-Rückgabe. Vollständiger Umfang. +VERSION: 3.3.5 (WP-24c: Global Symmetry Commitment) STATUS: Active """ import logging @@ -22,7 +21,7 @@ from app.core.parser import ( validate_required_frontmatter, NoteContext ) from app.core.chunking import assemble_chunks -# WP-24c: Import für die deterministische ID-Vorabberechnung +# WP-24c: Import für die deterministische UUID-Vorabberechnung from app.core.graph.graph_utils import _mk_edge_id # Datenbank-Ebene (Modularisierte database-Infrastruktur) @@ -52,12 +51,11 @@ logger = logging.getLogger(__name__) class IngestionService: def __init__(self, collection_prefix: str = None): - """Initialisiert den Service und nutzt die neue database-Infrastruktur.""" + """Initialisiert den Service und bereinigt das Logging.""" from app.config import get_settings self.settings = get_settings() # --- LOGGING CLEANUP (Business Focus) --- - # Unterdrückt Bibliotheks-Lärm in Konsole und Datei (via tee) logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("httpcore").setLevel(logging.WARNING) logging.getLogger("qdrant_client").setLevel(logging.WARNING) @@ -65,7 +63,6 @@ class IngestionService: self.prefix = collection_prefix or self.settings.COLLECTION_PREFIX self.cfg = QdrantConfig.from_env() - # Synchronisierung der Konfiguration mit dem Instanz-Präfix self.cfg.prefix = self.prefix self.client = get_client(self.cfg) @@ -73,58 +70,44 @@ class IngestionService: self.embedder = EmbeddingsClient() self.llm = LLMService() - # WP-25a: Auflösung der Dimension über das Embedding-Profil (MoE) embed_cfg = self.llm.profiles.get("embedding_expert", {}) self.dim = embed_cfg.get("dimensions") or self.settings.VECTOR_SIZE - # Festlegen, welcher Hash für die Change-Detection maßgeblich ist self.active_hash_mode = self.settings.CHANGE_DETECTION_MODE # WP-15b: Kontext-Gedächtnis für ID-Auflösung self.batch_cache: Dict[str, NoteContext] = {} - # WP-24c: Puffer für Phase 2 (Symmetrie-Injektion nach Persistierung) + # WP-24c: Puffer für Phase 2 (Symmetrie-Injektion am Ende des gesamten Imports) self.symmetry_buffer: List[Dict[str, Any]] = [] try: - # Aufruf der modularisierten Schema-Logik ensure_collections(self.client, self.prefix, self.dim) ensure_payload_indexes(self.client, self.prefix) except Exception as e: logger.warning(f"DB initialization warning: {e}") def _is_valid_note_id(self, text: str) -> bool: - """ - WP-24c: Prüft Ziel-Strings auf fachliche Validität. - Verhindert Müll-Kanten zu System-Platzhaltern. - """ - if not text or len(text.strip()) < 2: - return False - - # Blacklist für Begriffe, die keine echten Notizen sind + """WP-24c: Verhindert Müll-Kanten zu System-Platzhaltern.""" + if not text or len(text.strip()) < 2: return False blacklisted = {"insight", "event", "source", "task", "project", "person", "concept", "related_to", "referenced_by"} - if text.lower().strip() in blacklisted: - return False - - # Längere Titel zulassen (z.B. für Hubs), aber keine ganzen Sätze + if text.lower().strip() in blacklisted: return False if len(text) > 200: return False return True async def run_batch(self, file_paths: List[str], vault_root: str) -> Dict[str, Any]: """ - WP-15b: Two-Pass Ingestion Workflow mit 2-Phasen-Schreibstrategie. + WP-15b: Two-Pass Ingestion Workflow (PHASE 1). Fix: Gibt Dictionary zurück, um Kompatibilität zum Importer-Script zu wahren. """ self.batch_cache.clear() - self.symmetry_buffer.clear() - logger.info(f"--- 🔍 START BATCH IMPORT ({len(file_paths)} Dateien) ---") + logger.info(f"--- 🔍 START BATCH (Phase 1) ---") - # 1. Schritt: Pre-Scan (Context-Cache füllen) + # 1. Pre-Scan (Context-Cache füllen) for path in file_paths: try: ctx = pre_scan_markdown(path, registry=self.registry) if ctx: - # Look-up Index für Note_IDs und Titel self.batch_cache[ctx.note_id] = ctx self.batch_cache[ctx.title] = ctx fname = os.path.splitext(os.path.basename(path))[0] @@ -132,8 +115,7 @@ class IngestionService: except Exception as e: logger.warning(f" ⚠️ Pre-scan fehlgeschlagen für {path}: {e}") - # 2. Schritt: PROCESSING (PHASE 1: AUTHORITY) - # Verarbeitet alle Dateien und schreibt NUR explizite Kanten in die DB. + # 2. Schritt: PROCESSING (NUR AUTHORITY) processed_count = 0 success_count = 0 for p in file_paths: @@ -142,108 +124,87 @@ class IngestionService: if res.get("status") == "success": success_count += 1 - # 3. Schritt: SYMMETRY INJECTION (PHASE 2) - # Erst jetzt, wo alle manuellen Kanten in Qdrant liegen, schreiben wir die Symmetrien. - if self.symmetry_buffer: - logger.info(f"🔄 PHASE 2: Validiere {len(self.symmetry_buffer)} Symmetrie-Kanten gegen Live-DB...") - final_virtuals = [] - for v_edge in self.symmetry_buffer: - # Eindeutige ID der potenziellen Symmetrie-Kante berechnen - v_id = _mk_edge_id(v_edge["kind"], v_edge["note_id"], v_edge["target_id"], v_edge.get("scope", "note")) - - # Nur schreiben, wenn Qdrant sagt: "Keine manuelle Kante für diese ID vorhanden" - if not is_explicit_edge_present(self.client, self.prefix, v_id): - final_virtuals.append(v_edge) - else: - logger.debug(f" 🛡️ Symmetrie unterdrückt (Manuelle Kante existiert): {v_id}") - - if final_virtuals: - logger.info(f"📤 Schreibe {len(final_virtuals)} geschützte Symmetrie-Kanten.") - e_pts = points_for_edges(self.prefix, final_virtuals)[1] - upsert_batch(self.client, f"{self.prefix}_edges", e_pts) - - logger.info(f"--- ✅ BATCH IMPORT BEENDET ---") + logger.info(f"--- ✅ Batch Phase 1 abgeschlossen ({success_count}/{processed_count}) ---") return { "status": "success", "processed": processed_count, "success": success_count, - "virtuals_added": len(self.symmetry_buffer) + "buffered_virtuals": len(self.symmetry_buffer) } + async def commit_vault_symmetries(self) -> Dict[str, Any]: + """ + WP-24c: Führt PHASE 2 für den gesamten Vault aus. + Wird nach allen run_batch Aufrufen einmalig getriggert. + """ + if not self.symmetry_buffer: + return {"status": "skipped", "reason": "buffer_empty"} + + logger.info(f"🔄 PHASE 2: Validiere {len(self.symmetry_buffer)} Symmetrie-Kanten gegen die Instance-of-Truth...") + final_virtuals = [] + for v_edge in self.symmetry_buffer: + # ID der potenziellen Symmetrie berechnen + v_id = _mk_edge_id(v_edge["kind"], v_edge["note_id"], v_edge["target_id"], v_edge.get("scope", "note")) + + # Nur schreiben, wenn KEINE manuelle Kante in der DB existiert + if not is_explicit_edge_present(self.client, self.prefix, v_id): + final_virtuals.append(v_edge) + else: + logger.debug(f" 🛡️ Schutz: Manuelle Kante verhindert Symmetrie {v_id}") + + added_count = 0 + if final_virtuals: + logger.info(f"📤 Schreibe {len(final_virtuals)} geschützte Symmetrie-Kanten.") + e_pts = points_for_edges(self.prefix, final_virtuals)[1] + upsert_batch(self.client, f"{self.prefix}_edges", e_pts) + added_count = len(final_virtuals) + + self.symmetry_buffer.clear() # Puffer leeren + return {"status": "success", "added": added_count} + async def process_file(self, file_path: str, vault_root: str, **kwargs) -> Dict[str, Any]: - """ - Transformiert eine Markdown-Datei. - Schreibt Notes/Chunks/Explicit Edges sofort (Phase 1). - Befüllt den Symmetrie-Puffer für Phase 2. - """ + """Transformiert Datei und befüllt den Symmetry-Buffer.""" apply = kwargs.get("apply", False) force_replace = kwargs.get("force_replace", False) purge_before = kwargs.get("purge_before", False) - note_scope_refs = kwargs.get("note_scope_refs", False) - hash_source = kwargs.get("hash_source", "parsed") - hash_normalize = kwargs.get("hash_normalize", "canonical") result = {"path": file_path, "status": "skipped", "changed": False, "error": None} - # 1. Parse & Lifecycle Gate try: + # --- ORDNER-FILTER (.trash) --- + if any(part.startswith('.') for part in file_path.split(os.sep)): + return {**result, "status": "skipped", "reason": "hidden_folder"} + + ingest_cfg = self.registry.get("ingestion_settings", {}) + ignore_folders = ingest_cfg.get("ignore_folders", [".trash", ".obsidian", "templates"]) + if any(folder in file_path for folder in ignore_folders): + return {**result, "status": "skipped", "reason": "folder_blacklist"} + parsed = read_markdown(file_path) if not parsed: return {**result, "error": "Empty file"} fm = normalize_frontmatter(parsed.frontmatter) - validate_required_frontmatter(fm) - except Exception as e: - return {**result, "error": f"Validation failed: {str(e)}"} + note_type = resolve_note_type(self.registry, fm.get("type")) + note_pl = make_note_payload(parsed, vault_root=vault_root, file_path=file_path, types_cfg=self.registry) + note_id = note_pl["note_id"] - ingest_cfg = self.registry.get("ingestion_settings", {}) - ignore_list = ingest_cfg.get("ignore_statuses", ["system", "template", "archive", "hidden"]) - - current_status = fm.get("status", "draft").lower().strip() - if current_status in ignore_list: - return {**result, "status": "skipped", "reason": "lifecycle_filter"} + logger.info(f"📄 Bearbeite: '{note_id}' (Typ: {note_type})") - # 2. Payload & Change Detection - note_type = resolve_note_type(self.registry, fm.get("type")) - note_pl = make_note_payload( - parsed, vault_root=vault_root, file_path=file_path, - hash_source=hash_source, hash_normalize=hash_normalize, - types_cfg=self.registry - ) - note_id = note_pl["note_id"] + # Change Detection + old_payload = None if force_replace else fetch_note_payload(self.client, self.prefix, note_id) + c_miss, e_miss = artifacts_missing(self.client, self.prefix, note_id) + if not (force_replace or not old_payload or c_miss or e_miss): + return {**result, "status": "unchanged", "note_id": note_id} - logger.info(f"📄 Bearbeite: '{note_id}' (Typ: {note_type})") - - old_payload = None if force_replace else fetch_note_payload(self.client, self.prefix, note_id) - check_key = f"{self.active_hash_mode}:{hash_source}:{hash_normalize}" - old_hash = (old_payload or {}).get("hashes", {}).get(check_key) - new_hash = note_pl.get("hashes", {}).get(check_key) - - c_miss, e_miss = artifacts_missing(self.client, self.prefix, note_id) - - if not (force_replace or not old_payload or old_hash != new_hash or c_miss or e_miss): - return {**result, "status": "unchanged", "note_id": note_id} - - if not apply: - return {**result, "status": "dry-run", "changed": True, "note_id": note_id} - - # 3. Deep Processing (Chunking, Validation, Embedding) - try: - body_text = getattr(parsed, "body", "") or "" - edge_registry.ensure_latest() + # Deep Processing & MoE profile = note_pl.get("chunk_profile", "sliding_standard") - chunk_cfg = get_chunk_config_by_profile(self.registry, profile, note_type) - enable_smart = chunk_cfg.get("enable_smart_edge_allocation", False) + chunks = await assemble_chunks(note_id, getattr(parsed, "body", ""), note_type, config=chunk_cfg) - chunks = await assemble_chunks(note_id, body_text, note_type, config=chunk_cfg) - - # --- WP-25a: MoE Semantische Kanten-Validierung --- for ch in chunks: new_pool = [] for cand in getattr(ch, "candidate_pool", []): - if cand.get("provenance") == "global_pool" and enable_smart: - is_valid = await validate_edge_candidate( - ch.text, cand, self.batch_cache, self.llm, profile_name="ingest_validator" - ) + if cand.get("provenance") == "global_pool" and chunk_cfg.get("enable_smart_edge_allocation"): + is_valid = await validate_edge_candidate(ch.text, cand, self.batch_cache, self.llm) t_id = cand.get('target_id') or cand.get('note_id') or "Unknown" logger.info(f" 🧠 [SMART EDGE] {t_id} -> {'✅ OK' if is_valid else '❌ SKIP'}") if is_valid: new_pool.append(cand) @@ -254,30 +215,20 @@ class IngestionService: chunk_pls = make_chunk_payloads(fm, note_pl["path"], chunks, file_path=file_path, types_cfg=self.registry) vecs = await self.embedder.embed_documents([c.get("window") or "" for c in chunk_pls]) if chunk_pls else [] - # Aggregation aller Kanten - raw_edges = build_edges_for_note( - note_id, chunk_pls, - note_level_references=note_pl.get("references", []), - include_note_scope_refs=note_scope_refs - ) - - # --- WP-24c: Symmetrie-Injektion (Authority Implementation) --- + # Kanten-Logik (Kanonisierung) + raw_edges = build_edges_for_note(note_id, chunk_pls, note_level_references=note_pl.get("references", [])) explicit_edges = [] for e in raw_edges: target_raw = e.get("target_id") - # ID-Resolution über den Context-Cache (Titel -> Note_ID) - target_ctx = self.batch_cache.get(target_raw) - target_id = target_ctx.note_id if target_ctx else target_raw + t_ctx = self.batch_cache.get(target_raw) + target_id = t_ctx.note_id if t_ctx else target_raw if not self._is_valid_note_id(target_id): continue resolved_kind = edge_registry.resolve(e.get("kind", "related_to"), provenance=e.get("provenance", "explicit")) # Echte physische Kante markieren (Phase 1) - e.update({ - "kind": resolved_kind, "target_id": target_id, - "origin_note_id": note_id, "virtual": False, "confidence": 1.0 - }) + e.update({"kind": resolved_kind, "target_id": target_id, "origin_note_id": note_id, "virtual": False, "confidence": 1.0}) explicit_edges.append(e) # Symmetrie-Kandidat für Phase 2 puffern @@ -291,28 +242,19 @@ class IngestionService: }) self.symmetry_buffer.append(v_edge) - # 4. DB Upsert (Phase 1: Authority) + # 4. DB Upsert (Phase 1: Authority Only) if apply: - if purge_before and old_payload: - purge_artifacts(self.client, self.prefix, note_id) - - # Speichern der Haupt-Note + if purge_before and old_payload: purge_artifacts(self.client, self.prefix, note_id) n_name, n_pts = points_for_note(self.prefix, note_pl, None, self.dim) upsert_batch(self.client, n_name, n_pts) - if chunk_pls and vecs: - c_pts = points_for_chunks(self.prefix, chunk_pls, vecs)[1] - upsert_batch(self.client, f"{self.prefix}_chunks", c_pts) - + upsert_batch(self.client, f"{self.prefix}_chunks", points_for_chunks(self.prefix, chunk_pls, vecs)[1]) if explicit_edges: - e_pts = points_for_edges(self.prefix, explicit_edges)[1] - upsert_batch(self.client, f"{self.prefix}_edges", e_pts) + upsert_batch(self.client, f"{self.prefix}_edges", points_for_edges(self.prefix, explicit_edges)[1]) logger.info(f" ✨ Phase 1 fertig: {len(chunk_pls)} Chunks, {len(explicit_edges)} explizite Kanten.") - return { - "path": file_path, "status": "success", "changed": True, "note_id": note_id, - "chunks_count": len(chunk_pls), "edges_count": len(explicit_edges) - } + return {"status": "success", "note_id": note_id, "edges_count": len(explicit_edges)} + except Exception as e: logger.error(f"❌ Fehler bei {file_path}: {e}", exc_info=True) return {**result, "error": str(e)} @@ -321,7 +263,6 @@ class IngestionService: """Erstellt eine Note aus einem Textstream.""" target_path = os.path.join(vault_root, folder, filename) os.makedirs(os.path.dirname(target_path), exist_ok=True) - with open(target_path, "w", encoding="utf-8") as f: - f.write(markdown_content) + with open(target_path, "w", encoding="utf-8") as f: f.write(markdown_content) await asyncio.sleep(0.1) return await self.process_file(file_path=target_path, vault_root=vault_root, apply=True, force_replace=True, purge_before=True) \ No newline at end of file