303 lines
14 KiB
Python
303 lines
14 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 []
|
|
|
|
# Advanced Persona Context
|
|
persona_context = f"""
|
|
BESCHREIBUNG/DENKWEISE: {persona.description or 'Nicht definiert'}
|
|
WAS DIESE PERSON ÜBERZEUGT: {persona.convincing_arguments or 'Nicht definiert'}
|
|
RELEVANTE KPIs: {persona.kpis or 'Nicht definiert'}
|
|
"""
|
|
|
|
prompt = f"""
|
|
Du bist ein kompetenter Lösungsberater und brillanter Texter für B2B-Marketing.
|
|
AUFGABE: Erstelle 3 hoch-personalisierte 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}
|
|
{persona_context}
|
|
|
|
SPEZIFISCHE HERAUSFORDERUNGEN (PAIN POINTS) DER ROLLE:
|
|
{chr(10).join(['- ' + str(p) for p in persona_pains])}
|
|
|
|
SPEZIFISCHE NUTZEN (GAINS) DER ROLLE:
|
|
{chr(10).join(['- ' + str(g) for g in persona_gains])}
|
|
|
|
--- DEINE AUFGABE ---
|
|
Deine Texte müssen "voll ins Zentrum" der Rolle treffen. Vermeide oberflächliche Floskeln. Nutze die Details zur Denkweise, den KPIs und den Überzeugungsargumenten, um eine tiefgreifende Relevanz zu erzeugen.
|
|
|
|
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 einem Nutzen, der für diese Rolle (siehe `WAS DIESE PERSON ÜBERZEUGT` und `GAINS`) besonders kritisch ist.
|
|
- **Satz 2 (Die Relevanz):** Stelle die Relevanz für die Zielperson her, indem du eine ihrer `PERSÖNLICHE HERAUSFORDERUNGEN` oder `KPIs` adressierst. Beispiel: "Für Sie als [Rolle] bedeutet dies vor allem [Nutzen bezogen auf KPI oder Pain]."
|
|
|
|
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. Nutze dabei die Informationen aus `BESCHREIBUNG/DENKWEISE`, um den Ton perfekt zu treffen.
|
|
|
|
--- 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) |