entity_extractor.py

Code Hygiene Score: 100

Keine Issues gefunden.

Dependencies 5

Klassen 1

Code

"""Entitäten-Extraktion für Wissensextraktion."""

import sys

sys.path.insert(0, "/var/www/scripts/pipeline")

from db import db

from .llm_service import LLMService
from .models import KnowledgeLevel, KnowledgeType


class EntityExtractor:
    """Extrahiert Entitäten aus Text."""

    def __init__(self, llm_service: LLMService, store_knowledge_fn):
        """Initialisiere mit LLM-Service und Storage-Funktion."""
        self.llm = llm_service
        self.store_knowledge = store_knowledge_fn

    def extract_entities(self, text: str, level: KnowledgeLevel, source_id: int) -> list[dict]:
        """
        Extrahiere Entitäten aus Text.

        Args:
            text: Eingabetext
            level: Ebene (PAGE, SECTION, DOCUMENT)
            source_id: ID der Quelle (page_id, section_id, document_id)

        Returns:
            Liste von Entitäten mit DB-IDs
        """
        prompt = f"""Analysiere den folgenden deutschen Text und extrahiere alle wichtigen Entitäten.

Kategorien:
- PERSON: Namen von Personen, Autoren, Therapeuten
- ORGANIZATION: Firmen, Institute, Verbände
- CONCEPT: Fachbegriffe, Theorien, Modelle
- METHOD: Methoden, Techniken, Verfahren
- TOOL: Werkzeuge, Instrumente, Materialien
- LOCATION: Orte, Länder, Regionen
- EVENT: Ereignisse, Konferenzen

Antworte NUR als JSON:
{{"entities": [
    {{"name": "...", "type": "CONCEPT", "context": "kurzer Kontext", "importance": 0.0-1.0}}
]}}

Text ({level.value}-Ebene):
{text[:4000]}"""

        result = self.llm.call_llm(prompt)
        data = self.llm.parse_json(result)
        entities = data.get("entities", [])

        # Speichere und gleiche mit DB ab
        stored_entities = []
        for entity in entities:
            stored = self._store_entity(entity, level, source_id)
            if stored:
                stored_entities.append(stored)

        # Speichere in Knowledge-Tabelle
        self.store_knowledge(
            level,
            source_id,
            KnowledgeType.ENTITY,
            {"entities": [e["name"] for e in stored_entities], "count": len(stored_entities)},
        )

        return stored_entities

    def _store_entity(self, entity: dict, level: KnowledgeLevel, source_id: int) -> dict | None:
        """Speichere Entität in DB mit Duplikatprüfung."""
        try:
            name = entity.get("name", "").strip()
            entity_type = entity.get("type", "OTHER").upper()
            context = entity.get("context", "")

            if not name:
                return None

            # Prüfe ob Entität existiert (case-insensitive)
            cursor = db.execute(
                """SELECT id, name, type, description
                   FROM entities
                   WHERE LOWER(name) = LOWER(%s) OR LOWER(canonical_name) = LOWER(%s)
                   LIMIT 1""",
                (name, name),
            )
            existing = cursor.fetchone()
            cursor.close()

            if existing:
                entity_id = existing["id"]
                db.log("DEBUG", f"Entität '{name}' existiert bereits (ID: {entity_id})")
            else:
                # Neue Entität anlegen
                cursor = db.execute(
                    """INSERT INTO entities (name, type, canonical_name, created_at)
                       VALUES (%s, %s, %s, NOW())""",
                    (name, entity_type, name.lower()),
                )
                db.commit()
                entity_id = cursor.lastrowid
                cursor.close()
                db.log("INFO", f"Neue Entität angelegt: '{name}' (ID: {entity_id})")

            return {"id": entity_id, "name": name, "type": entity_type, "context": context, "is_new": existing is None}

        except Exception as e:
            db.log("ERROR", f"Fehler beim Speichern der Entität: {e}")
            return None
← Übersicht Graph