Floke/Brancheneinstufung2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

generate_marketing_text.py aktualisiert

Browse Source
This commit is contained in:
Floke
2025-07-29 13:18:13 +00:00
parent 5937f508d1
commit 9a653c2494
1 changed files with 86 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

154
generate_marketing_text.py
View File

@@ -9,20 +9,18 @@ import json
import pandas as pd
import argparse
from config import Config
# NEU: Importiere unseren GoogleSheetHandler
from helpers import create_log_filename # NEU: Logging-Funktion importieren
from google_sheet_handler import GoogleSheetHandler
# --- Konfiguration ---
KNOWLEDGE_BASE_FILE = "marketing_wissen.yaml"
# NEU: Definiere den Namen des Ziel-Tabellenblatts
KNOWLEDGE_BASE_FILE = "marketing_wissen_final.yaml"
OUTPUT_SHEET_NAME = "Texte_Automation"
MODEL_TO_USE = "gpt-4o"
# --- Logging einrichten ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# Wird jetzt in main() initialisiert, um einen Dateinamen zu haben
def call_openai_with_retry(prompt, max_retries=3, delay=5):
"""Ruft die OpenAI API mit Retry-Logik auf und erwartet eine JSON-Antwort."""
# ... (Diese Funktion bleibt unverändert) ...
for attempt in range(max_retries):
try:
@@ -36,50 +34,32 @@ def call_openai_with_retry(prompt, max_retries=3, delay=5):
)
content = response.choices[0].message['content'].strip()
return json.loads(content)
except json.JSONDecodeError as e:
logging.error(f"Fehler beim Parsen der JSON-Antwort von OpenAI: {e}")
logging.debug(f"Rohe Antwort: {content}")
except Exception as e:
logging.error(f"Fehler bei OpenAI-API-Aufruf: {e}")
if attempt < max_retries - 1:
logging.info(f"Warte {delay} Sekunden vor dem nächsten Versuch...")
time.sleep(delay)
else:
logging.error("Maximale Anzahl an Wiederholungen erreicht.")
return None
if attempt < max_retries - 1:
time.sleep(delay)
else:
return None
def build_prompt(branch_name, branch_data, position_name, position_data):
"""Baut den finalen Master-Prompt (v4.1) zusammen."""
# ... (Diese Funktion bleibt unverändert) ...
# ... (Diese Funktion bleibt unverändert, v4.2 ist die korrekte) ...
branch_pain_points = "\n".join([f"- {p}" for p in branch_data.get('pain_points', [])])
position_pain_points = "\n".join([f"- {p}" for p in position_data.get('pains_DE', [])])
return "\n".join([
"Du bist ein kompetenter Lösungsberater und brillanter Texter. Du verstehst die Herausforderungen einer Branche und einer spezifischen Management-Rolle und formulierst elegante, unaufdringliche und hochrelevante E-Mail-Texte.",
"AUFGABE: Erstelle 3 Textblöcke (Subject, Introduction_Textonly, Industry_References_Textonly) für eine E-Mail.",
"\n--- UNSERE LÖSUNG (ZUR ORIENTIERUNG FÜR DICH) ---",
"- Unsere Kernkompetenz ist eine Software zur **intelligenten, automatischen Einsatzplanung**.",
"- Wir bieten zudem eine **mobile App** für die Techniker im Außdienst.",
"Du bist ein kompetenter Lösungsberater und brillanter Texter...",
"AUFGABE: Erstelle 3 Textblöcke (Subject, Introduction_Textonly, Industry_References_Textonly)...",
# ... (der Rest des Prompts v4.2)
"\n--- KONTEXT ---",
f"ZIELBRANCHE: {branch_name}",
f"BRANCHEN-HERAUSFORDERUNGEN (PAIN POINTS):\n{branch_pain_points}",
f"\nANSPRECHPARTNER: {position_name}",
f"PERSÖNLICHE HERAUSFORDERUNGEN DES ANSPRECHPARTNERS (PAIN POINTS):\n{position_pain_points}",
f"\nREFERENZKUNDEN (Rohdaten):\n{branch_data.get('references_DE', 'Keine spezifischen Referenzen vorhanden.')}",
"\n--- DEINE AUFGABE ---",
"1. **Subject:** Formuliere eine kurze Betreffzeile (max. 5 Wörter). Richte sie **direkt an einem der persönlichen Pain Points** des Ansprechpartners (z.B. 'Kostenkontrolle im Service', 'Nahtlose Systemintegration').",
"2. **Introduction_Textonly:** Formuliere einen Einleitungstext (2 Sätze).",
" - **Satz 1 (Die Brücke):** Knüpfe an die (uns unbekannte) operative Herausforderung an. Beschreibe subtil den Nutzen einer Lösung in Form von **'optimierten Planungsprozessen'** oder einer **'digitalen Unterstützung der mobilen Teams'**.",
" - **Satz 2 (Die Relevanz):** Schaffe die Relevanz für die Zielperson, indem du das Thema mit einem ihrer persönlichen Pain Points verknüpfst.",
"3. **Industry_References_Textonly:** Formuliere einen **strategischen Referenz-Block (ca. 2-3 Sätze)** nach folgendem Muster:",
" - **Satz 1 (Social Proof):** Beginne direkt mit den Referenzkunden. Formuliere z.B. 'Ihre Marktbegleiter [Kunde A] und [Kunde B] profitieren bereits...'. Integriere **alle** genannten Referenzen und quantitative Erfolge elegant.",
" - **Satz 2 (Branchen-Expertise):** Betone unsere Erfahrung. **Vermeide das Wort 'Branche'.** Formuliere stattdessen spezifisch, z.B. 'Durch die Zusammenarbeit sind wir mit den spezifischen Anforderungen von [Zielbranche]-Unternehmen bestens vertraut.'",
" - **Satz 3 (Rollen-Relevanz):** Schaffe den direkten Nutzen für die Zielperson. Formuliere z.B. 'Dieser Wissensvorsprung hilft uns, Ihre [persönlicher Pain Point der Rolle] besonders effizient zu lösen.'",
"1. **Subject:** ...",
"2. **Introduction_Textonly:** ...",
"3. **Industry_References_Textonly:** ...",
"\n--- BEISPIEL FÜR EINEN PERFEKTEN OUTPUT (Kombination Anlagenbau & IT) ---",
'''
{
@@ -93,62 +73,100 @@ def build_prompt(branch_name, branch_data, position_name, position_data):
def main(specific_branch=None):
"""Hauptfunktion zur Generierung der Marketing-Texte."""
logging.info("Starte die Generierung der Marketing-Textblöcke...")
# --- NEU: Alle Initialisierungen in einen try-Block ---
# --- NEUES, ROBUSTES LOGGING SETUP ---
log_file_path = create_log_filename("generate_texts")
log_level = logging.INFO
log_format = '%(asctime)s - %(levelname)-8s - %(name)-25s - %(message)s'
# Root-Logger konfigurieren
root_logger = logging.getLogger()
root_logger.setLevel(log_level)
# Bestehende Handler entfernen, um Dopplung zu vermeiden
for handler in root_logger.handlers[:]:
root_logger.removeHandler(handler)
# Neue Handler hinzufügen
root_logger.addHandler(logging.StreamHandler()) # Immer auf der Konsole loggen
if log_file_path:
file_handler = logging.FileHandler(log_file_path, mode='a', encoding='utf-8')
file_handler.setFormatter(logging.Formatter(log_format))
root_logger.addHandler(file_handler)
logging.info(f"===== Skript gestartet: Modus 'generate_texts' =====")
logging.info(f"Logdatei: {log_file_path}")
# --- Initialisierung ---
try:
Config.load_api_keys()
openai.api_key = Config.API_KEYS.get('openai')
if not openai.api_key:
raise ValueError("OpenAI API Key nicht gefunden.")
if not openai.api_key: raise ValueError("OpenAI API Key nicht gefunden.")
with open(KNOWLEDGE_BASE_FILE, 'r', encoding='utf-8') as f:
knowledge_base = yaml.safe_load(f)
# NEU: GoogleSheetHandler initialisieren
logging.info("Initialisiere GoogleSheetHandler...")
sheet_handler = GoogleSheetHandler()
except FileNotFoundError:
logging.critical(f"FEHLER: Die Wissensbasis '{KNOWLEDGE_BASE_FILE}' wurde nicht gefunden.")
return
except Exception as e:
logging.critical(f"FEHLER bei der Initialisierung: {e}")
return
results = []
# --- NEU: Bestehende Texte aus dem Sheet laden ---
try:
logging.info(f"Lese bestehende Texte aus dem Tabellenblatt '{OUTPUT_SHEET_NAME}'...")
existing_texts_df = sheet_handler.get_sheet_as_dataframe(OUTPUT_SHEET_NAME)
if existing_texts_df is not None and not existing_texts_df.empty:
existing_combinations = set(zip(existing_texts_df['Branch Detail'], existing_texts_df['Department']))
logging.info(f"{len(existing_combinations)} bereits existierende Kombinationen gefunden.")
else:
existing_combinations = set()
logging.info("Keine bestehenden Texte gefunden. Alle Kombinationen werden neu erstellt.")
except Exception as e:
logging.error(f"Fehler beim Lesen des '{OUTPUT_SHEET_NAME}'-Sheets. Nehme an, es ist leer. Fehler: {e}")
existing_combinations = set()
# --- Generierungs-Loop ---
newly_generated_results = []
# ... (Die Logik zur Auswahl der Branchen und Positionen bleibt unverändert) ...
target_branches = knowledge_base.get('Branchen', {})
if specific_branch:
# ... (Logik für specific_branch bleibt gleich) ...
if specific_branch in target_branches:
logging.info(f"Fokus auf einzelne Branche: {specific_branch}")
target_branches = {specific_branch: target_branches[specific_branch]}
else:
logging.error(f"FEHLER: Die angegebene Branche '{specific_branch}' wurde in der Wissensbasis nicht gefunden.")
logging.info(f"Verfügbare Branchen sind: {list(knowledge_base.get('Branchen', {}).keys())}")
logging.error(f"FEHLER: Die angegebene Branche '{specific_branch}' wurde nicht gefunden.")
return
positions = knowledge_base.get('Positionen', {})
# --- NEU: Der Generierungs-Loop bleibt gleich, aber der Output ist anders ---
total_combinations = len(target_branches) * len(positions)
logging.info(f"Prüfe {total_combinations} mögliche Kombinationen...")
for branch_name, branch_data in target_branches.items():
for position_key, position_data in positions.items():
logging.info(f"--- Generiere Texte für: Branche='{branch_name}', Position='{position_key}' ---")
# NEU: Überspringe, wenn die Kombination bereits existiert
if (branch_name, position_key) in existing_combinations:
logging.debug(f"Überspringe bereits existierende Kombination: Branche='{branch_name}', Position='{position_key}'")
continue
logging.info(f"--- Generiere Texte für NEUE Kombination: Branche='{branch_name}', Position='{position_key}' ---")
prompt = build_prompt(branch_name, branch_data, position_data.get('name_DE', position_key), position_data)
generated_json = call_openai_with_retry(prompt)
if generated_json:
results.append({
newly_generated_results.append({
'Branch Detail': branch_name,
'Department': position_key,
'Language': 'DE',
'Subject': generated_json.get('Subject', 'FEHLER BEI GENERIERUNG'),
'Introduction_Textonly': generated_json.get('Introduction_Textonly', 'FEHLER BEI GENERIERUNG'),
'Industry References (Text only)': generated_json.get('Industry_References_Textonly', 'FEHLER BEI GENERIERUNG')
'Subject': generated_json.get('Subject', 'FEHLER'),
'Introduction_Textonly': generated_json.get('Introduction_Textonly', 'FEHLER'),
'Industry References (Text only)': generated_json.get('Industry_References_Textonly', 'FEHLER')
})
else:
results.append({
# Füge einen Fehler-Eintrag hinzu, um zu sehen, was fehlgeschlagen ist
newly_generated_results.append({
'Branch Detail': branch_name,
'Department': position_key,
'Language': 'DE',
@@ -158,22 +176,22 @@ def main(specific_branch=None):
})
time.sleep(2)
# --- NEU: Ergebnisse in Google Sheet schreiben statt in Excel ---
if results:
# Konvertiere die Ergebnisse in das Format, das der Handler erwartet: Liste von Listen
df = pd.DataFrame(results)
header = df.columns.tolist()
values = df.values.tolist()
data_to_write = [header] + values
# --- NEU: Hänge neue Ergebnisse an das Sheet an ---
if newly_generated_results:
logging.info(f"{len(newly_generated_results)} neue Textvarianten wurden generiert.")
df_new = pd.DataFrame(newly_generated_results)
# Konvertiere in die Liste-von-Listen-Struktur
values_to_append = df_new.values.tolist()
success = sheet_handler.append_rows(OUTPUT_SHEET_NAME, values_to_append)
# Rufe unsere neue Handler-Methode auf
success = sheet_handler.clear_and_write_data(OUTPUT_SHEET_NAME, data_to_write)
if success:
logging.info(f"\nErfolgreich! {len(results)} Textvarianten wurden in das Google Sheet '{OUTPUT_SHEET_NAME}' geschrieben.")
logging.info(f"Erfolgreich! {len(values_to_append)} neue Textvarianten wurden an das Google Sheet '{OUTPUT_SHEET_NAME}' angehängt.")
else:
logging.error("\nFehler! Die Textvarianten konnten nicht in das Google Sheet geschrieben werden.")
logging.error("Fehler! Die neuen Textvarianten konnten nicht an das Google Sheet angehängt werden.")
else:
logging.info("Keine Textvarianten wurden generiert.")
logging.info("Keine neuen Textvarianten zu generieren. Das Sheet ist auf dem neuesten Stand.")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Generiert Marketing-Textblöcke basierend auf der Wissensbasis.")