""" excel_importer.py -- Importiert Excel/CSV-Kontaktlisten in die cold_outreach Datenbank. Verwendung: python excel_importer.py --file import/leads.xlsx --campaign "Mario Leads April 2026" --icp AG python excel_importer.py --file import/leads.csv --campaign "LP Hotels Q2" --icp LP """ import os import re import csv import sys import argparse import logging from dataclasses import dataclass, field from datetime import datetime from typing import Dict, List, Optional, Tuple try: import openpyxl HAS_OPENPYXL = True except ImportError: HAS_OPENPYXL = False from cold_outreach_db import ColdOutreachDB logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Spalten-Mapping (DE / ES / EN Aliase) # --------------------------------------------------------------------------- COLUMN_ALIASES: Dict[str, List[str]] = { 'first_name': ['vorname', 'name', 'first_name', 'firstname', 'nombre', 'prenom', 'first name', 'ansprechpartner', 'kontaktperson'], 'last_name': ['nachname', 'last_name', 'lastname', 'apellido', 'surname', 'familienname', 'last name'], 'email': ['contact_email', 'contact email', 'email', 'e-mail', 'e_mail', 'emailadresse', 'email address', 'emailaddress', 'correo', 'mail'], 'company': ['firma', 'company', 'unternehmen', 'empresa', 'organisation', 'organization', 'betrieb', 'arbeitgeber', 'company name', 'company_name'], 'role': ['position', 'rolle', 'title', 'job_title', 'cargo', 'funktion', 'designation', 'job title', 'berufsbezeichnung', 'abteilung'], 'industry': ['branche', 'industry', 'sektor', 'sector', 'branchenbezeichnung', 'wirtschaftsbereich', 'industry_segment', 'industry segment'], 'notes': ['notizen', 'notes', 'bemerkung', 'anmerkung', 'kommentar', 'comentarios', 'remarks', 'comment', 'comments', 'outreach_angle', 'outreach angle'], 'website': ['website', 'url', 'web', 'homepage', 'webseite', 'internetseite', 'web address', 'webaddress', 'webpage'], 'phone': ['telefon', 'phone', 'tel', 'telefono', 'handy', 'mobil', 'mobile', 'phone number', 'rufnummer'], 'language': ['sprache', 'language', 'idioma', 'lang'], } EMAIL_REGEX = re.compile(r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$') # --------------------------------------------------------------------------- # Datenklassen # --------------------------------------------------------------------------- @dataclass class ImportReport: campaign_id: int campaign_name: str total_rows: int = 0 imported: int = 0 skipped_duplicate: int = 0 skipped_no_email: int = 0 skipped_invalid_email: int = 0 errors: List[str] = field(default_factory=list) def summary(self) -> str: lines = [ f"Kampagne: {self.campaign_name} (ID: {self.campaign_id})", f"Gesamt gelesen: {self.total_rows}", f"Importiert: {self.imported}", f"Duplikate: {self.skipped_duplicate}", f"Keine Email: {self.skipped_no_email}", f"Ungueltige Email: {self.skipped_invalid_email}", ] if self.errors: lines.append(f"Fehler: {len(self.errors)}") for e in self.errors[:5]: lines.append(f" - {e}") return '\n'.join(lines) # --------------------------------------------------------------------------- # Kern-Logik # --------------------------------------------------------------------------- class ExcelImporter: """Importiert Excel (.xlsx) und CSV-Dateien in cold_contacts.""" def __init__(self, db_path: str = 'email_agent.db'): self.db = ColdOutreachDB(db_path) # ── Oeffentliche API ───────────────────────────────────────────────────── def import_file(self, file_path: str, campaign_name: str, icp: str = 'AG') -> ImportReport: """ Hauptfunktion. Importiert eine Excel- oder CSV-Datei. Args: file_path: Pfad zur Excel/CSV-Datei campaign_name: Name der Kampagne (z.B. "Mario Leads April 2026") icp: AG / LP / beide Returns: ImportReport mit Statistiken """ if not os.path.exists(file_path): raise FileNotFoundError(f"Datei nicht gefunden: {file_path}") # Kampagne anlegen campaign_id = self.db.create_campaign( name=campaign_name, icp=icp, source_file=os.path.basename(file_path), ) report = ImportReport(campaign_id=campaign_id, campaign_name=campaign_name) # Datei einlesen ext = os.path.splitext(file_path)[1].lower() if ext == '.csv': rows = self._read_csv(file_path) elif ext in ('.xlsx', '.xls'): rows = self._read_excel(file_path) else: raise ValueError(f"Unbekanntes Dateiformat: {ext} (erwartet: .xlsx, .csv)") if not rows: logger.warning("Datei ist leer oder hat keine erkennbaren Daten.") return report # Spalten mappen headers = list(rows[0].keys()) col_map = self._detect_columns(headers) logger.info(f"Erkannte Spalten: {col_map}") if 'email' not in col_map: raise ValueError("Keine Email-Spalte gefunden. Bitte Spaltenbezeichnung prüfen.") # Zeilen verarbeiten for i, row in enumerate(rows, start=2): report.total_rows += 1 try: contact = self._map_row(row, col_map) contact['campaign_id'] = campaign_id contact['icp'] = icp # Validierung valid, reason = self._validate_contact(contact) if not valid: if reason == 'no_email': report.skipped_no_email += 1 elif reason == 'invalid_email': report.skipped_invalid_email += 1 else: report.errors.append(f"Zeile {i}: {reason}") continue # Duplikat-Check: als Variant speichern statt zu skippen existing_contact_id = None with self.db.get_connection() as conn: result = conn.execute( "SELECT id FROM cold_contacts WHERE email = ?", (contact['email'].lower().strip(),) ).fetchone() existing_contact_id = result[0] if result else None if existing_contact_id: # Duplikat: als Variant zum bestehenden Kontakt hinzufügen variant_data = { 'offer_title': row.get('offer_title', ''), 'industry_segment': row.get('industry_segment', contact.get('industry', '')), 'buyer_relevance': row.get('buyer_relevance', ''), 'outreach_angle': row.get('outreach_angle', ''), 'priority_tier': row.get('priority_tier', ''), 'contact_page': row.get('contact_page', ''), } self.db.add_variant(existing_contact_id, variant_data, is_primary=False) report.skipped_duplicate += 1 logger.debug(f"Variant hinzugefuegt zu {contact['email']}") continue # Importieren self.db.add_contact(contact, create_sequence=False) report.imported += 1 logger.debug(f"Importiert: {contact['email']}") except Exception as e: report.errors.append(f"Zeile {i}: {e}") logger.error(f"Fehler bei Zeile {i}: {e}") # Kampagnen-Statistik aktualisieren self.db.update_campaign_stats(campaign_id) logger.info(report.summary()) return report # ── Hilfsmethoden ─────────────────────────────────────────────────────── def _read_csv(self, file_path: str) -> List[Dict]: """CSV-Datei einlesen. Versucht verschiedene Trennzeichen.""" for delimiter in [',', ';', '\t', '|']: try: with open(file_path, newline='', encoding='utf-8-sig') as f: reader = csv.DictReader(f, delimiter=delimiter) rows = list(reader) if rows and len(rows[0]) > 1: return rows except Exception: continue # Fallback with open(file_path, newline='', encoding='utf-8-sig') as f: return list(csv.DictReader(f)) def _read_excel(self, file_path: str) -> List[Dict]: """Excel-Datei einlesen via openpyxl.""" if not HAS_OPENPYXL: raise ImportError("openpyxl nicht installiert. Bitte: pip install openpyxl") wb = openpyxl.load_workbook(file_path, read_only=True, data_only=True) ws = wb.active rows = list(ws.iter_rows(values_only=True)) if not rows: return [] # Erste Zeile = Header headers = [str(h).strip() if h is not None else f'col_{i}' for i, h in enumerate(rows[0])] result = [] for row in rows[1:]: if all(v is None for v in row): continue # Leere Zeile überspringen result.append({h: (str(v).strip() if v is not None else '') for h, v in zip(headers, row)}) wb.close() return result def _detect_columns(self, headers: List[str]) -> Dict[str, str]: """ Mappt Excel-Spaltenbezeichnungen auf interne Feldnamen. Returns: Dict: {internes_feld: excel_spalte} """ normalized = {h.lower().strip().replace(' ', '_'): h for h in headers} mapping = {} for field_name, aliases in COLUMN_ALIASES.items(): for alias in aliases: alias_norm = alias.lower().replace(' ', '_') if alias_norm in normalized: mapping[field_name] = normalized[alias_norm] break return mapping def _map_row(self, row: Dict, col_map: Dict[str, str]) -> Dict: """ Mappt eine Zeile auf interne Felder. Nicht erkannte Spalten werden in 'notes' konkateniert. """ contact = {} mapped_cols = set(col_map.values()) for field_name, excel_col in col_map.items(): contact[field_name] = str(row.get(excel_col, '') or '').strip() # Source-ID speichern (z.B. list_index aus Original-Datei) if 'list_index' in row and row['list_index']: contact['source_id'] = str(int(float(row['list_index']))) # Nicht gemappte Spalten in notes (aber nicht list_index) extra_parts = [] for col, val in row.items(): if col not in mapped_cols and col != 'list_index' and val: extra_parts.append(f"{col}: {val}") if extra_parts: existing_notes = contact.get('notes', '') extra = ' | '.join(extra_parts) contact['notes'] = f"{existing_notes} | {extra}".strip(' |') # Email normalisieren if 'email' in contact: contact['email'] = contact['email'].lower().strip() # Sprachvermutung aus Domain if 'email' in contact and not contact.get('language'): domain = contact['email'].split('@')[-1] if '@' in contact['email'] else '' if domain.endswith('.es') or domain.endswith('.cat'): contact['language'] = 'es' elif domain.endswith('.de') or domain.endswith('.at') or domain.endswith('.ch'): contact['language'] = 'de' else: contact['language'] = 'de' # Default return contact def _validate_contact(self, contact: Dict) -> Tuple[bool, str]: """ Validiert einen Kontakt. Gibt (True, '') oder (False, Grund) zurück. Wenn Email fehlt ABER Website vorhanden: trotzdem OK (Email-Lookup später). """ email = contact.get('email', '').strip() company = contact.get('company', '').strip() website = contact.get('website', '').strip() # Wenn Email leer, aber Website vorhanden: markieren für Email-Lookup if not email: if website: # Website vorhanden → kann später Email-Lookup machen # Künstliche Email generieren als Placeholder domain = website.replace('http://', '').replace('https://', '').split('/')[0] contact['email'] = f'contact@{domain}' contact['notes'] = (contact.get('notes', '') + ' | EMAIL_PLACEHOLDER: Website-Lookup noetig').strip() return True, '' else: return False, 'no_email_and_no_website' if not EMAIL_REGEX.match(email): return False, 'invalid_email' # Mindestens Name oder Firma if not company and not contact.get('first_name'): return False, 'no_name_or_company' return True, '' # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser(description='Importiert Excel/CSV in die Outreach-Datenbank.') parser.add_argument('--file', required=True, help='Pfad zur Excel- oder CSV-Datei') parser.add_argument('--campaign', required=True, help='Kampagnenname (z.B. "Mario Leads April 2026")') parser.add_argument('--icp', default='AG', choices=['AG', 'LP', 'beide'], help='Zielgruppe: AG (KI) / LP (Lueftung) / beide') parser.add_argument('--db', default='email_agent.db', help='Pfad zur Datenbankdatei') args = parser.parse_args() # Import-Verzeichnis anlegen falls noetig import_dir = os.path.join(os.path.dirname(args.file)) if import_dir: os.makedirs(import_dir, exist_ok=True) importer = ExcelImporter(db_path=args.db) print(f"\nImportiere: {args.file}") print(f"Kampagne: {args.campaign}") print(f"ICP: {args.icp}") print("-" * 50) try: report = importer.import_file( file_path=args.file, campaign_name=args.campaign, icp=args.icp, ) print("\nERGEBNIS:") print(report.summary()) print(f"\nNaechster Schritt:") print(f" python contact_enricher.py --campaign {report.campaign_id}") except Exception as e: print(f"\nFehler: {e}") sys.exit(1) if __name__ == '__main__': main()