"""
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