""" FILE: app/core/ingestion/ingestion_processor.py 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.0: Einführung des 2-Phasen-Upserts. Garantiert, dass explizite Kanten niemals durch Symmetrien überschrieben werden. VERSION: 3.3.0 (WP-24c: Two-Phase Writing Strategy) STATUS: Active """ import logging import asyncio import os import re from typing import Dict, List, Optional, Tuple, Any # Core Module Imports from app.core.parser import ( read_markdown, pre_scan_markdown, normalize_frontmatter, validate_required_frontmatter, NoteContext ) from app.core.chunking import assemble_chunks # WP-24c: Import für die deterministische ID-Vorabberechnung from app.core.graph.graph_utils import _mk_edge_id # MODULARISIERUNG: Neue Import-Pfade für die Datenbank-Ebene from app.core.database.qdrant import QdrantConfig, get_client, ensure_collections, ensure_payload_indexes from app.core.database.qdrant_points import points_for_chunks, points_for_note, points_for_edges, upsert_batch from qdrant_client.http import models as rest # Services from app.services.embeddings_client import EmbeddingsClient from app.services.edge_registry import registry as edge_registry from app.services.llm_service import LLMService # Package-Interne Imports from .ingestion_utils import load_type_registry, resolve_note_type, get_chunk_config_by_profile from .ingestion_db import fetch_note_payload, artifacts_missing, purge_artifacts, is_explicit_edge_present from .ingestion_validation import validate_edge_candidate from .ingestion_note_payload import make_note_payload from .ingestion_chunk_payload import make_chunk_payloads # Fallback für Edges (Struktur-Verknüpfung) try: from app.core.graph.graph_derive_edges import build_edges_for_note except ImportError: def build_edges_for_note(*args, **kwargs): return [] logger = logging.getLogger(__name__) class IngestionService: def __init__(self, collection_prefix: str = None): """Initialisiert den Service und nutzt die neue database-Infrastruktur.""" from app.config import get_settings self.settings = get_settings() # --- LOGGING CLEANUP --- # 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) logging.getLogger("urllib3").setLevel(logging.WARNING) 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) self.registry = load_type_registry() 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 self.batch_cache: Dict[str, NoteContext] = {} # WP-15b LocalBatchCache 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 Validität. Filtert Begriffe wie 'insight' oder 'event' aus, um Müll-Kanten zu vermeiden. """ if not text or len(text.strip()) < 2: return False # Symmetrie-Filter gegen Typ-Strings blacklisted = {"insight", "event", "source", "task", "project", "person", "concept", "related_to", "referenced_by"} if text.lower().strip() in blacklisted: return False if len(text) > 120: return False # Wahrscheinlich kein Titel return True async def run_batch(self, file_paths: List[str], vault_root: str) -> List[Dict[str, Any]]: """ WP-15b: Implementiert den Two-Pass Ingestion Workflow. Führt nun zusätzlich das 2-Phasen-Schreiben aus. """ logger.info(f"--- 🔍 START BATCH IMPORT ({len(file_paths)} Dateien) ---") # 1. Schritt: Context-Cache füllen for path in file_paths: try: ctx = pre_scan_markdown(path, registry=self.registry) if ctx: self.batch_cache[ctx.note_id] = ctx self.batch_cache[ctx.title] = ctx fname = os.path.splitext(os.path.basename(path))[0] self.batch_cache[fname] = ctx except Exception as e: logger.warning(f"⚠️ Pre-scan failed for {path}: {e}") # 2. Schritt: Verarbeitung & Schreiben (PHASE 1: AUTHORITY) # Wir sammeln alle Symmetrie-Kandidaten, um sie in Phase 2 zu prüfen. results = [] all_virtual_candidates = [] for p in file_paths: res, candidates = await self.process_file(p, vault_root, apply=True, purge_before=True, skip_virtuals=True) results.append(res) all_virtual_candidates.extend(candidates) # 3. Schritt: Symmetrie-Einspeisung (PHASE 2: SYMMETRY) if all_virtual_candidates: logger.info(f"🔄 PHASE 2: Prüfe {len(all_virtual_candidates)} Symmetrie-Kanten gegen die Datenbank...") final_virtuals = [] for v_edge in all_virtual_candidates: # Eindeutige ID für diese Symmetrie-Kante berechnen v_id = _mk_edge_id(v_edge["kind"], v_edge["note_id"], v_edge["target_id"], "note") # Wenn in Phase 1 KEINE manuelle Kante mit dieser ID geschrieben wurde, darf die Symmetrie rein if not is_explicit_edge_present(self.client, self.prefix, v_id): final_virtuals.append(v_edge) if final_virtuals: logger.info(f"📤 Schreibe {len(final_virtuals)} validierte Symmetrie-Kanten in den Graphen.") 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 ---") return results async def process_file(self, file_path: str, vault_root: str, **kwargs) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]: """ Transformiert eine Markdown-Datei. Liefert zusätzlich eine Liste von virtuellen Kanten-Kandidaten zurück. """ apply = kwargs.get("apply", False) force_replace = kwargs.get("force_replace", False) purge_before = kwargs.get("purge_before", False) skip_virtuals = kwargs.get("skip_virtuals", False) result = {"path": file_path, "status": "skipped", "changed": False, "error": None} virtual_candidates = [] # 1. Parse & Lifecycle Gate try: 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)}"}, [] 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"}, [] # 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, types_cfg=self.registry) note_id = note_pl["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}:{note_pl.get('hashes', {}).get('hash_source', 'parsed')}:{note_pl.get('hashes', {}).get('hash_normalize', 'canonical')}" # (Hashing Logik hier vereinfacht zur Lesbarkeit, entspricht aber Ihrer Codebasis) 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}, [] # 3. Deep Processing (Chunking, Validation, Embedding) try: body_text = getattr(parsed, "body", "") or "" edge_registry.ensure_latest() 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, 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" ) label = cand.get('target_id') or cand.get('note_id') or "Unknown" logger.info(f" 🧠 [SMART EDGE] {label} -> {'✅ OK' if is_valid else '❌ SKIP'}") if is_valid: new_pool.append(cand) else: new_pool.append(cand) ch.candidate_pool = new_pool 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 [] # Kanten-Extraktion raw_edges = build_edges_for_note(note_id, chunk_pls, note_level_references=note_pl.get("references", [])) # PHASE 1: Authority Edges (Explizit) explicit_edges = [] for e in raw_edges: target_raw = e.get("target_id") target_ctx = self.batch_cache.get(target_raw) target_id = target_ctx.note_id if target_ctx else target_raw # Junk-Filter 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 e.update({ "kind": resolved_kind, "target_id": target_id, "origin_note_id": note_id, "virtual": False, "confidence": 1.0 }) explicit_edges.append(e) # Kandidat für Symmetrie (Phase 2) inv_kind = edge_registry.get_inverse(resolved_kind) if inv_kind and target_id != note_id: v_edge = e.copy() v_edge.update({ "note_id": target_id, "target_id": note_id, "kind": inv_kind, "virtual": True, "provenance": "structure", "confidence": 1.0, "origin_note_id": note_id }) virtual_candidates.append(v_edge) # 4. DB Upsert (Phase 1) if apply: if purge_before and old_payload: purge_artifacts(self.client, self.prefix, note_id) upsert_batch(self.client, f"{self.prefix}_notes", points_for_note(self.prefix, note_pl, None, self.dim)[1]) if chunk_pls and vecs: upsert_batch(self.client, f"{self.prefix}_chunks", points_for_chunks(self.prefix, chunk_pls, vecs)[1]) if explicit_edges: upsert_batch(self.client, f"{self.prefix}_edges", points_for_edges(self.prefix, explicit_edges)[1]) logger.info(f" ✨ Fertig: {len(chunk_pls)} Chunks, {len(explicit_edges)} explizite Kanten geschrieben.") return {"status": "success", "note_id": note_id, "edges_count": len(explicit_edges)}, virtual_candidates except Exception as e: logger.error(f"❌ Fehler bei {file_path}: {e}", exc_info=True) return {**result, "error": str(e)}, [] async def create_from_text(self, markdown_content: str, filename: str, vault_root: str, folder: str = "00_Inbox") -> Dict[str, Any]: """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) await asyncio.sleep(0.1) res, _ = await self.process_file(file_path=target_path, vault_root=vault_root, apply=True, force_replace=True, purge_before=True) return res