311 lines
15 KiB
Python
311 lines
15 KiB
Python
"""
|
|
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.1: Strikte Trennung von Explicit-Write (Phase 1) und
|
|
Symmetry-Validation (Phase 2). 100% Datenhoheit für den Nutzer.
|
|
VERSION: 3.3.1 (WP-24c: Authority-First Ingestion)
|
|
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 UUID-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] = {} # Globaler Kontext-Cache (Pass 1)
|
|
|
|
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 das Anlegen von Kanten zu reinen System-Platzhaltern.
|
|
"""
|
|
if not text or len(text.strip()) < 2:
|
|
return False
|
|
|
|
# Blacklist für Begriffe, die keine echten Notizen sind
|
|
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: Two-Pass Ingestion Workflow mit 2-Phasen-Schreibstrategie.
|
|
"""
|
|
self.batch_cache.clear()
|
|
logger.info(f"--- 🔍 START BATCH IMPORT ({len(file_paths)} Dateien) ---")
|
|
|
|
# SCHRITT 1: Pre-Scan (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 fehlgeschlagen für {path}: {e}")
|
|
|
|
# SCHRITT 2: PHASE 1 (Authority-Schreiben)
|
|
# Wir verarbeiten alle Dateien und schreiben NUR explizite Kanten in die DB.
|
|
results = []
|
|
all_virtual_candidates = []
|
|
|
|
for p in file_paths:
|
|
# process_file liefert in dieser Version (res, virtual_candidates) zurück
|
|
res, candidates = await self.process_file(p, vault_root, apply=True, purge_before=True)
|
|
results.append(res)
|
|
all_virtual_candidates.extend(candidates)
|
|
|
|
# SCHRITT 3: PHASE 2 (Symmetrie-Ergänzung)
|
|
# Nachdem alle expliziten Kanten fest in Qdrant liegen, prüfen wir die Inversen.
|
|
if all_virtual_candidates:
|
|
logger.info(f"🔄 PHASE 2: Validiere {len(all_virtual_candidates)} Symmetrie-Kandidaten gegen Live-DB...")
|
|
final_virtuals = []
|
|
for v_edge in all_virtual_candidates:
|
|
# Eindeutige ID berechnen (muss exakt der ID in Phase 1 entsprechen)
|
|
v_id = _mk_edge_id(v_edge["kind"], v_edge["note_id"], v_edge["target_id"], "note")
|
|
|
|
# Check: Liegt dort bereits eine manuelle Kante?
|
|
if not is_explicit_edge_present(self.client, self.prefix, v_id):
|
|
final_virtuals.append(v_edge)
|
|
else:
|
|
logger.debug(f" 🛡️ Symmetrie übersprungen (Manuelle Kante hat Vorrang): {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 ---")
|
|
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.
|
|
Schreibt Notes/Chunks/Explicit Edges sofort (Phase 1).
|
|
Gibt potenzielle Symmetrien für Phase 2 zurück.
|
|
"""
|
|
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}
|
|
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,
|
|
hash_source=hash_source, hash_normalize=hash_normalize,
|
|
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}:{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()
|
|
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"
|
|
)
|
|
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)
|
|
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 []
|
|
|
|
# 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
|
|
)
|
|
|
|
# PHASE 1: Authority-Check & Kanonisierung
|
|
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
|
|
|
|
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 explizite Kante für Phase 1
|
|
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 vorbereiten
|
|
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" ✨ 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)
|
|
}, 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 |