"""
review_cli.py -- Interaktiver CLI fuer manuelle Email-Pruefung vor dem Versand.
Karlo geht jede generierte Email durch und:
- Genehmigt sie [g]
- Editiert sie [e]
- Ueberspringt sie [s]
- Regeneriert sie [r]
NUR genehmigte Emails werden versendet.
Verwendung:
python review_cli.py
python review_cli.py --campaign 1
python review_cli.py --step 1
"""
import os
import sys
import argparse
import subprocess
import tempfile
import logging
from typing import List, Dict, Optional
from cold_outreach_db import ColdOutreachDB
from outreach_generator import OutreachGenerator
logging.basicConfig(level=logging.WARNING) # Im Review-Modus weniger Logs
# ---------------------------------------------------------------------------
# Hilfsfunktionen
# ---------------------------------------------------------------------------
def clear_screen():
os.system('cls' if os.name == 'nt' else 'clear')
def print_separator(char='=', width=70):
print(char * width)
def print_email_card(seq: Dict, index: int, total: int):
"""Zeigt eine Email formatiert im Terminal."""
clear_screen()
print_separator()
print(f" EMAIL {index}/{total}")
print_separator()
# Kontakt-Info
name = f"{seq.get('first_name', '')} {seq.get('last_name', '')}".strip()
print(f" AN: {name} <{seq['email']}>")
print(f" FIRMA: {seq.get('company', '---')}")
print(f" BRANCHE: {seq.get('industry', '---')}")
print(f" ICP: {seq.get('icp', '?')} | Produkt: {seq.get('icp_recommended_product', '?')}")
print(f" STEP: {seq.get('sequence_step', '?')}")
score = seq.get('personalization_score')
score_str = f"{score:.2f}" if score is not None else "?"
score_color = "OK" if (score or 0) >= 0.6 else "NIEDRIG"
print(f" SCORE: {score_str} ({score_color})")
print_separator('-')
print(f" BETREFF: {seq.get('subject', '---')}")
print_separator('-')
body = seq.get('body', '')
# Zeilenumbruch bei langen Zeilen
for line in body.split('\n'):
if len(line) > 68:
words = line.split()
current = ' '
for word in words:
if len(current) + len(word) + 1 > 70:
print(current)
current = ' ' + word
else:
current += ' ' + word if current != ' ' else ' ' + word
print(current)
else:
print(f" {line}")
print_separator()
# Personalisierungs-Elemente anzeigen
elements = seq.get('personalization_elements')
if elements and isinstance(elements, str):
try:
import ast
elements = ast.literal_eval(elements)
except Exception:
elements = [elements]
if elements:
print(f" PERSONALISIERUNG: {', '.join(elements[:4])}")
print_separator('-')
print()
def edit_in_editor(text: str) -> str:
"""Oeffnet den Text in einem Editor und gibt das Ergebnis zurueck."""
# Temporaere Datei
with tempfile.NamedTemporaryFile(
mode='w', suffix='.txt', delete=False, encoding='utf-8'
) as f:
f.write(text)
tmp_path = f.name
# Editor oeffnen (EDITOR-Variable, fallback auf notepad/nano)
editor = os.environ.get('EDITOR', 'notepad' if os.name == 'nt' else 'nano')
try:
subprocess.run([editor, tmp_path], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
# Fallback: einfache stdin-Eingabe
print("\nEditor nicht gefunden. Bitte neuen Text eingeben (Leerzeile + Enter zum Beenden):")
lines = []
while True:
line = input()
if line == '' and lines and lines[-1] == '':
break
lines.append(line)
return '\n'.join(lines[:-1]) if lines else text
# Gespeicherten Text einlesen
with open(tmp_path, 'r', encoding='utf-8') as f:
result = f.read()
os.unlink(tmp_path)
return result.strip()
# ---------------------------------------------------------------------------
# Review-Session
# ---------------------------------------------------------------------------
class ReviewSession:
"""Interaktive Review-Session fuer generierte Emails."""
def __init__(self, db_path: str = 'email_agent.db'):
self.db = ColdOutreachDB(db_path)
self.generator = OutreachGenerator(db_path=db_path)
self.stats = {'approved': 0, 'edited': 0, 'skipped': 0, 'regenerated': 0}
def run(self, campaign_id: int = None, step: int = None):
"""Startet die Review-Session."""
emails = self.db.get_pending_review(campaign_id=campaign_id)
if step:
emails = [e for e in emails if e.get('sequence_step') == step]
if not emails:
print("\nKeine Emails zur Pruefung gefunden.")
print("Tipp: Zuerst 'python outreach_generator.py --step 1' ausfuehren.")
return
print(f"\n{len(emails)} Emails zur Pruefung.")
print("Tasten: [g]enehmigen | [e]ditieren | [s]kippen | [r]egenerieren | [q]uit\n")
input("Enter zum Starten...")
i = 0
while i < len(emails):
seq = emails[i]
print_email_card(seq, i + 1, len(emails))
action = self._get_action()
if action == 'g':
self.db.approve_email(seq['id'])
self.stats['approved'] += 1
print(" Genehmigt.")
i += 1
elif action == 'e':
new_body = edit_in_editor(seq.get('body', ''))
new_subject = input(f"\n Betreff [{seq.get('subject', '')}]: ").strip()
if not new_subject:
new_subject = seq.get('subject', '')
self.db.approve_email(
seq['id'],
edited_body=new_body if new_body != seq.get('body') else None,
edited_subject=new_subject if new_subject != seq.get('subject') else None,
)
self.stats['edited'] += 1
print(" Editiert und genehmigt.")
i += 1
elif action == 's':
self.db.reject_email(seq['id'])
self.stats['skipped'] += 1
print(" Uebersprungen.")
i += 1
elif action == 'r':
print(" Regeneriere...")
result = self.generator.regenerate(seq['id'])
if result.get('accepted'):
print(f" Neue Email generiert (Score: {result['score']:.2f}). Pruefe neu.")
# Aktualisierte Sequenz laden
pending = self.db.get_pending_review(campaign_id)
new_seq = next((e for e in pending if e['id'] == seq['id']), None)
if new_seq:
emails[i] = new_seq
else:
print(f" Regenerierung fehlgeschlagen (Score: {result.get('score', 0):.2f}).")
print(" Bitte manuell editieren [e] oder ueberspringen [s].")
self.stats['regenerated'] += 1
# Nicht weiterspringen -- gleiche Email nochmal zeigen
elif action == 'q':
print("\nReview beendet.")
break
self._print_summary(len(emails))
def _get_action(self) -> str:
"""Liest eine Taste vom Benutzer."""
while True:
choice = input(" Aktion [g/e/s/r/q]: ").strip().lower()
if choice in ('g', 'e', 's', 'r', 'q', ''):
return choice if choice else 's'
def _print_summary(self, total: int):
print_separator()
print(" REVIEW-ZUSAMMENFASSUNG")
print_separator('-')
print(f" Genehmigt: {self.stats['approved']}")
print(f" Editiert: {self.stats['edited']}")
print(f" Uebersprungen: {self.stats['skipped']}")
print(f" Regeneriert: {self.stats['regenerated']}")
versandbar = self.stats['approved'] + self.stats['edited']
print_separator('-')
print(f" Versandbereit: {versandbar} von {total}")
print_separator()
if versandbar > 0:
print(f"\nNaechster Schritt:")
print(f" python outreach_orchestrator.py --send-approved")
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main():
parser = argparse.ArgumentParser(
description='Interaktiver Review fuer generierte Outreach-Emails.'
)
parser.add_argument('--campaign', type=int, metavar='ID',
help='Nur Emails einer Kampagne pruefen')
parser.add_argument('--step', type=int, choices=[1, 2, 3],
help='Nur Emails eines Sequence-Steps pruefen')
parser.add_argument('--db', default='email_agent.db', help='Pfad zur DB-Datei')
args = parser.parse_args()
session = ReviewSession(db_path=args.db)
session.run(
campaign_id=args.campaign,
step=args.step,
)
if __name__ == '__main__':
main()