Explorer
/opt/struktur/lead-engine/llm/validator.py
← Zurück ↓ Download
"""
Gestaffelte Validierung für Enrichment-Outputs.
CRITICAL + IMPORTANT + OPTIONAL mit Schätzungs-Unterstützung.

Dottores Regeln:
- CRITICAL: alle müssen vorhanden sein
- IMPORTANT: mind. 2/2 (explizit ODER als Schätzung markiert)
- >50% Felder geschätzt → needs_manual
"""

from typing import Dict, Tuple, List


# Feldkategorien
CRITICAL_FIELDS = [
    'pain_point',
    'company_description'
]

IMPORTANT_FIELDS = [
    'company_size',
    'headquarters'
]

OPTIONAL_FIELDS = [
    'founding_year',
    'technology_stack',
    'key_products',
    'linkedin_url',
    'industry',
    'language'
]

ALL_DATA_FIELDS = CRITICAL_FIELDS + IMPORTANT_FIELDS + OPTIONAL_FIELDS

# Erlaubte Werte für company_size
VALID_COMPANY_SIZES = {
    '1-10 mitarbeiter',
    '10-50 mitarbeiter',
    '50-200 mitarbeiter',
    '200+ mitarbeiter',
}


def validate(data: Dict) -> Tuple[bool, str]:
    """
    Validiert einen Enrichment-Output gegen strikte Regeln.

    Regeln:
    - ALLE CRITICAL müssen vorhanden und nicht leer sein
    - Mindestens 2/2 IMPORTANT: Wert vorhanden UND (explizit ODER estimated=true)
    - company_size: nur erlaubte Kategorien
    - >50% Felder geschätzt → FAIL (needs_manual)

    Returns:
        (True, "OK") oder (False, error_reason)
    """
    if not isinstance(data, dict):
        return False, "INVALID_JSON_TYPE"

    meta = data.get('meta', {}) or {}

    # 1. CRITICAL Felder
    for field in CRITICAL_FIELDS:
        if field not in data or is_empty_value(data[field]):
            return False, f"MISSING_CRITICAL: {field}"

    # 2. IMPORTANT Felder (explizit ODER geschätzt)
    valid_important = 0
    for field in IMPORTANT_FIELDS:
        value = data.get(field)
        if is_empty_value(value):
            continue

        # company_size: muss erlaubter Wert sein
        if field == 'company_size':
            if str(value).strip().lower() not in VALID_COMPANY_SIZES:
                continue

        # headquarters: kein reiner Fantasie-Ort (min. 3 Zeichen)
        if field == 'headquarters':
            if len(str(value).strip()) < 3:
                continue

        # Gilt als valid: Wert vorhanden (egal ob estimated oder nicht)
        valid_important += 1

    if valid_important < 2:
        return False, f"INSUFFICIENT_IMPORTANT: {valid_important}/2"

    # 3. founding_year Typ-Check
    if 'founding_year' in data and data['founding_year'] is not None:
        try:
            int(data['founding_year'])
        except (ValueError, TypeError):
            return False, "INVALID_YEAR_TYPE"

    # 4. Qualitätsregel: >50% geschätzte Felder → needs_manual
    estimated_fields = meta.get('estimated_fields', []) or []
    if isinstance(estimated_fields, list) and len(estimated_fields) > 0:
        total_fields = len(ALL_DATA_FIELDS)
        estimated_ratio = len(estimated_fields) / total_fields
        if estimated_ratio > 0.50:
            return False, (
                f"TOO_MANY_ESTIMATES: {len(estimated_fields)}/{total_fields} "
                f"({estimated_ratio*100:.0f}%) Felder geschaetzt"
            )

    return True, "OK"


def get_estimated_fields(data: Dict) -> List[str]:
    """Gibt die Liste der geschätzten Felder zurück."""
    meta = data.get('meta', {}) or {}
    return meta.get('estimated_fields', []) or []


def is_empty_value(value) -> bool:
    """Prüft ob ein Wert praktisch leer ist."""
    if value is None:
        return True
    if isinstance(value, str):
        cleaned = value.strip().lower()
        if cleaned in ('', 'n/a', 'unknown', 'unbekannt', 'null', '-', 'keine angabe'):
            return True
    if isinstance(value, list) and len(value) == 0:
        return True
    return False