""" A0 Software Orchestrator – Patterns-Bibliothek Zentrale Datenbank-Klasse mit FTS5, Vektor-Suche, Projekt-Registry, Pattern-Feedback, Konflikt-Management und Aging. Verwendung: from library.db import PatternDB db = PatternDB() # Singleton, verwendet Standard-Pfad results = db.search_fts("FastAPI Docker") patterns = db.search_semantic(query_text, top_k=5) projects = db.get_active_projects() """ import sqlite3 import json import hashlib import threading from pathlib import Path from datetime import datetime, timedelta from typing import Optional, List, Dict, Any, Tuple # --------------------------------------------------------------------------- # Singleton PatternDB # --------------------------------------------------------------------------- class PatternDB: """ Singleton-Datenbank-Klasse für die Patterns-Bibliothek. Automatische Initialisierung beim ersten Zugriff. """ _instance = None _lock = threading.Lock() DEFAULT_DB_PATH = Path(__file__).parent / "patterns.db" DEFAULT_SCHEMA_PATH = Path(__file__).parent / "schema.sql" def __new__(cls, db_path: Optional[Path] = None): if cls._instance is None: with cls._lock: if cls._instance is None: instance = super().__new__(cls) instance._initialized = False cls._instance = instance return cls._instance def __init__(self, db_path: Optional[Path] = None): if self._initialized: return self.db_path = Path(db_path) if db_path else self.DEFAULT_DB_PATH self.schema_path = self.DEFAULT_SCHEMA_PATH self._conn: Optional[sqlite3.Connection] = None self._vec_available: Optional[bool] = None self._embedding_model = None # Datenbank initialisieren self._ensure_db() self._initialized = True # ----------------------------------------------------------------------- # Connection Management # ----------------------------------------------------------------------- @property def conn(self) -> sqlite3.Connection: """Thread-sichere Connection mit WAL-Mode.""" if self._conn is None: self._conn = sqlite3.connect(str(self.db_path), check_same_thread=False) self._conn.row_factory = sqlite3.Row self._conn.execute("PRAGMA journal_mode=WAL") self._conn.execute("PRAGMA foreign_keys=ON") self._conn.execute("PRAGMA cache_size=-64000") # 64 MB Cache self._conn.execute("PRAGMA busy_timeout=5000") # 5 Sekunden Timeout return self._conn def _ensure_db(self): """Stellt sicher, dass die Datenbank existiert und das Schema aktuell ist.""" self.db_path.parent.mkdir(parents=True, exist_ok=True) if not self.db_path.exists(): # Neue Datenbank: Basisschema ausführen if self.schema_path.exists(): schema = self.schema_path.read_text(encoding='utf-8') self.conn.executescript(schema) self.conn.commit() # Neue und existierende DBs immer auf Runtime-Schema migrieren. self._run_migrations() def _run_migrations(self): """Führt ausstehende Schema-Migrationen aus.""" cursor = self.conn.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name='schema_version'" ) if cursor.fetchone() is None: # Alte DB ohne schema_version – initialisieren self.conn.executescript(self.schema_path.read_text(encoding='utf-8')) self.conn.commit() # Weitere Migrationen aus migrations.py try: from .migrations import run_migrations run_migrations(self) except ImportError: pass def close(self): """Schließt die Datenbank-Verbindung.""" if self._conn: self._conn.close() self._conn = None # ----------------------------------------------------------------------- # Projekt-Registry # ----------------------------------------------------------------------- def register_project(self, name: str, path: str, tech_stack: Optional[Dict] = None, description: str = "", git_url: str = "") -> int: """Registriert ein neues Projekt oder aktualisiert ein bestehendes. ARCHITEKTUR-NOTE (Bugfix-Auto-Registration §3.6 / Fix 4 Befund): Diese Methode gehört zum DB-Layer (helpers/library/db.py), NICHT zum Project-Layer (helpers/db_state_store.py). Sie wird derzeit NUR von extractor.py (Pattern-Extraktion aus Repo-Verzeichnissen) aufgerufen, mit Namen aus `project_path` (Verzeichnisname). User-facing Project- Registrierung läuft über `db_state_store.register_project()`, das seit Plan v3 §3.1 Pattern+Blacklist validiert. Daher: KEINE Plausi-Prüfung hier. Bewusst out-of-scope, weil: 1. Einziger Caller ist ein internes Library-Tool (extractor.py). 2. extractor.py nutzt Verzeichnisnamen, die bereits durch project_path-Lookup semi-kontrolliert sind. 3. Plan v3 hat DB-Layer-Plausi explizit als separater Fix markiert. Wenn ein neuer Caller mit user-input-Namen diese Methode aufruft, MUSS er Pattern+Blacklist-Prüfung VORAB durchführen (siehe `db_state_store._validate_project_name`). """ tech_json = json.dumps(tech_stack) if tech_stack is not None else "{}" cur = self.conn.execute(""" INSERT INTO projects (name, project_path, git_url, tech_stack, description) VALUES (?, ?, ?, ?, ?) ON CONFLICT(name) DO UPDATE SET project_path = excluded.project_path, git_url = excluded.git_url, tech_stack = excluded.tech_stack, description = excluded.description """, (name, path, git_url, tech_json, description)) row = self.conn.execute("SELECT id FROM projects WHERE name = ?", (name,)).fetchone() project_id = int(row[0] if row else cur.lastrowid) self.conn.execute(""" INSERT INTO project_state (project_id, status, phase, last_active_at) VALUES (?, 'active', 'intake', datetime('now')) ON CONFLICT(project_id) DO UPDATE SET last_active_at = datetime('now') """, (project_id,)) self.conn.commit() return project_id def update_project_phase(self, name: str, phase: str, plan_mode: str = None): """Aktualisiert Phase und Plan-Mode eines Projekts.""" project_id = self.register_project(name, "", description="auto-created by PatternDB") if plan_mode: self.conn.execute(""" UPDATE project_state SET phase = ?, plan_mode = ?, last_active_at = datetime('now'), updated_at = datetime('now') WHERE project_id = ? """, (phase, plan_mode, project_id)) else: self.conn.execute(""" UPDATE project_state SET phase = ?, last_active_at = datetime('now'), updated_at = datetime('now') WHERE project_id = ? """, (phase, project_id)) self.conn.commit() def update_project_metrics(self, name: str, total_tasks: int = None, completed_tasks: int = None, open_errors: int = None): """Aktualisiert die Projekt-Metriken.""" project_id = self.register_project(name, "", description="auto-created by PatternDB") updates = [] params: List[Any] = [] if total_tasks is not None: updates.append("total_tasks = ?"); params.append(total_tasks) if completed_tasks is not None: updates.append("completed_tasks = ?"); params.append(completed_tasks) if open_errors is not None: updates.append("open_errors = ?"); params.append(open_errors) if updates: updates.append("last_active_at = datetime('now')") updates.append("updated_at = datetime('now')") params.append(project_id) self.conn.execute(f"UPDATE project_state SET {', '.join(updates)} WHERE project_id = ?", params) self.conn.commit() def get_active_projects(self) -> List[sqlite3.Row]: """Alle aktiven Projekte.""" return self.conn.execute(""" SELECT p.*, ps.status, ps.phase, ps.plan_mode, ps.total_tasks, ps.completed_tasks, ps.open_errors, ps.last_active_at, ps.completed_at FROM projects p JOIN project_state ps ON ps.project_id = p.id WHERE ps.status = 'active' ORDER BY COALESCE(ps.last_active_at, p.created_at) DESC """).fetchall() def get_project_summary(self, project_name: str) -> Optional[sqlite3.Row]: """Kurzübersicht eines Projekts.""" return self.conn.execute(""" SELECT p.name AS project_name, ps.status, ps.phase, ps.plan_mode, ps.completed_tasks || '/' || ps.total_tasks AS progress, ps.open_errors, COALESCE(p.patterns_extracted, 0) AS patterns_extracted, p.tech_stack, ps.last_active_at, p.description FROM projects p JOIN project_state ps ON ps.project_id = p.id WHERE p.name = ? """, (project_name,)).fetchone() def get_projects_by_tech(self, tech: str) -> List[sqlite3.Row]: """Alle Projekte mit bestimmter Technologie.""" return self.conn.execute(""" SELECT p.*, ps.status, ps.phase, ps.plan_mode FROM projects p JOIN project_state ps ON ps.project_id = p.id WHERE p.tech_stack LIKE ? ORDER BY p.name """, (f'%{tech}%',)).fetchall() def get_orphaned_projects(self, days: int = 7) -> List[sqlite3.Row]: """Projekte, die >N Tage nicht aktiv waren.""" threshold = (datetime.utcnow() - timedelta(days=days)).isoformat() return self.conn.execute(""" SELECT p.*, ps.status, ps.phase, ps.plan_mode, ps.last_active_at FROM projects p JOIN project_state ps ON ps.project_id = p.id WHERE ps.status = 'active' AND COALESCE(ps.last_active_at, p.created_at) < ? """, (threshold,)).fetchall() def set_project_status(self, name: str, status: str, notes: str = ""): """Setzt den Projekt-Status (active, paused, completed, archived, failed).""" project_id = self.register_project(name, "", description=notes or "auto-created by PatternDB") if status == 'completed': self.conn.execute(""" UPDATE project_state SET status = ?, completed_at = datetime('now'), updated_at = datetime('now') WHERE project_id = ? """, (status, project_id)) else: self.conn.execute(""" UPDATE project_state SET status = ?, last_active_at = datetime('now'), updated_at = datetime('now') WHERE project_id = ? """, (status, project_id)) if notes: self.conn.execute("UPDATE projects SET description = ? WHERE id = ?", (notes, project_id)) self.conn.commit() # ----------------------------------------------------------------------- # Patterns CRUD # ----------------------------------------------------------------------- def add_pattern(self, title: str, category: str, pattern_type: str, description: str, **kwargs) -> int: """ Fügt ein neues Pattern hinzu. Args: title: Kurztitel category: docker, python, frontend, etc. pattern_type: code_snippet, error_solution, etc. description: Beschreibung **kwargs: subcategory, code_example, when_to_use, why_it_works, pitfalls, source_project_id, source_task_id, source_error, source_file, framework_version, tags (Liste), validated (0=auto, 1=reviewed, 2=manual) Returns: int: ID des neuen Patterns """ tags = kwargs.pop('tags', []) columns = ['title', 'category', 'pattern_type', 'description'] values = [title, category, pattern_type, description] allowed_kwargs = [ 'subcategory', 'code_example', 'when_to_use', 'why_it_works', 'pitfalls', 'source_project_id', 'source_task_id', 'source_error', 'source_file', 'framework_version', 'validated' ] for key in allowed_kwargs: if key in kwargs and kwargs[key] is not None: columns.append(key) values.append(kwargs[key]) placeholders = ', '.join(['?'] * len(columns)) columns_str = ', '.join(columns) cursor = self.conn.execute( f"INSERT INTO patterns ({columns_str}) VALUES ({placeholders})", values ) pattern_id = cursor.lastrowid # Tags hinzufügen for tag_name in tags: self._add_tag(pattern_id, tag_name) self.conn.commit() return pattern_id def update_pattern(self, pattern_id: int, **kwargs) -> bool: """Aktualisiert ein bestehendes Pattern.""" allowed = [ 'title', 'category', 'subcategory', 'pattern_type', 'description', 'code_example', 'when_to_use', 'why_it_works', 'pitfalls', 'framework_version', 'validated', 'validation_date', 'validated_by' ] updates = [] params: List[Any] = [] for key in allowed: if key in kwargs: updates.append(f"{key} = ?") params.append(kwargs[key]) if not updates: return False updates.append("updated_at = datetime('now')") params.append(pattern_id) self.conn.execute( f"UPDATE patterns SET {', '.join(updates)} WHERE id = ?", params ) self.conn.commit() # Tags aktualisieren if 'tags' in kwargs: self.conn.execute("DELETE FROM pattern_tags WHERE pattern_id = ?", (pattern_id,)) for tag_name in kwargs['tags']: self._add_tag(pattern_id, tag_name) return True def get_pattern(self, pattern_id: int) -> Optional[sqlite3.Row]: """Lädt ein einzelnes Pattern mit allen Details.""" return self.conn.execute( "SELECT * FROM patterns WHERE id = ?", (pattern_id,) ).fetchone() def get_patterns_by_category(self, category: str, pattern_type: str = None, validated_only: bool = True, limit: int = 20) -> List[sqlite3.Row]: """Patterns nach Kategorie filtern.""" sql = "SELECT * FROM patterns WHERE category = ?" params: List[Any] = [category] if pattern_type: sql += " AND pattern_type = ?" params.append(pattern_type) if validated_only: sql += " AND validated >= 0" sql += " ORDER BY usage_count DESC, success_rate DESC LIMIT ?" params.append(limit) return self.conn.execute(sql, params).fetchall() def get_pattern_tags(self, pattern_id: int) -> List[str]: """Alle Tags eines Patterns.""" rows = self.conn.execute(""" SELECT t.name FROM tags t JOIN pattern_tags pt ON t.id = pt.tag_id WHERE pt.pattern_id = ? """, (pattern_id,)).fetchall() return [r[0] for r in rows] def _add_tag(self, pattern_id: int, tag_name: str): """Fügt einen Tag hinzu und verknüpft ihn mit einem Pattern.""" self.conn.execute("INSERT OR IGNORE INTO tags (name) VALUES (?)", (tag_name,)) tag_id = self.conn.execute( "SELECT id FROM tags WHERE name = ?", (tag_name,) ).fetchone()[0] self.conn.execute( "INSERT OR IGNORE INTO pattern_tags (pattern_id, tag_id) VALUES (?, ?)", (pattern_id, tag_id) ) # ----------------------------------------------------------------------- # FTS5 Volltextsuche # ----------------------------------------------------------------------- def search_fts(self, query: str, limit: int = 10, category: str = None, pattern_type: str = None, validated_only: bool = True) -> List[sqlite3.Row]: """ Volltextsuche mit FTS5. Args: query: Suchbegriffe (FTS5-Syntax: "FastAPI Docker", "error AND fix", etc.) limit: Maximale Ergebnisse category: Optional, nach Kategorie filtern pattern_type: Optional, nach Typ filtern validated_only: Nur validierte Patterns (>= 0) Returns: Liste von Pattern-Rows mit rank-Spalte """ # FTS5-Abfrage vorbereiten (Wildcards für Teilwortsuche) fts_query = ' OR '.join(f'"{term}"*' for term in query.split()) sql = """ SELECT p.*, rank FROM patterns_fts JOIN patterns p ON patterns_fts.rowid = p.id WHERE patterns_fts MATCH ? """ params: List[Any] = [fts_query] if category: sql += " AND p.category = ?" params.append(category) if pattern_type: sql += " AND p.pattern_type = ?" params.append(pattern_type) if validated_only: sql += " AND p.validated >= 0" sql += " ORDER BY rank LIMIT ?" params.append(limit) return self.conn.execute(sql, params).fetchall() # ----------------------------------------------------------------------- # Vektor-Suche (sqlite-vec) # ----------------------------------------------------------------------- @property def vec_available(self) -> bool: """Prüft, ob sqlite-vec verfügbar ist.""" if self._vec_available is None: try: import sqlite_vec self._vec_available = True self._init_vec_table() except ImportError: self._vec_available = False return self._vec_available def _init_vec_table(self): """Initialisiert die Vektor-Tabelle, wenn sqlite-vec vorhanden ist.""" try: import sqlite_vec self.conn.enable_load_extension(True) sqlite_vec.load(self.conn) self.conn.execute(""" CREATE VIRTUAL TABLE IF NOT EXISTS pattern_embeddings USING vec0( embedding float[384] ) """) self.conn.commit() except Exception as e: print(f"[PatternDB] sqlite-vec konnte nicht initialisiert werden: {e}") self._vec_available = False def _get_embedding_model(self): """Lädt das Embedding-Modell (lazy, nur wenn benötigt).""" if self._embedding_model is None: try: from sentence_transformers import SentenceTransformer self._embedding_model = SentenceTransformer('all-MiniLM-L6-v2') except ImportError: print("[PatternDB] sentence-transformers nicht installiert. " "Vektor-Suche nicht verfügbar.") self._vec_available = False return self._embedding_model def store_embedding(self, pattern_id: int, text: str): """ Speichert das Embedding eines Patterns für die Vektor-Suche. Args: pattern_id: ID des Patterns text: Text zum Embedden (description + code_example) """ if not self.vec_available: return model = self._get_embedding_model() if model is None: return embedding = model.encode(text) # Bestehendes Embedding löschen (vec0 hat keine UPDATE-Logik) self.conn.execute( "DELETE FROM pattern_embeddings WHERE rowid = ?", (pattern_id,) ) self.conn.execute( "INSERT INTO pattern_embeddings (rowid, embedding) VALUES (?, ?)", (pattern_id, embedding.tobytes()) ) self.conn.commit() def search_semantic(self, query: str, top_k: int = 10, category: str = None, pattern_type: str = None, validated_only: bool = True) -> List[Tuple[sqlite3.Row, float]]: """ Semantische Vektor-Suche. Fallback auf FTS5, wenn sqlite-vec nicht verfügbar. Args: query: Natürlichsprachliche Suchanfrage top_k: Anzahl Ergebnisse category: Optional, Kategorie-Filter pattern_type: Optional, Typ-Filter validated_only: Nur validierte Patterns Returns: Liste von (Pattern-Row, similarity_score) Tupeln, absteigend nach Ähnlichkeit """ if not self.vec_available: # Fallback auf FTS5 rows = self.search_fts(query, limit=top_k, category=category, pattern_type=pattern_type, validated_only=validated_only) return [(r, -r['rank']) for r in rows] # rank ist negativ, invertieren model = self._get_embedding_model() if model is None: rows = self.search_fts(query, limit=top_k, category=category, pattern_type=pattern_type, validated_only=validated_only) return [(r, -r['rank']) for r in rows] # Query embedden query_embedding = model.encode(query) # KNN-Suche categories_where = "" if category: categories_where = f"AND p.category = '{category}'" if pattern_type: categories_where += f" AND p.pattern_type = '{pattern_type}'" if validated_only: categories_where += " AND p.validated >= 0" sql = f""" SELECT p.*, vec_distance_cosine(pe.embedding, ?) AS distance FROM pattern_embeddings pe JOIN patterns p ON pe.rowid = p.id WHERE 1=1 {categories_where} ORDER BY distance ASC LIMIT ? """ rows = self.conn.execute(sql, (query_embedding.tobytes(), top_k)).fetchall() # Distance in Similarity umrechnen (1 - distance) return [(r, 1.0 - r['distance']) for r in rows] # ----------------------------------------------------------------------- # Pattern-Feedback & Lernen # ----------------------------------------------------------------------- def add_feedback(self, pattern_id: int, project_id: int, outcome: str, reason: str = "", context_diff: str = "", alternative_used: str = "", alternative_pattern_id: int = None) -> int: """ Speichert Feedback zu einem Pattern (positiv oder negativ). Aktualisiert automatisch die Pattern-Statistiken. """ feedback_id = self.conn.execute(""" INSERT INTO pattern_feedback (pattern_id, project_id, outcome, reason, context_diff, alternative_used, alternative_pattern_id) VALUES (?, ?, ?, ?, ?, ?, ?) """, (pattern_id, project_id, outcome, reason, context_diff, alternative_used, alternative_pattern_id)).lastrowid # Pattern-Statistiken aktualisieren if outcome == 'success': self.conn.execute( "UPDATE patterns SET success_count = success_count + 1, usage_count = usage_count + 1, last_used_at = datetime('now') WHERE id = ?", (pattern_id,) ) elif outcome == 'failure': self.conn.execute( "UPDATE patterns SET failure_count = failure_count + 1, usage_count = usage_count + 1, last_used_at = datetime('now') WHERE id = ?", (pattern_id,) ) elif outcome == 'partial_success': self.conn.execute( "UPDATE patterns SET success_count = success_count + 1, failure_count = failure_count + 1, usage_count = usage_count + 1, last_used_at = datetime('now') WHERE id = ?", (pattern_id,) ) # Success-Rate neu berechnen self.conn.execute(""" UPDATE patterns SET success_rate = CASE WHEN success_count + failure_count > 0 THEN CAST(success_count AS REAL) / (success_count + failure_count) ELSE 0.0 END WHERE id = ? """, (pattern_id,)) self.conn.commit() return feedback_id def record_usage(self, pattern_id: int): """Vermerkt, dass ein Pattern verwendet wurde (ohne Erfolg/Misserfolg).""" self.conn.execute( "UPDATE patterns SET usage_count = usage_count + 1, last_used_at = datetime('now') WHERE id = ?", (pattern_id,) ) self.conn.commit() # ----------------------------------------------------------------------- # Pattern-Konflikte # ----------------------------------------------------------------------- def add_conflict(self, pattern_a_id: int, pattern_b_id: int, conflict_type: str, description: str = "", resolution_context: str = "") -> int: """Registriert einen Konflikt zwischen zwei Patterns.""" return self.conn.execute(""" INSERT OR IGNORE INTO pattern_conflicts (pattern_a_id, pattern_b_id, conflict_type, description, resolution_context) VALUES (?, ?, ?, ?, ?) """, (pattern_a_id, pattern_b_id, conflict_type, description, resolution_context)).lastrowid def get_conflicts(self) -> List[sqlite3.Row]: """Alle ungelösten Konflikte.""" return self.conn.execute(""" SELECT * FROM v_conflicts WHERE resolved_by = 'system' """).fetchall() def resolve_conflict(self, conflict_id: int, resolution: str, resolved_by: str = "quality_reviewer"): """Löst einen Pattern-Konflikt auf.""" self.conn.execute(""" UPDATE pattern_conflicts SET resolution_context = ?, resolved_by = ?, resolved_at = datetime('now') WHERE id = ? """, (resolution, resolved_by, conflict_id)) self.conn.commit() # ----------------------------------------------------------------------- # Pattern-Aging # ----------------------------------------------------------------------- def update_staleness_scores(self): """ Berechnet den staleness_score für alle Patterns neu. Faktoren: - Alter (Tage seit Erstellung): 0-100 → 0.0-0.4 - Letzte Verwendung (Tage): 0-365 → 0.0-0.3 - Letzte Validierung (Tage): 0-365 → 0.0-0.2 - Framework-Version vorhanden? nein → +0.1 """ self.conn.execute(""" UPDATE patterns SET staleness_score = ( MIN(1.0, -- Alter-Faktor: 0.4 nach 365 Tagen (julianday('now') - julianday(created_at)) / 365.0 * 0.4 + -- Letzte-Verwendung-Faktor: 0.3 nach 365 Tagen CASE WHEN last_used_at IS NOT NULL THEN (julianday('now') - julianday(last_used_at)) / 365.0 * 0.3 ELSE 0.3 -- Nie verwendet = voller Abzug END + -- Validierungs-Faktor: 0.2 nach 365 Tagen CASE WHEN validation_date IS NOT NULL THEN (julianday('now') - julianday(validation_date)) / 365.0 * 0.2 ELSE 0.2 -- Nie validiert = voller Abzug END + -- Framework-Version fehlt: +0.1 CASE WHEN framework_version IS NULL OR framework_version = '' THEN 0.1 ELSE 0.0 END ) ) WHERE validated >= 0 """) self.conn.commit() def deprecate_stale_patterns(self, threshold: float = 0.7): """Markiert stark veraltete Patterns als deprecated (validated = -1).""" self.conn.execute(""" UPDATE patterns SET validated = -1, updated_at = datetime('now') WHERE staleness_score >= ? AND validated >= 0 """, (threshold,)) self.conn.commit() def get_stale_patterns(self, threshold: float = 0.5) -> List[sqlite3.Row]: """Patterns, die zu veralten drohen.""" return self.conn.execute(""" SELECT * FROM patterns WHERE staleness_score >= ? AND validated >= 0 ORDER BY staleness_score DESC """, (threshold,)).fetchall() # ----------------------------------------------------------------------- # Relationen # ----------------------------------------------------------------------- def add_relation(self, pattern_a_id: int, pattern_b_id: int, relation_type: str, strength: float = 1.0, notes: str = ""): """Fügt eine Beziehung zwischen zwei Patterns hinzu.""" self.conn.execute(""" INSERT OR IGNORE INTO pattern_relations (pattern_a_id, pattern_b_id, relation_type, strength, notes) VALUES (?, ?, ?, ?, ?) """, (pattern_a_id, pattern_b_id, relation_type, strength, notes)) self.conn.commit() def get_related_patterns(self, pattern_id: int, relation_type: str = None) -> List[sqlite3.Row]: """Verwandte Patterns eines Patterns.""" sql = """ SELECT p.*, pr.relation_type, pr.strength FROM pattern_relations pr JOIN patterns p ON ( CASE WHEN pr.pattern_a_id = ? THEN pr.pattern_b_id = p.id ELSE pr.pattern_a_id = p.id END ) WHERE (pr.pattern_a_id = ? OR pr.pattern_b_id = ?) """ params: List[Any] = [pattern_id, pattern_id, pattern_id] if relation_type: sql += " AND pr.relation_type = ?" params.append(relation_type) return self.conn.execute(sql, params).fetchall() # ----------------------------------------------------------------------- # Lern-Log # ----------------------------------------------------------------------- def log_learning(self, pattern_id: int, project_id: int, trigger: str, source_file: str = "", source_content: str = ""): """Protokolliert einen Lern-Vorgang.""" content_hash = hashlib.md5(source_content.encode()).hexdigest() if source_content else None self.conn.execute(""" INSERT INTO learn_log (pattern_id, project_id, trigger, source_file, source_content_hash) VALUES (?, ?, ?, ?, ?) """, (pattern_id, project_id, trigger, source_file, content_hash)) self.conn.commit() # ----------------------------------------------------------------------- # Backup # ----------------------------------------------------------------------- def backup(self, backup_path: Path = None) -> Path: """Erstellt ein Backup der Datenbank.""" if backup_path is None: timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') backup_path = self.db_path.parent / f"patterns_backup_{timestamp}.db" # SQLite-Backup via Backup-API backup_conn = sqlite3.connect(str(backup_path)) self.conn.backup(backup_conn) backup_conn.close() return backup_path # ----------------------------------------------------------------------- # Statistik & Reporting # ----------------------------------------------------------------------- def get_stats(self) -> Dict[str, Any]: """Gesamtstatistik der Bibliothek.""" total_patterns = self.conn.execute("SELECT COUNT(*) FROM patterns").fetchone()[0] validated = self.conn.execute( "SELECT COUNT(*) FROM patterns WHERE validated >= 0" ).fetchone()[0] deprecated = self.conn.execute( "SELECT COUNT(*) FROM patterns WHERE validated = -1" ).fetchone()[0] unvalidated = self.conn.execute( "SELECT COUNT(*) FROM patterns WHERE validated = 0" ).fetchone()[0] categories = self.conn.execute(""" SELECT category, COUNT(*) as cnt FROM patterns WHERE validated >= 0 GROUP BY category ORDER BY cnt DESC """).fetchall() active_projects = self.conn.execute( "SELECT COUNT(*) FROM project_state WHERE status = 'active'" ).fetchone()[0] total_feedback = self.conn.execute( "SELECT COUNT(*) FROM pattern_feedback" ).fetchone()[0] avg_success_rate = self.conn.execute(""" SELECT AVG(success_rate) FROM patterns WHERE usage_count > 0 AND validated >= 0 """).fetchone()[0] or 0.0 return { 'total_patterns': total_patterns, 'validated_patterns': validated, 'unvalidated_patterns': unvalidated, 'deprecated_patterns': deprecated, 'active_projects': active_projects, 'total_feedback_entries': total_feedback, 'average_success_rate': round(avg_success_rate * 100, 1), 'categories': {r['category']: r['cnt'] for r in categories}, 'vec_available': self.vec_available, } # --------------------------------------------------------------------------- # Convenience-Funktionen # --------------------------------------------------------------------------- def get_db() -> PatternDB: """Singleton-Zugriff.""" return PatternDB()