"""
contact_enricher.py -- Reichert Kontaktdaten automatisch an.
Strategie:
1. Website aus Email-Domain ableiten (falls keine Website vorhanden)
2. Website scrapen (Titel, Meta-Description, Haupttext)
3. Claude analysiert den Text -> Branche, Firmenbeschreibung, Schmerzpunkt
4. Ergebnis in cold_contacts speichern
Verwendung:
python contact_enricher.py --campaign 1 # Alle 'raw' Kontakte einer Kampagne
python contact_enricher.py --batch 50 # Erste 50 'raw' Kontakte
python contact_enricher.py --manual # Interaktiver Modus fuer 'needs_manual'
python contact_enricher.py --contact 42 # Einzelnen Kontakt anreichern
"""
import os
import re
import json
import time
import logging
import argparse
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from urllib.parse import urlparse
# .env explizit aus eigenem Verzeichnis laden
from dotenv import load_dotenv
load_dotenv(dotenv_path=Path(__file__).parent / '.env', override=True)
try:
import requests
from bs4 import BeautifulSoup
HAS_SCRAPING = True
except ImportError:
HAS_SCRAPING = False
from cold_outreach_db import ColdOutreachDB
from llm.router import run_enrichment, check_performance, print_stats
from discovery.orchestrator import run_discovery
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Konfiguration
# ---------------------------------------------------------------------------
SCRAPE_TIMEOUT = 8 # Sekunden pro Request
SCRAPE_MAX_TEXT = 3000 # Zeichen aus Website-Text
DELAY_BETWEEN_REQUESTS = 2 # Sekunden Pause zwischen Scrapes (Anti-Spam)
HEADERS = {
'User-Agent': (
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
'AppleWebKit/537.36 (KHTML, like Gecko) '
'Chrome/120.0.0.0 Safari/537.36'
),
'Accept-Language': 'de-DE,de;q=0.9,en-US;q=0.8,en;q=0.7',
}
# Branchen-Liste fuer strukturierte Ausgabe
INDUSTRY_LIST = [
'Hotel', 'Hotellerie', 'Hostal', 'Pensione',
'Immobilienverwaltung', 'Hausverwaltung', 'Property Management',
'Immobilienmakler', 'Real Estate',
'Bauunternehmen', 'Konstruktion', 'Arquitectura',
'Architekt', 'Innenarchitekt',
'Einzelhandel', 'Retail', 'Online-Shop', 'E-Commerce',
'Gastronomie', 'Restaurant', 'Catering',
'Handwerk', 'Sanitaer', 'Elektro', 'Maler', 'Schreiner',
'IT', 'Software', 'SaaS', 'Tech',
'Marketing', 'Werbeagentur', 'PR',
'Beratung', 'Consulting',
'Rechtsanwalt', 'Anwalt', 'Kanzlei',
'Steuerberater', 'Buchhaltung',
'Arztpraxis', 'Medizin', 'Therapeut', 'Physiotherapie',
'Versicherung',
'Logistik', 'Transport', 'Spedition',
'Bildung', 'Schule', 'Akademie',
'Tourismus', 'Reisebuero', 'Reiseveranstalter',
'Sonstiges',
]
# ---------------------------------------------------------------------------
# Enricher-Klasse
# ---------------------------------------------------------------------------
class ContactEnricher:
"""
Reichert Kontaktdaten automatisch via Website-Scraping + Multi-LLM an.
Nutzt LLM Router mit Fallback: gpt-5-nano → gpt-5-mini → claude-haiku
Validierung & Logging sind automatisch integriert.
"""
def __init__(self, db_path: str = 'email_agent.db'):
self.db = ColdOutreachDB(db_path)
self.enriched_count = 0 # Zur Performance-Überwachung
# ── Oeffentliche API ─────────────────────────────────────────────────────
def enrich_batch(self, limit: int = 50, campaign_id: int = None,
status: str = 'raw') -> Dict:
"""
Bearbeitet bis zu 'limit' Kontakte mit status='raw'.
Nach jedem 50. Kontakt: Performance-Check (Modelle mit <70% runterpriorisieren).
Returns:
Dict mit Statistiken: enriched, needs_manual, errors
"""
contacts = self.db.get_contacts_by_status(status, campaign_id)[:limit]
stats = {'enriched': 0, 'needs_manual': 0, 'errors': 0}
logger.info(f"{len(contacts)} Kontakte zum Anreichern gefunden.")
for i, contact in enumerate(contacts, start=1):
logger.info(f"[{i}/{len(contacts)}] {contact['email']} ({contact.get('company', '')})")
try:
success = self._enrich_contact(contact)
if success:
stats['enriched'] += 1
self.enriched_count += 1
else:
stats['needs_manual'] += 1
time.sleep(DELAY_BETWEEN_REQUESTS)
except Exception as e:
logger.error(f"Fehler bei {contact['email']}: {e}")
self.db.update_contact(contact['id'], {'status': 'needs_manual'})
stats['errors'] += 1
# Performance-Check nach 50 Kontakten
if self.enriched_count > 0 and self.enriched_count % 50 == 0:
logger.info(f"Performance-Check nach {self.enriched_count} Kontakten...")
check_performance(threshold=50)
logger.info(f"Enrichment abgeschlossen: {stats}")
print_stats()
return stats
def enrich_single(self, contact_id: int) -> bool:
"""Einzelnen Kontakt anreichern."""
contact = self.db.get_contact(contact_id)
if not contact:
logger.error(f"Kontakt {contact_id} nicht gefunden.")
return False
return self._enrich_contact(contact)
def enrich_manual(self, campaign_id: int = None):
"""
Interaktiver CLI-Modus fuer Kontakte die manuell gepflegt werden muessen.
"""
contacts = self.db.get_contacts_by_status('needs_manual', campaign_id)
if not contacts:
print("Keine Kontakte mit Status 'needs_manual' gefunden.")
return
print(f"\n{len(contacts)} Kontakte brauchen manuelle Eingabe.\n")
print("Enter druecken ohne Eingabe = Feld leer lassen / unveraendert.")
print("'q' zum Beenden.\n")
for i, contact in enumerate(contacts, start=1):
print("=" * 60)
print(f"[{i}/{len(contacts)}]")
print(f" Name: {contact.get('first_name', '')} {contact.get('last_name', '')}")
print(f" Email: {contact['email']}")
print(f" Firma: {contact.get('company', '')}")
print(f" Branche: {contact.get('industry', '')}")
print("-" * 60)
industry = input(f" Branche [{contact.get('industry', '')}]: ").strip()
if industry == 'q':
print("Manuelles Enrichment beendet.")
break
description = input(f" Was macht die Firma? (1-2 Saetze): ").strip()
pain_point = input(f" Vermutetes Problem: ").strip()
language = input(f" Sprache [de/es/en] [{contact.get('language', 'de')}]: ").strip()
updates = {'status': 'enriched', 'enrichment_source': 'manual',
'enrichment_date': datetime.now().isoformat()}
if industry:
updates['industry'] = industry
if description:
updates['company_description'] = description
if pain_point:
updates['pain_point'] = pain_point
if language:
updates['language'] = language
self.db.update_contact(contact['id'], updates)
print(f" [OK] Gespeichert.")
# ── Interne Logik ────────────────────────────────────────────────────────
def _enrich_contact(self, contact: Dict) -> bool:
"""
Hauptfunktion fuer einen einzelnen Kontakt.
Strategie:
1. Website scrapen (oder aus Cache)
2. Prompt generieren
3. LLM Router aufrufen (mit automatischer Eskalation & Validierung)
4. Ergebnis in DB speichern ODER status=needs_manual setzen
Returns True wenn Enrichment erfolgreich, False wenn manuell noetig.
"""
email = contact['email']
contact_id = contact['id']
domain = self._get_domain(email, contact.get('website', ''))
if not domain:
logger.warning(f"Keine Domain fuer {email}")
self.db.update_contact(contact_id, {'status': 'needs_manual'})
return False
# 1. Discovery Layer -- sammelt Daten aus mehreren Quellen
company_name = contact.get('company', '')
website = contact.get('website', '') or f"https://{domain}"
discovery = run_discovery(company_name=company_name, website=website)
if discovery['low_data']:
logger.warning(
f"[{contact_id}] Discovery: zu wenig Daten "
f"({len(discovery['text_blob'])} Zeichen) → low_data_lead"
)
self.db.update_contact(contact_id, {'status': 'low_data_lead'})
return False
# 2. Prompt generieren (mit Discovery-Daten)
prompt = self._build_enrichment_prompt(
website_text=discovery['text_blob'],
meta_title='',
meta_description='',
company_name=company_name,
email=email,
)
# 3. LLM Router aufrufen (nano → mini → haiku mit Validierung)
success, analysis = run_enrichment(prompt, contact_id)
if not success or not analysis:
# Alle Level fehlgeschlagen ODER 3 hintereinander fail
logger.error(f"[{contact_id}] Alle LLM-Level fehlgeschlagen → needs_manual")
self.db.update_contact(contact_id, {'status': 'needs_manual'})
return False
# 4. Ergebnis speichern (nur wenn richtig validated vom Router)
updates = {
'status': 'enriched',
'enrichment_source': 'auto_llm_router',
'enrichment_date': datetime.now().isoformat(),
}
enrichment_fields = [
'industry', 'company_description', 'pain_point', 'language',
'company_size', 'headquarters', 'founding_year',
'technology_stack', 'key_products', 'linkedin_url'
]
for field in enrichment_fields:
value = analysis.get(field)
if value and not contact.get(field):
value_str = str(value).strip()
if value_str and value_str.lower() not in ('', 'unknown', 'unbekannt', 'n/a', 'null'):
updates[field] = value_str
if field == 'founding_year' and analysis.get('founding_year'):
try:
year = int(analysis['founding_year'])
if not contact.get('founding_year'):
updates['founding_year'] = year
except (ValueError, TypeError):
pass
self.db.update_contact(contact_id, updates)
logger.info(f"[{contact_id}] Erfolgreich angereichert: {email} → "
f"Branche={analysis.get('industry', '?')}, "
f"Größe={analysis.get('company_size', '?')}")
return True
def _get_domain(self, email: str, website: str = '') -> Optional[str]:
"""
Ermittelt die zu scrapende Domain.
Bevorzugt die angegebene Website, faellt auf Email-Domain zurueck.
"""
if website:
# Website-URL bereinigen
if not website.startswith('http'):
website = 'https://' + website
try:
parsed = urlparse(website)
if parsed.netloc:
return website # Vollstaendige URL zurueckgeben
except Exception:
pass
# Email-Domain ableiten
if '@' in email:
domain = email.split('@')[1].lower()
# Bekannte Freemail-Provider ausschliessen
freemail = {'gmail.com', 'yahoo.com', 'outlook.com', 'hotmail.com',
'web.de', 'gmx.de', 'gmx.net', 'icloud.com', 'me.com',
't-online.de', 'freenet.de', 'aol.com', 'mail.com'}
if domain not in freemail:
return f'https://{domain}'
return None
def _scrape_website(self, url: str) -> Tuple[str, str, str, str]:
"""
Scraped eine Website und extrahiert Hauptinhalt.
Returns:
(main_text, meta_title, meta_description, status)
"""
if not HAS_SCRAPING:
logger.error("requests und beautifulsoup4 nicht installiert.")
return '', '', '', 'no_lib'
try:
resp = requests.get(
url, timeout=SCRAPE_TIMEOUT, headers=HEADERS,
allow_redirects=True,
)
resp.raise_for_status()
soup = BeautifulSoup(resp.content, 'html.parser')
# Meta-Daten extrahieren
meta_title = ''
if soup.title:
meta_title = soup.title.string or ''
meta_description = ''
meta_tag = soup.find('meta', attrs={'name': re.compile('description', re.I)})
if meta_tag:
meta_description = meta_tag.get('content', '')
# Haupt-Text extrahieren (Navigation, Footer, Scripts entfernen)
for tag in soup(['script', 'style', 'nav', 'footer', 'header',
'aside', 'form', 'iframe', 'noscript']):
tag.decompose()
main_text = soup.get_text(separator=' ', strip=True)
main_text = re.sub(r'\s+', ' ', main_text)[:SCRAPE_MAX_TEXT]
return main_text, str(meta_title)[:200], str(meta_description)[:500], 'success'
except requests.exceptions.Timeout:
return '', '', '', 'timeout'
except requests.exceptions.SSLError:
# Nochmal ohne SSL-Verifikation versuchen
try:
resp = requests.get(url, timeout=SCRAPE_TIMEOUT, headers=HEADERS,
verify=False, allow_redirects=True)
soup = BeautifulSoup(resp.content, 'html.parser')
for tag in soup(['script', 'style', 'nav', 'footer']):
tag.decompose()
main_text = soup.get_text(separator=' ', strip=True)
main_text = re.sub(r'\s+', ' ', main_text)[:SCRAPE_MAX_TEXT]
return main_text, '', '', 'success_no_ssl'
except Exception:
return '', '', '', 'ssl_error'
except requests.exceptions.ConnectionError:
return '', '', '', 'connection_error'
except Exception as e:
logger.debug(f"Scraping-Fehler fuer {url}: {e}")
return '', '', '', 'error'
def _build_enrichment_prompt(self, website_text: str, meta_title: str,
meta_description: str, company_name: str,
email: str) -> str:
"""
Erstellt den Enrichment-Prompt für den LLM Router.
Dieser Prompt wird an den Router (nano → mini → haiku) gegeben.
Der Router erledigt die API-Calls und Validierung automatisch.
Returns:
Der vollständige Prompt als String (für JSON-Response)
"""
industry_list_str = ', '.join(INDUSTRY_LIST)
prompt = f"""Du analysierst eine Unternehmenswebsite fuer B2B-Akquise.
=== PHASE 1: EXTRAKTION ===
Extrahiere ALLE verfuegbaren Informationen aus den folgenden Daten.
Lies Impressum, Ueber-uns, Kontakt und Produktseiten genau.
FIRMA: {company_name or 'Unbekannt'}
EMAIL-DOMAIN: {email.split('@')[-1] if '@' in email else ''}
META-TITEL: {meta_title}
META-DESCRIPTION: {meta_description}
WEBSITE-TEXT:
{website_text[:2000]}
=== PHASE 2: STRUKTURIERUNG ===
Bringe die extrahierten Daten in das folgende JSON-Schema.
PFLICHTFELDER — duerfen NICHT leer sein:
1. pain_point: Wahrscheinlicher Bedarf/Problem dieser Firma.
Beispiele: Kundenkommunikation, Email-Ueberflutung, Fachkraeftemangel,
Effizienz, Datenverwaltung, Sicherheit, Prozessautomatisierung.
2. company_description: Was macht diese Firma? (1 konkreter Satz)
3. company_size: Leite aus Website-Tonalitaet, Teamgroesse, Produktbreite ab.
NUR diese Werte erlaubt: "1-10 Mitarbeiter", "10-50 Mitarbeiter",
"50-200 Mitarbeiter", "200+ Mitarbeiter"
Wenn nicht explizit → schaetze anhand des Kontexts und markiere als estimated.
Dieses Feld DARF NICHT leer sein.
4. headquarters: Stadt + Land. Ableitung in dieser Reihenfolge:
a) Impressum/Kontakt-Adresse
b) .de → Deutschland, .at → Oesterreich, .ch → Schweiz, .es → Spanien
c) Sprache der Website als Hinweis
Markiere als estimated wenn abgeleitet.
Dieses Feld DARF NICHT leer sein.
WEITERE FELDER:
5. industry: Branche. Waehle aus: {industry_list_str}
6. founding_year: Nur wenn explizit genannt (integer), sonst null
7. technology_stack: Erkennbare Technologien, sonst leer
8. key_products: Top 2-3 Produkte/Services, komma-getrennt
9. linkedin_url: Nur wenn verlinkt, sonst leer
10. language: Hauptsprache (de/es/en)
=== BEISPIEL ===
Website-Text: "Wir sind eine kreative Marketingagentur aus Hamburg..."
Korrekte Ausgabe:
{{
"company_size": "10-50 Mitarbeiter",
"headquarters": "Hamburg, Deutschland",
"meta": {{
"company_size_estimated": true,
"headquarters_estimated": false,
"estimated_fields": ["company_size"]
}}
}}
=== AUSGABE ===
Antworte NUR mit validem JSON, ohne Markdown-Formatierung:
{{
"industry": "...",
"company_description": "...",
"pain_point": "...",
"company_size": "...",
"headquarters": "...",
"founding_year": null,
"technology_stack": "...",
"key_products": "...",
"linkedin_url": "...",
"language": "de",
"meta": {{
"company_size_estimated": false,
"headquarters_estimated": false,
"estimated_fields": []
}}
}}
Wenn die Website eindeutig keinen Unternehmensinhalt hat (Fehlerseite, Parkdomain):
{{"error": "no_business_content"}}"""
return prompt
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main():
parser = argparse.ArgumentParser(
description='Reichert Kontakte via Website-Scraping und Claude an.'
)
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument('--batch', type=int, metavar='N',
help='Erste N Kontakte mit status=raw anreichern')
group.add_argument('--campaign', type=int, metavar='ID',
help='Alle raw-Kontakte einer Kampagne anreichern')
group.add_argument('--manual', action='store_true',
help='Interaktiver Modus fuer needs_manual Kontakte')
group.add_argument('--contact', type=int, metavar='ID',
help='Einzelnen Kontakt anreichern')
parser.add_argument('--db', default='email_agent.db', help='Pfad zur DB-Datei')
parser.add_argument('--campaign-filter', type=int, metavar='ID',
help='Kampagnen-Filter fuer --manual Modus')
args = parser.parse_args()
enricher = ContactEnricher(db_path=args.db)
if args.batch:
print(f"\nEnrichment fuer {args.batch} Kontakte...")
stats = enricher.enrich_batch(limit=args.batch)
print(f"\nErgebnis:")
print(f" Angereichert: {stats['enriched']}")
print(f" Manuell noetig: {stats['needs_manual']}")
print(f" Fehler: {stats['errors']}")
if stats['needs_manual'] > 0:
print(f"\nNaechster Schritt:")
print(f" python contact_enricher.py --manual")
elif args.campaign:
print(f"\nEnrichment fuer Kampagne {args.campaign}...")
stats = enricher.enrich_batch(campaign_id=args.campaign, limit=1000)
print(f"\nErgebnis: {stats}")
elif args.manual:
campaign_filter = getattr(args, 'campaign_filter', None)
enricher.enrich_manual(campaign_id=campaign_filter)
elif args.contact:
print(f"\nEnriche Kontakt {args.contact}...")
success = enricher.enrich_single(args.contact)
print("OK" if success else "Manuell noetig")
if __name__ == '__main__':
main()