""" A0 Software Orchestrator – Pattern-Extraktor Extrahiert Patterns aus abgeschlossenen Projekt-Artefakten und speichert sie in der Patterns-Datenbank. Kontinuierliche Trigger: - error_fixed: known_errors.md wurde aktualisiert - task_completed: Task in task_graph.json als 'done' markiert - deploy_ok: runtime_report.md zeigt Erfolg - decision_made: decisions.md neuer Eintrag - release_done: Release Audit abgeschlossen """ import json import re import hashlib from pathlib import Path from typing import Optional, List, Dict, Any from datetime import datetime from usr.plugins.a0_software_orchestrator.helpers.library.db import PatternDB, get_db # --------------------------------------------------------------------------- # Extraktor-Klasse # --------------------------------------------------------------------------- class PatternExtractor: """ Extrahiert Patterns aus Projekt-Artefakten. Verwendung: extractor = PatternExtractor(project_path) patterns = extractor.extract_from_known_errors() patterns += extractor.extract_from_decisions() patterns += extractor.extract_from_architecture() patterns += extractor.extract_from_docker() patterns += extractor.extract_from_deployment() """ def __init__(self, project_path: Path, db: PatternDB = None): self.project_path = Path(project_path) self.a0_path = self.project_path / ".a0" self.db = db or get_db() self.project_name = self.project_path.name self.project_id = None # Projekt in Registry finden/registrieren self._ensure_project_registered() def _ensure_project_registered(self): """Stellt sicher, dass das Projekt in der Registry ist.""" existing = self.db.conn.execute( "SELECT id FROM projects WHERE project_path = ? OR name = ?", (str(self.project_path), self.project_name) ).fetchone() if existing: self.project_id = existing[0] else: # Tech-Stack aus Projekt-Artefakten erkennen tech_stack = self._detect_tech_stack() self.project_id = self.db.register_project( name=self.project_name, path=str(self.project_path), tech_stack=tech_stack, description=f"Automatisch registriert am {datetime.now().isoformat()}" ) def _detect_tech_stack(self) -> Dict[str, Any]: """Erkennt den Tech-Stack aus Projekt-Dateien.""" stack = {} # Python if (self.project_path / "requirements.txt").exists(): content = (self.project_path / "requirements.txt").read_text() if 'fastapi' in content.lower(): stack['backend'] = 'FastAPI' elif 'flask' in content.lower(): stack['backend'] = 'Flask' elif 'django' in content.lower(): stack['backend'] = 'Django' if 'sqlalchemy' in content.lower(): stack['orm'] = 'SQLAlchemy' if 'pydantic' in content.lower(): stack['validation'] = 'Pydantic' if 'alembic' in content.lower(): stack['migrations'] = 'Alembic' # Node/Frontend if (self.project_path / "package.json").exists(): try: pkg = json.loads((self.project_path / "package.json").read_text()) deps = {**pkg.get('dependencies', {}), **pkg.get('devDependencies', {})} if 'react' in deps: stack['frontend'] = 'React' if 'next' in deps: stack['frontend'] = 'Next.js' if 'vue' in deps: stack['frontend'] = 'Vue' if 'typescript' in deps: stack['language'] = 'TypeScript' if 'vite' in deps: stack['bundler'] = 'Vite' except (json.JSONDecodeError, KeyError): pass # Docker if (self.project_path / "Dockerfile").exists(): stack['docker'] = True if (self.project_path / "docker-compose.yml").exists(): stack['docker_compose'] = True # Datenbank aus env.md oder config env_md = self.a0_path / "env.md" if env_md.exists(): content = env_md.read_text().lower() if 'postgresql' in content or 'postgres' in content: stack['db'] = 'PostgreSQL' elif 'mysql' in content: stack['db'] = 'MySQL' elif 'sqlite' in content: stack['db'] = 'SQLite' return stack # ----------------------------------------------------------------------- # Extraktoren für verschiedene Artefakte # ----------------------------------------------------------------------- def extract_from_known_errors(self) -> List[int]: """ Extrahiert error_solution-Patterns aus known_errors.md. Nur Einträge, die eine Lösung enthalten (nicht nur Beschreibung). """ errors_file = self.a0_path / "known_errors.md" if not errors_file.exists(): return [] content = errors_file.read_text(encoding='utf-8') content_hash = hashlib.md5(content.encode()).hexdigest() # Prüfen, ob diese Datei bereits extrahiert wurde already = self.db.conn.execute( "SELECT id FROM learn_log WHERE source_file = ? AND source_content_hash = ?", (str(errors_file), content_hash) ).fetchone() if already: return [] # Keine Änderungen seit letzter Extraktion pattern_ids = [] # Einfache Heuristik: Nach "## Error:" oder "### Lösung:" Blöcken suchen # Block-Split an Doppel-Newlines blocks = content.split('\n\n') for block in blocks: block = block.strip() if not block or len(block) < 20: continue # Nur Blöcke mit Lösung extrahieren has_solution = any(kw in block.lower() for kw in ['lösung', 'solution', 'fix', 'behoben', 'resolved']) if not has_solution: continue # Titel aus erster Zeile lines = block.split('\n') title = lines[0].lstrip('#').strip()[:100] if not title: title = block[:80] + '...' if len(block) > 80 else block # Pattern speichern pid = self.db.add_pattern( title=title, category='error_fix', pattern_type='error_solution', description=block[:500], code_example=self._extract_code_block(block), source_project_id=self.project_id, source_file=str(errors_file), source_error=title, validated=0 # Automatisch extrahiert → ungeprüft ) # Embedding speichern (für Vektor-Suche) self.db.store_embedding(pid, block[:1000]) # Lern-Log self.db.log_learning(pid, self.project_id, 'error_fixed', str(errors_file), content) pattern_ids.append(pid) return pattern_ids def extract_from_decisions(self) -> List[int]: """ Extrahiert architecture_decision-Patterns aus decisions.md. """ decisions_file = self.a0_path / "decisions.md" if not decisions_file.exists(): return [] content = decisions_file.read_text(encoding='utf-8') content_hash = hashlib.md5(content.encode()).hexdigest() already = self.db.conn.execute( "SELECT id FROM learn_log WHERE source_file = ? AND source_content_hash = ?", (str(decisions_file), content_hash) ).fetchone() if already: return [] pattern_ids = [] # ADR-Blöcke: ## ADR-001: Titel adr_blocks = re.split(r'\n## ADR-\d+:', content)[1:] # Skip header for i, block in enumerate(adr_blocks): lines = block.strip().split('\n') title = lines[0].strip() if lines else f"ADR-{i+1}" # Entscheidung + Begründung extrahieren decision = "" rationale = "" for line in lines: if 'entscheidung' in line.lower() or 'decision' in line.lower(): decision = line.split(':', 1)[-1].strip() if ':' in line else line if 'begründung' in line.lower() or 'rationale' in line.lower(): rationale = line.split(':', 1)[-1].strip() if ':' in line else line description = f"Entscheidung: {decision}\nBegründung: {rationale}" if decision else block[:300] pid = self.db.add_pattern( title=title[:100], category='architecture', pattern_type='architecture_decision', description=description, why_it_works=rationale[:500] if rationale else None, source_project_id=self.project_id, source_file=str(decisions_file), validated=0 ) self.db.store_embedding(pid, description) self.db.log_learning(pid, self.project_id, 'decision_made', str(decisions_file), content) pattern_ids.append(pid) return pattern_ids def extract_from_architecture(self) -> List[int]: """ Extrahiert Patterns aus architecture.md. Erkennt: Tech-Stack, Architekturmuster, Docker-Setup. """ arch_file = self.a0_path / "architecture.md" if not arch_file.exists(): return [] content = arch_file.read_text(encoding='utf-8') content_hash = hashlib.md5(content.encode()).hexdigest() already = self.db.conn.execute( "SELECT id FROM learn_log WHERE source_file = ? AND source_content_hash = ?", (str(arch_file), content_hash) ).fetchone() if already: return [] pattern_ids = [] # Nach bekannten Architekturmustern suchen patterns_to_detect = [ ('FastAPI', 'python', 'fastapi', 'best_practice'), ('SQLAlchemy', 'database', 'sqlalchemy', 'best_practice'), ('React', 'frontend', 'react', 'best_practice'), ('Docker', 'docker', None, 'setup_guide'), ('JWT', 'security', None, 'best_practice'), ('REST API', 'architecture', None, 'architecture_decision'), ] for tech, category, subcat, ptype in patterns_to_detect: if tech.lower() in content.lower(): # Kontext um das Keyword extrahieren idx = content.lower().find(tech.lower()) start = max(0, idx - 100) end = min(len(content), idx + 300) context = content[start:end].strip() pid = self.db.add_pattern( title=f"{tech} in {self.project_name}", category=category, subcategory=subcat, pattern_type=ptype, description=context, source_project_id=self.project_id, source_file=str(arch_file), validated=0 ) self.db.store_embedding(pid, context) self.db.log_learning(pid, self.project_id, 'release_audit', str(arch_file), content) pattern_ids.append(pid) return pattern_ids def extract_from_docker(self) -> List[int]: """ Extrahiert Docker-Patterns aus Dockerfile und docker-compose.yml. """ pattern_ids = [] dockerfile = self.project_path / "Dockerfile" compose_file = self.project_path / "docker-compose.yml" if dockerfile.exists(): content = dockerfile.read_text() content_hash = hashlib.md5(content.encode()).hexdigest() already = self.db.conn.execute( "SELECT id FROM learn_log WHERE source_file = ? AND source_content_hash = ?", (str(dockerfile), content_hash) ).fetchone() if not already: pid = self.db.add_pattern( title=f"Dockerfile aus {self.project_name}", category='docker', pattern_type='setup_guide', description=f"Docker-Konfiguration für {self.project_name}", code_example=content, source_project_id=self.project_id, source_file=str(dockerfile), validated=0 ) self.db.store_embedding(pid, content[:1000]) self.db.log_learning(pid, self.project_id, 'deployment_success', str(dockerfile), content) pattern_ids.append(pid) if compose_file.exists(): content = compose_file.read_text() content_hash = hashlib.md5(content.encode()).hexdigest() already = self.db.conn.execute( "SELECT id FROM learn_log WHERE source_file = ? AND source_content_hash = ?", (str(compose_file), content_hash) ).fetchone() if not already: pid = self.db.add_pattern( title=f"Docker Compose aus {self.project_name}", category='docker', pattern_type='deployment_config', description=f"Docker-Compose-Konfiguration für {self.project_name}", code_example=content, source_project_id=self.project_id, source_file=str(compose_file), validated=0 ) self.db.store_embedding(pid, content[:1000]) self.db.log_learning(pid, self.project_id, 'deployment_success', str(compose_file), content) pattern_ids.append(pid) return pattern_ids def extract_from_deployment(self) -> List[int]: """ Extrahiert Deployment-Patterns aus runtime_report.md und env.md. """ pattern_ids = [] for filename in ['runtime_report.md', 'env.md']: filepath = self.a0_path / filename if not filepath.exists(): continue content = filepath.read_text() content_hash = hashlib.md5(content.encode()).hexdigest() already = self.db.conn.execute( "SELECT id FROM learn_log WHERE source_file = ? AND source_content_hash = ?", (str(filepath), content_hash) ).fetchone() if not already: pid = self.db.add_pattern( title=f"{filename.replace('.md','')} aus {self.project_name}", category='deployment', pattern_type='setup_guide', description=content[:500], source_project_id=self.project_id, source_file=str(filepath), validated=0 ) self.db.store_embedding(pid, content[:1000]) self.db.log_learning(pid, self.project_id, 'deployment_success', str(filepath), content) pattern_ids.append(pid) return pattern_ids def extract_all(self) -> Dict[str, List[int]]: """ Führt alle Extraktoren aus und gibt Summary zurück. """ results = { 'errors': self.extract_from_known_errors(), 'decisions': self.extract_from_decisions(), 'architecture': self.extract_from_architecture(), 'docker': self.extract_from_docker(), 'deployment': self.extract_from_deployment(), } # Projekt-Metriken updaten total = sum(len(v) for v in results.values()) if total > 0: self.db.conn.execute(""" UPDATE projects SET patterns_extracted = COALESCE(patterns_extracted, 0) + ?, last_extraction_at = datetime('now') WHERE id = ? """, (total, self.project_id)) self.db.conn.commit() return results @staticmethod def _extract_code_block(text: str) -> Optional[str]: """Extrahiert einen Code-Block (```...```) aus Text.""" match = re.search(r'```[\s\S]*?```', text) if match: code = match.group(0) # Markdown-Fences entfernen code = re.sub(r'^```\w*\n?', '', code) code = re.sub(r'\n?```$', '', code) return code.strip() return None # --------------------------------------------------------------------------- # Convenience-Funktion # --------------------------------------------------------------------------- def extract_from_project(project_path: str) -> Dict[str, Any]: """ Extrahiert Patterns aus einem Projekt und gibt eine Zusammenfassung zurück. Args: project_path: Pfad zum Projekt-Root Returns: Dict mit Summary der Extraktion """ extractor = PatternExtractor(Path(project_path)) results = extractor.extract_all() total = sum(len(v) for v in results.values()) return { 'project': extractor.project_name, 'project_id': extractor.project_id, 'total_patterns_extracted': total, 'by_category': results, 'timestamp': datetime.now().isoformat() }