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