Floke
4a1f392c40
Überarbeite den Prompt für die Generierung von Intro-Texten in der Marketing Matrix. Das Intro beginnt nun direkt mit der branchenspezifischen Produktkategorie und dem zentralen Nutzen, anstatt die Herausforderung zu wiederholen. Dies verbessert die Prägnanz und Relevanz der E-Mails.
289 lines
13 KiB
Python
289 lines
13 KiB
Python
import sys
|
|
import os
|
|
import json
|
|
import argparse
|
|
import re
|
|
import google.generativeai as genai
|
|
|
|
# Setup Environment
|
|
sys.path.append(os.path.join(os.path.dirname(__file__), "../../"))
|
|
|
|
from backend.database import SessionLocal, Industry, Persona, MarketingMatrix
|
|
from backend.config import settings
|
|
|
|
# --- Configuration ---
|
|
MODEL_NAME = "gemini-2.0-flash" # High quality copy
|
|
|
|
def extract_segment(text: str, marker: str) -> str:
|
|
"""
|
|
Extracts a text block starting with [marker].
|
|
Example: [Primary Product: Cleaning] ... [Secondary Product: Service]
|
|
"""
|
|
if not text: return ""
|
|
|
|
# Split by square brackets that look like headers [Text: ...]
|
|
# We look for the marker inside the header
|
|
# Simplified Regex: Capture everything inside brackets as ONE group
|
|
pattern = r'\[(.*?)\]'
|
|
segments = re.split(pattern, text)
|
|
|
|
# segments[0] is text before first bracket
|
|
# segments[1] is content of first bracket (header)
|
|
# segments[2] is content after first bracket (body)
|
|
# ...
|
|
|
|
best_match = ""
|
|
|
|
for i in range(1, len(segments), 2):
|
|
header = segments[i]
|
|
content = segments[i+1]
|
|
|
|
# print(f"DEBUG: Checking Header: '{header}' for Marker: '{marker}'") # Uncomment for deep debug
|
|
|
|
if marker.lower() in header.lower():
|
|
return content.strip()
|
|
|
|
# Fallback: If no markers found, return full text (legacy support)
|
|
if "Primary Product" not in text and "Secondary Product" not in text:
|
|
return text
|
|
|
|
return ""
|
|
|
|
def generate_prompt(industry: Industry, persona: Persona) -> str:
|
|
"""
|
|
Builds the prompt for the AI to generate the marketing texts.
|
|
Combines Industry context with Persona specific pains/gains and Product Category.
|
|
"""
|
|
|
|
# 1. Determine Product Focus Strategy
|
|
# Default: Primary
|
|
target_scope = "Primary Product"
|
|
target_category = industry.primary_category
|
|
|
|
# Special Rule: "Operativer Entscheider" gets Secondary Product IF ops_focus_secondary is True
|
|
# Logic: A Nursing Director (Ops) doesn't care about floor cleaning (Facility),
|
|
# but cares about Service Robots (Secondary).
|
|
if persona.name == "Operativer Entscheider" and industry.ops_focus_secondary:
|
|
target_scope = "Secondary Product"
|
|
target_category = industry.secondary_category
|
|
print(f" -> STRATEGY SWITCH: Using {target_scope} for {persona.name}")
|
|
|
|
# Fallback if secondary was requested but not defined
|
|
if not target_category:
|
|
target_category = industry.primary_category
|
|
target_scope = "Primary Product" # Fallback to primary if secondary category object is missing
|
|
|
|
product_context = f"{target_category.name}: {target_category.description}" if target_category else "Intelligente Robotik-Lösungen"
|
|
|
|
# 2. Extract specific segments from industry pains/gains based on scope
|
|
industry_pains = extract_segment(industry.pains, target_scope)
|
|
industry_gains = extract_segment(industry.gains, target_scope)
|
|
|
|
# Fallback: If specific scope is empty (e.g. no Secondary Pains defined), try Primary
|
|
if not industry_pains and target_scope == "Secondary Product":
|
|
print(f" -> WARNING: No specific Pains found for {target_scope}. Fallback to Primary.")
|
|
industry_pains = extract_segment(industry.pains, "Primary Product")
|
|
industry_gains = extract_segment(industry.gains, "Primary Product")
|
|
|
|
# 3. Handle Persona Data
|
|
try:
|
|
persona_pains = json.loads(persona.pains) if persona.pains else []
|
|
persona_gains = json.loads(persona.gains) if persona.gains else []
|
|
except:
|
|
persona_pains = [persona.pains] if persona.pains else []
|
|
persona_gains = [persona.gains] if persona.gains else []
|
|
|
|
prompt = f"""
|
|
Du bist ein kompetenter Lösungsberater und brillanter Texter.
|
|
AUFGABE: Erstelle 3 Textblöcke (Subject, Introduction_Textonly, Industry_References_Textonly) für eine E-Mail an einen Entscheider.
|
|
|
|
--- KONTEXT ---
|
|
ZIELBRANCHE: {industry.name}
|
|
BRANCHEN-HERAUSFORDERUNGEN (PAIN POINTS):
|
|
{industry_pains}
|
|
|
|
FOKUS-PRODUKT (LÖSUNG):
|
|
{product_context}
|
|
|
|
ANSPRECHPARTNER (ROLLE): {persona.name}
|
|
PERSÖNLICHE HERAUSFORDERUNGEN DES ANSPRECHPARTNERS (PAIN POINTS):
|
|
{chr(10).join(['- ' + str(p) for p in persona_pains])}
|
|
|
|
--- DEINE AUFGABE ---
|
|
1. **Subject:** Formuliere eine kurze Betreffzeile (max. 6 Wörter). Richte sie **direkt an einem der persönlichen Pain Points** des Ansprechpartners oder dem zentralen Branchen-Pain. Sei scharfsinnig, nicht werblich.
|
|
|
|
2. **Introduction_Textonly:** Formuliere einen prägnanten Einleitungstext (max. 2 Sätze).
|
|
- **WICHTIG:** Gehe davon aus, dass die spezifische Herausforderung des Kunden bereits im Satz davor [Opener] genannt wurde. **Wiederhole die Herausforderung NICHT.**
|
|
- **Satz 1 (Die Lösung & der Gain):** Beginne direkt mit der Lösung. Nenne die im Kontext `FOKUS-PRODUKT` definierte **Produktkategorie** (z.B. "automatisierte Reinigungsroboter") und verbinde sie mit dem zentralen Nutzen (Gain) aus den `BRANCHEN-HERAUSFORDERUNGEN`. Beispiel: "Genau hier setzen unsere automatisierten Reinigungsroboter an, indem sie eine lückenlose und auditsichere Hygiene gewährleisten."
|
|
- **Satz 2 (Die Relevanz):** Stelle die Relevanz für die Zielperson her, indem du einen ihrer `PERSÖNLICHE HERAUSFORDERUNGEN` adressierst. Beispiel: "Für Sie als Infrastruktur-Verantwortlicher bedeutet dies vor allem eine reibungslose Integration in bestehende Abläufe, ohne den Betrieb zu stören."
|
|
|
|
3. **Industry_References_Textonly:** Formuliere einen **strategischen Referenz-Block (ca. 2-3 Sätze)** nach folgendem Muster:
|
|
- **Satz 1 (Social Proof):** Beginne direkt mit dem Nutzen, den vergleichbare Unternehmen in der Branche {industry.name} bereits erzielen. (Erfinde keine Firmennamen, sprich von "Führenden Einrichtungen" oder "Vergleichbaren Häusern").
|
|
- **Satz 2 (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.'
|
|
|
|
--- BEISPIEL FÜR EINEN PERFEKTEN OUTPUT ---
|
|
{{
|
|
"Subject": "Kostenkontrolle im Service",
|
|
"Introduction_Textonly": "Genau bei der Optimierung dieser Serviceprozesse können erhebliche Effizienzgewinne erzielt werden. Für Sie als Finanzleiter ist dabei die Sicherstellung der Profitabilität bei gleichzeitiger Kostentransparenz von zentraler Bedeutung.",
|
|
"Industry_References_Textonly": "Vergleichbare Unternehmen profitieren bereits massiv von automatisierten Prozessen. Unsere Erfahrung zeigt, dass die grundlegenden Herausforderungen in der Einsatzplanung oft branchenübergreifend ähnlich sind. Dieser Wissensvorsprung hilft uns, Ihre Ziele bei der Kostenkontrolle und Profitabilitätssteigerung besonders effizient zu unterstützen."
|
|
}}
|
|
|
|
--- FORMAT ---
|
|
Antworte NUR mit einem validen JSON-Objekt. Keine Markdown-Blöcke (```json), kein erklärender Text.
|
|
Format:
|
|
{{
|
|
"subject": "...",
|
|
"intro": "...",
|
|
"social_proof": "..."
|
|
}}
|
|
"""
|
|
return prompt
|
|
|
|
def mock_call(prompt: str):
|
|
"""Simulates an API call for dry runs."""
|
|
print(f"\n--- [MOCK] GENERATING PROMPT ---\n{prompt[:800]}...\n--------------------------------")
|
|
return {
|
|
"subject": "[MOCK] Effizienzsteigerung in der Produktion",
|
|
"intro": "[MOCK] Als Produktionsleiter wissen Sie, wie teuer Stillstand ist. Unsere Roboter helfen.",
|
|
"social_proof": "[MOCK] Ähnliche Betriebe sparten 20% Kosten."
|
|
}
|
|
|
|
def real_gemini_call(prompt: str):
|
|
if not settings.GEMINI_API_KEY:
|
|
raise ValueError("GEMINI_API_KEY not set in config/env")
|
|
|
|
genai.configure(api_key=settings.GEMINI_API_KEY)
|
|
|
|
# Configure Model
|
|
generation_config = {
|
|
"temperature": 0.7,
|
|
"top_p": 0.95,
|
|
"top_k": 64,
|
|
"max_output_tokens": 1024,
|
|
"response_mime_type": "application/json",
|
|
}
|
|
|
|
model = genai.GenerativeModel(
|
|
model_name=MODEL_NAME,
|
|
generation_config=generation_config,
|
|
)
|
|
|
|
response = model.generate_content(prompt)
|
|
|
|
try:
|
|
# Clean response if necessary (Gemini usually returns clean JSON with mime_type set, but safety first)
|
|
text = response.text.strip()
|
|
if text.startswith("```json"):
|
|
text = text[7:-3].strip()
|
|
elif text.startswith("```"):
|
|
text = text[3:-3].strip()
|
|
|
|
parsed_json = json.loads(text)
|
|
if isinstance(parsed_json, list):
|
|
if len(parsed_json) > 0:
|
|
return parsed_json[0]
|
|
else:
|
|
raise ValueError("Empty list returned from API")
|
|
return parsed_json
|
|
except Exception as e:
|
|
print(f"JSON Parse Error: {e}. Raw Response: {response.text}")
|
|
raise
|
|
|
|
def run_matrix_generation(dry_run: bool = True, force: bool = False, specific_industry: str = None):
|
|
db = SessionLocal()
|
|
try:
|
|
query = db.query(Industry)
|
|
if specific_industry:
|
|
query = query.filter(Industry.name == specific_industry)
|
|
|
|
industries = query.all()
|
|
personas = db.query(Persona).all()
|
|
|
|
print(f"Found {len(industries)} Industries and {len(personas)} Personas.")
|
|
print(f"Mode: {'DRY RUN (No API calls, no DB writes)' if dry_run else 'LIVE - GEMINI GENERATION'}")
|
|
|
|
# Pre-load categories to avoid lazy load issues if detached
|
|
# (SQLAlchemy session is open, so should be fine, but good practice)
|
|
|
|
total_combinations = len(industries) * len(personas)
|
|
processed = 0
|
|
|
|
for ind in industries:
|
|
print(f"\n>>> Processing Industry: {ind.name} (Ops Secondary: {ind.ops_focus_secondary})")
|
|
for pers in personas:
|
|
processed += 1
|
|
print(f"[{processed}/{total_combinations}] Check: {ind.name} x {pers.name}")
|
|
|
|
# Check existing
|
|
existing = db.query(MarketingMatrix).filter(
|
|
MarketingMatrix.industry_id == ind.id,
|
|
MarketingMatrix.persona_id == pers.id
|
|
).first()
|
|
|
|
if existing and not force:
|
|
print(f" -> Skipped (Already exists)")
|
|
continue
|
|
|
|
# Generate
|
|
prompt = generate_prompt(ind, pers)
|
|
|
|
if dry_run:
|
|
result = mock_call(prompt)
|
|
else:
|
|
try:
|
|
result = real_gemini_call(prompt)
|
|
|
|
# Normalize Keys (Case-Insensitive)
|
|
normalized_result = {}
|
|
for k, v in result.items():
|
|
normalized_result[k.lower()] = v
|
|
|
|
# Map known variations to standardized keys
|
|
if "introduction_textonly" in normalized_result:
|
|
normalized_result["intro"] = normalized_result["introduction_textonly"]
|
|
if "industry_references_textonly" in normalized_result:
|
|
normalized_result["social_proof"] = normalized_result["industry_references_textonly"]
|
|
|
|
# Validation using normalized keys
|
|
if not normalized_result.get("subject") or not normalized_result.get("intro"):
|
|
print(f" -> Invalid result structure. Keys found: {list(result.keys())}")
|
|
print(f" -> Raw Result: {json.dumps(result, indent=2)}")
|
|
continue
|
|
|
|
except Exception as e:
|
|
print(f" -> API ERROR: {e}")
|
|
continue
|
|
|
|
# Write to DB (only if not dry run)
|
|
if not dry_run:
|
|
if not existing:
|
|
new_entry = MarketingMatrix(
|
|
industry_id=ind.id,
|
|
persona_id=pers.id,
|
|
subject=normalized_result.get("subject"),
|
|
intro=normalized_result.get("intro"),
|
|
social_proof=normalized_result.get("social_proof")
|
|
)
|
|
db.add(new_entry)
|
|
print(f" -> Created new entry.")
|
|
else:
|
|
existing.subject = normalized_result.get("subject")
|
|
existing.intro = normalized_result.get("intro")
|
|
existing.social_proof = normalized_result.get("social_proof")
|
|
print(f" -> Updated entry.")
|
|
|
|
db.commit()
|
|
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
finally:
|
|
db.close()
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--live", action="store_true", help="Actually call Gemini and write to DB")
|
|
parser.add_argument("--force", action="store_true", help="Overwrite existing matrix entries")
|
|
parser.add_argument("--industry", type=str, help="Specific industry name to process")
|
|
args = parser.parse_args()
|
|
|
|
run_matrix_generation(dry_run=not args.live, force=args.force, specific_industry=args.industry) |