"""
multi_source_enricher.py -- Multi-Source Recherche fuer needs_manual Kontakte.
Strategie (Dottore Spec):
1. Website (Startseite + Impressum + Ueber uns + Leistungen)
2. Impressum separat analysieren
3. LinkedIn via DuckDuckGo Suche
4. Google Business / Maps via DuckDuckGo
5. Weitere Quellen wenn noetig
Verwendung:
python multi_source_enricher.py --campaign 6
python multi_source_enricher.py --contact 42
"""
import os
import re
import json
import time
import logging
import argparse
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from urllib.parse import urljoin, urlparse
from dotenv import load_dotenv
load_dotenv(dotenv_path=Path(__file__).parent / '.env', override=True)
import requests
import warnings
warnings.filterwarnings('ignore', message='Unverified HTTPS request')
from bs4 import BeautifulSoup
from ddgs import DDGS
from cold_outreach_db import ColdOutreachDB
from llm.router import EnrichmentRouter, run_enrichment
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)
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'
}
SCRAPE_TIMEOUT = 12
MAX_TEXT_PER_PAGE = 2000
# Unterseiten die geprueft werden (relativ zur Basis-URL)
SUBPAGES = [
'', # Startseite
'/impressum',
'/ueber-uns', '/about', '/about-us', '/ueber-mich',
'/leistungen', '/services', '/angebote',
'/kontakt', '/contact',
]
class MultiSourceEnricher:
def __init__(self, db_path: str = 'email_agent.db'):
self.db = ColdOutreachDB(db_path)
self.router = EnrichmentRouter()
# ── Web-Scraping ──────────────────────────────────────────────────────────
def _scrape_url(self, url: str) -> str:
"""Scraped eine einzelne URL, gibt bereinigten Text zurueck."""
try:
resp = requests.get(url, timeout=SCRAPE_TIMEOUT, headers=HEADERS,
allow_redirects=True, verify=False)
resp.raise_for_status()
soup = BeautifulSoup(resp.content, 'html.parser')
for tag in soup(['script', 'style', 'nav', 'footer', 'header',
'aside', 'form', 'iframe', 'noscript']):
tag.decompose()
text = soup.get_text(separator=' ', strip=True)
return re.sub(r'\s+', ' ', text)[:MAX_TEXT_PER_PAGE]
except Exception:
return ''
def _scrape_all_pages(self, base_url: str) -> Dict[str, str]:
"""
Scraped Startseite + relevante Unterseiten.
Gibt Dict {seitenname: text} zurueck.
"""
parsed = urlparse(base_url)
root = f"{parsed.scheme}://{parsed.netloc}"
results = {}
for path in SUBPAGES:
url = root + path if path else base_url
text = self._scrape_url(url)
if text.strip():
label = path.strip('/') or 'startseite'
results[label] = text
logger.debug(f" Gescrapt: {url} ({len(text)} Zeichen)")
return results
# ── DuckDuckGo Suche ──────────────────────────────────────────────────────
def _search_linkedin(self, company_name: str) -> Tuple[str, str]:
"""
Sucht LinkedIn-Firmenprofil via DuckDuckGo.
Gibt (linkedin_url, snippet) zurueck.
"""
try:
with DDGS() as ddgs:
results = list(ddgs.text(
f'site:linkedin.com/company "{company_name}"',
max_results=3
))
for r in results:
href = r.get('href', '')
if 'linkedin.com/company/' in href:
return href, r.get('body', '')
except Exception as e:
logger.debug(f"LinkedIn-Suche fehlgeschlagen: {e}")
return '', ''
def _search_google_business(self, company_name: str, domain: str) -> str:
"""
Sucht Google Business / Maps Eintrag via DuckDuckGo.
Gibt Snippet-Text zurueck.
"""
try:
with DDGS() as ddgs:
results = list(ddgs.text(
f'"{company_name}" Standort Mitarbeiter Branche',
max_results=5
))
snippets = []
for r in results:
href = r.get('href', '')
# Nur externe Quellen (nicht die eigene Website)
if domain and domain in href:
continue
snippets.append(r.get('body', ''))
return ' '.join(snippets)[:2000]
except Exception as e:
logger.debug(f"Business-Suche fehlgeschlagen: {e}")
return ''
# ── Prompt-Bau ────────────────────────────────────────────────────────────
def _build_prompt(self, company_name: str, website: str,
pages: Dict[str, str], linkedin_url: str,
linkedin_snippet: str, external_info: str) -> str:
"""Baut den Multi-Source Enrichment Prompt."""
pages_text = ''
for name, text in pages.items():
pages_text += f"\n--- {name.upper()} ---\n{text}\n"
sources_used = list(pages.keys())
if linkedin_url:
sources_used.append('linkedin')
if external_info:
sources_used.append('web_search')
return f"""Du analysierst ein Unternehmen fuer B2B-Kaltakquise. Nutze ALLE bereitgestellten Quellen.
FIRMA: {company_name}
WEBSITE: {website}
LINKEDIN: {linkedin_url or 'nicht gefunden'}
=== WEBSITE-INHALTE (mehrere Seiten) ===
{pages_text}
=== LINKEDIN SNIPPET ===
{linkedin_snippet or 'kein Ergebnis'}
=== EXTERNE QUELLEN (Web-Suche) ===
{external_info or 'kein Ergebnis'}
=== AUFGABE ===
Erstelle ein vollstaendiges JSON mit diesen Feldern. Alle Felder muessen befuellt sein.
Schätzungen sind erlaubt — markiere sie in meta.estimated_fields.
ERLAUBTE Werte fuer company_size:
"1-10 Mitarbeiter" | "10-50 Mitarbeiter" | "50-200 Mitarbeiter" | "200+ Mitarbeiter"
REGELN:
- pain_point: konkret und geschaeftsrelevant (kein Generisches)
- headquarters: "Stadt, Land" Format, bevorzugt aus Impressum
- founding_year: nur Zahl oder leer
- language: "de" | "en" | "fr" | "it" | "es"
WICHTIG: Antworte NUR mit validem JSON, keine Erklaerungen.
{{
"company": "{company_name}",
"website": "{website}",
"industry": "",
"company_description": "",
"company_size": "",
"headquarters": "",
"founding_year": "",
"technology_stack": [],
"key_products": [],
"linkedin_url": "{linkedin_url}",
"pain_point": "",
"icp_recommended_product": "",
"language": "",
"meta": {{
"sources_used": {json.dumps(sources_used)},
"estimated_fields": [],
"confidence": "high|medium|low"
}}
}}"""
# ── Hauptlogik ────────────────────────────────────────────────────────────
def enrich_contact(self, contact: Dict) -> bool:
"""
Reichert einen einzelnen Kontakt mit Multi-Source Recherche an.
Gibt True zurueck wenn erfolgreich.
"""
contact_id = contact['id']
company = contact.get('company', '')
website = contact.get('website', '') or ''
email = contact.get('email', '') or ''
if not website and '@' in email:
domain = email.split('@')[1]
website = f"https://{domain}"
domain = urlparse(website).netloc if website else ''
logger.info(f"[{contact_id}] Multi-Source: {company} | {website}")
# 1. Alle Website-Seiten scrapen
pages = {}
if website:
pages = self._scrape_all_pages(website)
logger.info(f" Website-Seiten gefunden: {list(pages.keys())}")
# 2. LinkedIn suchen
linkedin_url, linkedin_snippet = self._search_linkedin(company)
if linkedin_url:
logger.info(f" LinkedIn: {linkedin_url}")
# 3. Google Business
external_info = self._search_google_business(company, domain)
# 4. Pruefen ob genuegend Daten vorhanden
total_text = sum(len(t) for t in pages.values()) + len(linkedin_snippet) + len(external_info)
if total_text < 100:
logger.warning(f"[{contact_id}] Zu wenig Daten gefunden — ueberspringe")
return False
# 5. Prompt bauen und LLM aufrufen
prompt = self._build_prompt(
company_name=company,
website=website,
pages=pages,
linkedin_url=linkedin_url,
linkedin_snippet=linkedin_snippet,
external_info=external_info
)
max_budget = int(os.getenv('MAX_TOKENS_PER_CONTACT_MULTI', '40000'))
success, result = self.router.run_enrichment(prompt, contact_id=contact_id,
max_budget=max_budget)
if not success or not result:
logger.error(f"[{contact_id}] LLM Router: kein Ergebnis")
return False
# 6. Ergebnis in DB speichern
updates = {
'industry': result.get('industry', ''),
'company_description': result.get('company_description', ''),
'company_size': result.get('company_size', ''),
'headquarters': result.get('headquarters', ''),
'founding_year': result.get('founding_year', ''),
'technology_stack': json.dumps(result.get('technology_stack', []), ensure_ascii=False),
'key_products': json.dumps(result.get('key_products', []), ensure_ascii=False),
'linkedin_url': result.get('linkedin_url', '') or linkedin_url,
'pain_point': result.get('pain_point', ''),
'icp_recommended_product': result.get('icp_recommended_product', ''),
'language': result.get('language', ''),
'enrichment_source': 'multi_source',
'enrichment_date': __import__('datetime').datetime.now().isoformat(),
'status': 'enriched',
}
self.db.update_contact(contact_id, updates)
logger.info(f"[{contact_id}] Gespeichert: {company} | {result.get('industry')} | {result.get('company_size')}")
return True
def enrich_campaign(self, campaign_id: int) -> Dict:
"""Alle needs_manual Kontakte einer Kampagne anreichern."""
contacts = self.db.get_contacts_by_status('needs_manual', campaign_id=campaign_id)
logger.info(f"{len(contacts)} needs_manual Kontakte gefunden.")
stats = {'enriched': 0, 'failed': 0}
for i, contact in enumerate(contacts, 1):
logger.info(f"[{i}/{len(contacts)}] {contact.get('company')}")
try:
success = self.enrich_contact(contact)
if success:
stats['enriched'] += 1
else:
stats['failed'] += 1
except Exception as e:
logger.error(f"Fehler bei {contact.get('company')}: {e}")
stats['failed'] += 1
time.sleep(2) # Rate limiting
return stats
def main():
parser = argparse.ArgumentParser(description='Multi-Source Enricher fuer needs_manual Kontakte.')
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument('--campaign', type=int, help='Kampagnen-ID')
group.add_argument('--contact', type=int, help='Einzelne Kontakt-ID')
parser.add_argument('--db', default='email_agent.db')
args = parser.parse_args()
enricher = MultiSourceEnricher(db_path=args.db)
if args.campaign:
print(f"\nMulti-Source Enrichment fuer Kampagne {args.campaign}...")
stats = enricher.enrich_campaign(args.campaign)
print(f"\nErgebnis: {stats}")
elif args.contact:
contact = enricher.db.get_contact(args.contact)
if not contact:
print(f"Kontakt {args.contact} nicht gefunden.")
return
success = enricher.enrich_contact(contact)
print(f"\n{'Erfolgreich' if success else 'Fehlgeschlagen'}")
if __name__ == '__main__':
main()