"""
excel_exporter.py -- Exportiert DB-Daten zu Excel nach jedem Verarbeitungsschritt.
Automatisch aufgerufen nach:
- contact_enricher.py
- icp_classifier.py
- outreach_generator.py
- outreach_orchestrator.py
Verwendung:
python excel_exporter.py --campaign 2
python excel_exporter.py --campaign 2 --output mein_export.xlsx
"""
import sys
import os
import argparse
from datetime import datetime
from typing import Optional
from cold_outreach_db import ColdOutreachDB
try:
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
HAS_OPENPYXL = True
except ImportError:
HAS_OPENPYXL = False
class ExcelExporter:
"""Exportiert Kampagnen-Daten zu XLSX."""
# Feldkategorien für Farbkodierung
ORIGINAL_FIELDS = {'source_id', 'company', 'email', 'website'}
CRITICAL_ENRICHMENT = {'pain_point', 'company_description'}
IMPORTANT_ENRICHMENT = {'company_size', 'headquarters'}
OPTIONAL_ENRICHMENT = {
'founding_year', 'technology_stack', 'key_products',
'linkedin_url', 'industry', 'language', 'icp_recommended_product'
}
# Header-Farben pro Kategorie
FILL_HEADER_ORIGINAL = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid") # Blau
FILL_HEADER_CRITICAL = PatternFill(start_color="1F5C1F", end_color="1F5C1F", fill_type="solid") # Dunkelgrün
FILL_HEADER_IMPORTANT = PatternFill(start_color="70AD47", end_color="70AD47", fill_type="solid") # Mittelgrün
FILL_HEADER_OPTIONAL = PatternFill(start_color="FFF2CC", end_color="FFF2CC", fill_type="solid") # Hellgelb
# Zell-Farben für Enrichment-Daten
FILL_CELL_FILLED = PatternFill(start_color="E2EFDA", end_color="E2EFDA", fill_type="solid") # Hellgrün
FILL_CELL_EMPTY = PatternFill(start_color="FFE0E0", end_color="FFE0E0", fill_type="solid") # Hellrot
# Schriftfarben
FONT_LIGHT = Font(bold=True, color="FFFFFF")
FONT_DARK = Font(bold=True, color="333333")
# Projekt-Mapping: Kampagnen-Namen -> Projekt-Verzeichnis
PROJECT_MAPPING = {
'portier': 'K:/projekte-AG/portier',
'lp': 'K:/projekte-LP',
'lueftung': 'K:/projekte-LP',
}
def __init__(self, db_path: str = 'email_agent.db'):
self.db = ColdOutreachDB(db_path)
def _get_output_directory(self, campaign: dict) -> str:
"""
Bestimmt das Output-Verzeichnis basierend auf Kampagnen-Namen.
Falls nicht erkannt: email_agent/exports/
"""
campaign_name = (campaign.get('name') or '').lower()
# Nach bekannten Projekten suchen
for keyword, project_dir in self.PROJECT_MAPPING.items():
if keyword in campaign_name:
os.makedirs(project_dir, exist_ok=True)
return project_dir
# Default: email_agent/exports/
default_dir = os.path.join(os.path.dirname(__file__), 'exports')
os.makedirs(default_dir, exist_ok=True)
return default_dir
def export_campaign(self, campaign_id: int, output_file: Optional[str] = None) -> str:
"""
Exportiert alle Daten einer Kampagne zu Excel.
Args:
campaign_id: ID der Kampagne
output_file: Optionaler Dateiname. Wenn None: auto-generated (ins richtige Projekt-Dir)
Returns:
Pfad der erstellten Datei
"""
if not HAS_OPENPYXL:
raise ImportError(
"openpyxl nicht installiert. Bitte: pip install openpyxl"
)
campaign = self.db.get_campaign(campaign_id)
if not campaign:
raise ValueError(f"Kampagne {campaign_id} nicht gefunden.")
# Output-Verzeichnis basierend auf Kampagnen-Name bestimmen
output_dir = self._get_output_directory(campaign)
if not output_file:
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
campaign_name = campaign['name'].replace(' ', '_')[:30]
filename = f"export_{campaign_name}_{timestamp}.xlsx"
output_file = os.path.join(output_dir, filename)
else:
# Wenn Dateiname gegeben: relativ zum Output-Dir interpretieren
if not os.path.isabs(output_file):
output_file = os.path.join(output_dir, output_file)
wb = Workbook()
wb.remove(wb.active)
# Style-Vorlagen
header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
header_font = Font(bold=True, color="FFFFFF")
border = Border(
left=Side(style='thin'),
right=Side(style='thin'),
top=Side(style='thin'),
bottom=Side(style='thin')
)
# Sheet 1: Kampagnen-Info
ws_info = wb.create_sheet("Info", 0)
ws_info['A1'] = "Kampagne"
ws_info['B1'] = campaign['name']
ws_info['A2'] = "ICP"
ws_info['B2'] = campaign['icp']
ws_info['A3'] = "Quelle"
ws_info['B3'] = campaign.get('source_file', '')
ws_info['A4'] = "Erstellt"
ws_info['B4'] = campaign.get('created_at', '')
ws_info['A5'] = "Kontakte (gesamt)"
ws_info['B5'] = campaign.get('contacts_total', 0)
ws_info['A6'] = "Angereichert"
ws_info['B6'] = campaign.get('contacts_enriched', 0)
ws_info['A7'] = "Gesendet"
ws_info['B7'] = campaign.get('contacts_sent', 0)
ws_info['A8'] = "Antworten"
ws_info['B8'] = campaign.get('contacts_replied', 0)
ws_info['A9'] = "Konvertiert"
ws_info['B9'] = campaign.get('contacts_converted', 0)
# Sheet 2-6: Kontakte nach Status
for status in ['raw', 'enriched', 'active', 'replied', 'converted']:
contacts = self.db.get_contacts_by_status(status, campaign_id=campaign_id)
if contacts:
ws = wb.create_sheet(f"Kontakte ({status})")
# source_id zuerst, dann Basis-Info, dann angereicherte Daten
self._write_contacts_sheet(ws, contacts, header_fill, header_font, border,
column_order=[
'source_id', 'company', 'email', 'website',
'industry', 'company_description',
'company_size', 'headquarters', 'founding_year',
'technology_stack', 'key_products', 'linkedin_url',
'pain_point', 'icp_recommended_product', 'language'
])
# Sheet 7: Produkt-Varianten (alle Duplikate die gespeichert wurden)
self._add_variants_sheet(wb, campaign_id, header_fill, header_font, border)
# Sheet 8: Sequenzen (pending, sent, failed)
self._add_sequences_sheets(wb, campaign_id, header_fill, header_font, border)
# Spaltenbreiten anpassen
for ws in wb.sheetnames:
for column in wb[ws].columns:
max_length = 0
column_letter = column[0].column_letter
for cell in column:
if cell.value:
max_length = max(max_length, len(str(cell.value)))
adjusted_width = min(max_length + 2, 50)
wb[ws].column_dimensions[column_letter].width = adjusted_width
wb.save(output_file)
return output_file
def _write_contacts_sheet(self, ws, contacts: list, header_fill, header_font, border, column_order=None):
"""Schreibt Kontakte in ein Sheet."""
if not contacts:
return
if column_order:
# Nutze Custom-Reihenfolge
headers = column_order
# Fehlende Spalten hinzufügen
all_keys = set(contacts[0].keys())
for key in all_keys:
if key not in headers:
headers.append(key)
else:
headers = list(contacts[0].keys())
ws.append(headers)
# Header formatieren
for col_num, header in enumerate(headers, 1):
cell = ws.cell(row=1, column=col_num)
cell.fill = header_fill
cell.font = header_font
cell.alignment = Alignment(horizontal='center', vertical='center')
# Daten einfügen
for contact in contacts:
row = [contact.get(h, '') for h in headers]
ws.append(row)
def _add_variants_sheet(self, wb, campaign_id: int, header_fill, header_font, border):
"""Fügt ein Sheet mit allen Produkt-Varianten hinzu."""
with self.db.get_connection() as conn:
rows = conn.execute("""
SELECT c.source_id, c.company, c.email, v.*
FROM contact_variants v
JOIN cold_contacts c ON v.contact_id = c.id
WHERE c.campaign_id = ?
ORDER BY CAST(c.source_id AS INTEGER), v.id
""", (campaign_id,)).fetchall()
if rows:
ws = wb.create_sheet("Varianten (Produkte)")
rows_dict = [dict(r) for r in rows]
# Nur wichtige Spalten anzeigen
headers = ['source_id', 'company', 'email', 'offer_title', 'priority_tier', 'industry_segment', 'buyer_relevance']
ws.append(headers)
# Header formatieren
for col_num in range(1, len(headers) + 1):
cell = ws.cell(row=1, column=col_num)
cell.fill = header_fill
cell.font = header_font
# Daten einfügen
for row in rows_dict:
ws.append([row.get(h, '') for h in headers])
def _add_sequences_sheets(self, wb, campaign_id: int, header_fill, header_font, border):
"""Fügt Sequenz-Daten hinzu."""
with self.db.get_connection() as conn:
for status in ['pending', 'sent', 'failed']:
rows = conn.execute("""
SELECT s.*, c.company, c.email, c.icp
FROM cold_sequences s
JOIN cold_contacts c ON s.contact_id = c.id
WHERE c.campaign_id = ? AND s.status = ?
ORDER BY s.scheduled_for ASC
""", (campaign_id, status)).fetchall()
if rows:
ws = wb.create_sheet(f"Sequenzen ({status})")
rows_dict = [dict(r) for r in rows]
headers = list(rows_dict[0].keys())
ws.append(headers)
# Header formatieren
for col_num in range(1, len(headers) + 1):
cell = ws.cell(row=1, column=col_num)
cell.fill = header_fill
cell.font = header_font
# Daten einfügen
for row in rows_dict:
ws.append([row.get(h, '') for h in headers])
def main():
parser = argparse.ArgumentParser(
description='Exportiert Kampagnen-Daten zu Excel.'
)
parser.add_argument('--campaign', type=int, required=True, help='Kampagnen-ID')
parser.add_argument('--output', help='Optionaler Dateiname (auto-generated wenn leer)')
parser.add_argument('--db', default='email_agent.db', help='Pfad zur DB')
args = parser.parse_args()
try:
exporter = ExcelExporter(db_path=args.db)
output_file = exporter.export_campaign(args.campaign, args.output)
print(f"\n[OK] Export erfolgreich:")
print(f" Datei: {output_file}")
print(f" Groesse: {__import__('os').path.getsize(output_file) / 1024:.1f} KB")
except Exception as e:
print(f"\n[ERROR] Fehler: {e}")
sys.exit(1)
if __name__ == '__main__':
main()