"""
icp_classifier.py -- Ordnet Kontakte dem richtigen ICP (LP/AG/beide) zu.
LP = Lüftungsprofi (Mallorca AirServices): Lueftung, Schimmel, Luftqualitaet
AG = Agent Solutions: KI-Automatisierung, Email Agent, Telefon Agent
Verwendung:
python icp_classifier.py --campaign 1
python icp_classifier.py --all
python icp_classifier.py --contact 42
"""
import os
import json
import logging
import argparse
from typing import Dict, List, Optional, Tuple
import anthropic
from cold_outreach_db import ColdOutreachDB
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Regelbasierte Branchen-Listen
# ---------------------------------------------------------------------------
# LP: Gebaeudebezogen -- Lueftung und Luftqualitaet relevant
LP_INDUSTRIES = {
'hotel', 'hotellerie', 'hostal', 'hostel', 'pensione', 'hotel & resort',
'resort', 'aparthotel', 'ferienwohnung', 'alquiler vacacional',
'immobilienverwaltung', 'hausverwaltung', 'property management', 'inmobiliaria',
'immobilienmakler', 'real estate', 'makler',
'bauunternehmen', 'konstruktion', 'baufirma', 'construccion', 'bautraeger',
'architekt', 'architektur', 'innenarchitekt', 'arquitectura',
'facility management', 'gebaeudemanagement', 'gebaeudereinigung',
'wohnungsbau', 'wohnanlage', 'mietverwaltung',
}
# AG: Digitale Prozesse -- KI-Automatisierung macht Sinn
AG_INDUSTRIES = {
'it', 'software', 'saas', 'tech', 'technologie',
'ecommerce', 'e-commerce', 'online-shop', 'onlineshop', 'online shop',
'marketing', 'werbeagentur', 'agentur', 'pr', 'werbung', 'digitalmarketing',
'beratung', 'consulting', 'unternehmensberatung',
'rechtsanwalt', 'anwalt', 'kanzlei', 'notar',
'steuerberater', 'buchhaltung', 'wirtschaftspruefer',
'arztpraxis', 'medizin', 'zahnarzt', 'therapeut', 'physiotherapie',
'psychologe', 'praxis',
'versicherung', 'versicherungsmakler',
'logistik', 'transport', 'spedition', 'kurier',
'bildung', 'schule', 'akademie', 'nachhilfe', 'coaching',
'finanzberatung', 'finanzdienstleistung', 'bank',
'handel', 'einzelhandel', 'retail', 'grosshandel',
'handwerk', 'sanitaer', 'elektro', 'maler', 'schreiner', 'installateur',
'gastronomie', 'restaurant', 'catering', 'imbiss',
'tourismus', 'reisebuero', 'reiseveranstalter',
}
# Beide: Hotel und Immobilien koennen BEIDE Produkte nutzen
BOTH_INDUSTRIES = {
'hotel', 'hotellerie', 'hostal', 'aparthotel',
'immobilienverwaltung', 'hausverwaltung', 'property management',
'ferienwohnung', 'alquiler vacacional',
}
# ---------------------------------------------------------------------------
# ICP-Spezifische Produkt-Empfehlungen
# ---------------------------------------------------------------------------
LP_PRODUCTS = {
'hotel': 'lueftung',
'immobilienverwaltung': 'lueftung',
'hausverwaltung': 'lueftung',
'bauunternehmen': 'lueftung',
'wohnungsbau': 'lueftung',
'facility management': 'lueftung',
}
AG_PRODUCTS = {
'rechtsanwalt': 'email_agent',
'steuerberater': 'email_agent',
'arztpraxis': 'telefon_agent',
'zahnarzt': 'telefon_agent',
'handwerk': 'telefon_agent',
'sanitaer': 'telefon_agent',
'gastronomie': 'email_agent',
'hotel': 'email_agent',
'it': 'email_agent',
'ecommerce': 'email_agent',
'logistik': 'telefon_agent',
'versicherung': 'email_agent',
}
# ---------------------------------------------------------------------------
# Classifier
# ---------------------------------------------------------------------------
class ICPClassifier:
"""Klassifiziert Kontakte als LP, AG oder beide."""
def __init__(self, db_path: str = 'email_agent.db',
anthropic_api_key: str = None):
self.db = ColdOutreachDB(db_path)
self.claude = anthropic.Anthropic(
api_key=anthropic_api_key or os.getenv('ANTHROPIC_API_KEY')
)
# ── Oeffentliche API ─────────────────────────────────────────────────────
def classify_all(self, campaign_id: int = None) -> Dict:
"""
Klassifiziert alle 'enriched' Kontakte die noch kein ICP haben.
Returns:
Statistiken: {lp: int, ag: int, beide: int, unclear: int}
"""
contacts = self.db.get_contacts_by_status('enriched', campaign_id)
unclassified = [c for c in contacts if not c.get('icp') or c['icp'] == 'AG']
# Alle enriched neu klassifizieren (auch bereits mit 'AG' Standard-Wert)
contacts_to_classify = [c for c in contacts]
stats = {'lp': 0, 'ag': 0, 'beide': 0, 'unclear': 0}
logger.info(f"{len(contacts_to_classify)} Kontakte zu klassifizieren.")
for contact in contacts_to_classify:
icp, product, confidence = self._classify_contact(contact)
updates = {
'icp': icp,
'icp_confidence': confidence,
'icp_recommended_product': product,
}
self.db.update_contact(contact['id'], updates)
if icp == 'LP':
stats['lp'] += 1
elif icp == 'AG':
stats['ag'] += 1
elif icp == 'beide':
stats['beide'] += 1
else:
stats['unclear'] += 1
logger.info(
f"{contact['email']} -> ICP: {icp} | Produkt: {product} "
f"| Confidence: {confidence:.2f}"
)
logger.info(f"Klassifikation abgeschlossen: {stats}")
return stats
def classify_contact(self, contact_id: int) -> Tuple[str, str, float]:
"""Einzelnen Kontakt klassifizieren und in DB speichern."""
contact = self.db.get_contact(contact_id)
if not contact:
logger.error(f"Kontakt {contact_id} nicht gefunden.")
return 'AG', '', 0.5
icp, product, confidence = self._classify_contact(contact)
self.db.update_contact(contact_id, {
'icp': icp,
'icp_confidence': confidence,
'icp_recommended_product': product,
})
return icp, product, confidence
# ── Interne Logik ────────────────────────────────────────────────────────
def _classify_contact(self, contact: Dict) -> Tuple[str, str, float]:
"""
Hauptklassifikation. Gibt (icp, product, confidence) zurueck.
Versucht zuerst regelbasiert, dann Claude.
"""
industry = (contact.get('industry') or '').lower().strip()
# 1. Regelbasierte Klassifikation (schnell, kostenlos)
icp, product = self._rule_based(industry)
if icp:
return icp, product, 0.9
# 2. Claude fuer unklare Faelle
if contact.get('company_description') or contact.get('pain_point'):
return self._claude_classify(contact)
# 3. Fallback
logger.debug(f"Kein Match fuer Branche '{industry}' -- Fallback AG")
return 'AG', 'email_agent', 0.4
def _rule_based(self, industry: str) -> Tuple[Optional[str], str]:
"""
Schnelle regelbasierte Klassifikation.
Returns: (icp, product) oder (None, '') wenn unklar.
"""
if not industry:
return None, ''
# Normalisierung
industry_norm = industry.lower().replace('-', '').replace(' ', '')
# 1. Pruefe auf "beide" zuerst
for both_ind in BOTH_INDUSTRIES:
if both_ind.lower().replace(' ', '') in industry_norm:
product = AG_PRODUCTS.get(both_ind, 'portier')
return 'beide', f"lueftung+{product}"
# 2. LP-Match
for lp_ind in LP_INDUSTRIES - BOTH_INDUSTRIES:
if lp_ind.lower().replace(' ', '') in industry_norm:
product = LP_PRODUCTS.get(lp_ind, 'lueftung')
return 'LP', product
# 3. AG-Match
for ag_ind in AG_INDUSTRIES:
if ag_ind.lower().replace(' ', '') in industry_norm:
product = AG_PRODUCTS.get(ag_ind, 'email_agent')
return 'AG', product
return None, ''
def _claude_classify(self, contact: Dict) -> Tuple[str, str, float]:
"""
Claude-basierte Fein-Klassifikation fuer unklare Faelle.
"""
prompt = f"""Analysiere diesen Geschaeftskontakt und bestimme das passende Produkt.
KONTAKT:
- Firma: {contact.get('company', 'Unbekannt')}
- Branche: {contact.get('industry', 'Unbekannt')}
- Beschreibung: {contact.get('company_description', '')}
- Schmerzpunkt: {contact.get('pain_point', '')}
PRODUKTE:
PRODUKT LP (Mallorca AirServices):
Dezentrale Lueftungssysteme, CO2/Feuchte-Messung, Schimmelpraevention.
Relevant fuer: Gebaeudeeigentuemer, Verwalter, Hotels, Bauunternehmen auf Mallorca.
NUR fuer physische Gebaeude mit Lueftungs-/Feuchteproblemen relevant.
PRODUKT AG (Agent Solutions):
- Email Agent: KI-automatisierte Emailbearbeitung
- Telefon Agent: KI-Telefonassistent
Relevant fuer: Jedes KMU mit hohem Email/Telefon-Aufkommen.
Antworte als JSON:
{{
"icp": "LP" | "AG" | "beide",
"recommended_product": "lueftung" | "email_agent" | "telefon_agent" | "lueftung+email_agent",
"reasoning": "Kurz, max 20 Woerter",
"confidence": 0.0-1.0
}}"""
try:
response = self.claude.messages.create(
model='claude-haiku-4-5-20251001',
max_tokens=200,
messages=[{'role': 'user', 'content': prompt}],
)
raw = response.content[0].text.strip()
start = raw.find('{')
end = raw.rfind('}') + 1
data = json.loads(raw[start:end])
icp = data.get('icp', 'AG')
product = data.get('recommended_product', 'email_agent')
confidence = float(data.get('confidence', 0.7))
logger.debug(f"Claude ICP: {icp} / {product} ({confidence:.2f}) - {data.get('reasoning', '')}")
return icp, product, confidence
except Exception as e:
logger.error(f"Claude ICP-Klassifikation fehlgeschlagen: {e}")
return 'AG', 'email_agent', 0.4
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main():
parser = argparse.ArgumentParser(
description='ICP-Klassifikation: Ordnet Kontakte LP/AG/beide zu.'
)
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument('--all', action='store_true',
help='Alle enriched Kontakte klassifizieren')
group.add_argument('--campaign', type=int, metavar='ID',
help='Alle enriched Kontakte einer Kampagne klassifizieren')
group.add_argument('--contact', type=int, metavar='ID',
help='Einzelnen Kontakt klassifizieren')
parser.add_argument('--db', default='email_agent.db', help='Pfad zur DB-Datei')
args = parser.parse_args()
classifier = ICPClassifier(db_path=args.db)
if args.all:
print("\nKlassifiziere alle enriched Kontakte...")
stats = classifier.classify_all()
print(f"\nErgebnis:")
print(f" LP (Lueftung): {stats['lp']}")
print(f" AG (KI): {stats['ag']}")
print(f" Beide: {stats['beide']}")
print(f" Unklar: {stats['unclear']}")
print(f"\nNaechster Schritt:")
print(f" python outreach_generator.py --generate-step 1")
elif args.campaign:
print(f"\nKlassifiziere Kampagne {args.campaign}...")
stats = classifier.classify_all(campaign_id=args.campaign)
print(f"\nErgebnis: {stats}")
elif args.contact:
icp, product, confidence = classifier.classify_contact(args.contact)
print(f"\nErgebnis: ICP={icp} | Produkt={product} | Confidence={confidence:.2f}")
if __name__ == '__main__':
main()