error_tree.py

# core/error_tree.py
"""
Harbor — Arbre d'erreurs sémantique.
Mémorise les échecs, leurs causes, leçons et correctifs.
"""
import hashlib
import time
import json
import logging
from pathlib import Path
from typing import Dict, List

log = logging.getLogger("vortex.harbor")


class ErrorTree:
    def __init__(self, storage_path: str = "data/error_tree.json"):
        self.storage_path = Path(storage_path)
        self.nodes: Dict[str, Dict] = {}
        self.load()

    def add_error(self, error_msg: str, context: str) -> str:
        node_id = hashlib.sha256(f"{error_msg}{time.time()}".encode()).hexdigest()[:8]
        self.nodes[node_id] = {
            "error": error_msg,
            "context": context[:500],
            "lesson": None,
            "fix": None,
            "timestamp": time.time()
        }
        self.save()
        log.info(f"[Harbor] Erreur enregistrée : {node_id} — {error_msg[:50]}")
        return node_id

    def add_lesson(self, node_id: str, lesson: str, fix: str):
        if node_id in self.nodes:
            self.nodes[node_id]["lesson"] = lesson
            self.nodes[node_id]["fix"] = fix
            self.save()
            log.info(f"[Harbor] Leçon ajoutée à {node_id}")

    def search_similar(self, query: str) -> List[Dict]:
        results = []
        query_words = set(query.lower().split())
        for nid, data in self.nodes.items():
            error_words = set(data["error"].lower().split())
            overlap = len(query_words & error_words)
            if overlap > 0:
                sim = overlap / max(len(query_words), 1)
                results.append({
                    "id": nid,
                    "error": data["error"],
                    "lesson": data.get("lesson") or "Aucune leçon",
                    "fix": data.get("fix") or "Aucun correctif",
                    "similarity": round(sim, 3)
                })
        results.sort(key=lambda x: x["similarity"], reverse=True)
        return results[:5]

    def stats(self) -> Dict:
        with_lessons = sum(1 for n in self.nodes.values() if n.get("lesson"))
        return {
            "total_errors": len(self.nodes),
            "with_lessons": with_lessons
        }

    def save(self):
        try:
            self.storage_path.parent.mkdir(parents=True, exist_ok=True)
            with open(self.storage_path, "w", encoding="utf-8") as f:
                json.dump(self.nodes, f, indent=2, default=str, ensure_ascii=False)
        except Exception as e:
            log.error(f"[Harbor] Échec sauvegarde: {e}")

    def load(self):
        if self.storage_path.exists():
            try:
                with open(self.storage_path, "r", encoding="utf-8") as f:
                    self.nodes = json.load(f)
                log.info(f"[Harbor] Chargé {len(self.nodes)} erreurs")
            except Exception as e:
                log.error(f"[Harbor] Échec chargement: {e}")
                self.nodes = {}
        else:
            self.nodes = {}