258 lines
9.5 KiB
Python
258 lines
9.5 KiB
Python
import argparse
|
|
import base64
|
|
import json
|
|
import logging
|
|
import sys
|
|
import os
|
|
from datetime import datetime
|
|
import content_db_manager as db_manager
|
|
|
|
# Ensure helper path is correct
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
|
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
|
|
|
|
from helpers import call_gemini_flash, scrape_website_details
|
|
from config import Config
|
|
|
|
# --- LOGGING CONFIGURATION ---
|
|
LOG_DIR = "/app/Log_from_docker"
|
|
if not os.path.exists(LOG_DIR):
|
|
try:
|
|
os.makedirs(LOG_DIR)
|
|
except:
|
|
pass # Should be mounted
|
|
|
|
run_timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
|
log_file = os.path.join(LOG_DIR, f"content_python_debug.log")
|
|
|
|
# Configure root logger
|
|
logging.basicConfig(
|
|
level=logging.DEBUG,
|
|
format='%(asctime)s [%(levelname)s] %(message)s',
|
|
handlers=[
|
|
logging.FileHandler(log_file),
|
|
logging.StreamHandler(sys.stderr) # Write logs to stderr so stdout keeps clean JSON
|
|
]
|
|
)
|
|
|
|
logging.info(f"--- Content Orchestrator Started ({run_timestamp}) ---")
|
|
|
|
try:
|
|
Config.load_api_keys()
|
|
logging.info("API Keys loaded successfully.")
|
|
except Exception as e:
|
|
logging.error(f"Failed to load API keys: {e}")
|
|
|
|
def get_copywriter_instruction(lang='de'):
|
|
return r"""
|
|
Du bist ein Senior Copywriter und SEO-Experte. Deine Spezialität ist der 'Challenger Sale'.
|
|
Du schreibst Texte, die fachlich tief fundiert, professionell und leicht aggressiv/fordernd sind.
|
|
|
|
DEIN STIL:
|
|
- Keine Buzzwords ohne Substanz.
|
|
- Fokus auf den 'Cost of Inaction' (Was kostet es den Kunden, wenn er NICHT handelt?).
|
|
- Übersetzung von Technik in geschäftlichen Nutzen.
|
|
- SEO-Integration: Baue Keywords natürlich aber präsent ein.
|
|
"""
|
|
|
|
# --- MODES ---
|
|
|
|
def list_gtm_projects(payload):
|
|
logging.info("Executing list_gtm_projects")
|
|
try:
|
|
projects = db_manager.get_all_gtm_projects()
|
|
logging.debug(f"Found {len(projects)} GTM projects")
|
|
return {"projects": projects}
|
|
except Exception as e:
|
|
logging.error(f"Error listing GTM projects: {e}", exc_info=True)
|
|
return {"error": str(e), "projects": []}
|
|
|
|
def import_project(payload):
|
|
logging.info(f"Executing import_project with payload keys: {list(payload.keys())}")
|
|
gtm_id = payload.get('gtmProjectId')
|
|
|
|
if not gtm_id:
|
|
logging.error("Missing gtmProjectId in payload")
|
|
return {"error": "Missing gtmProjectId"}
|
|
|
|
try:
|
|
logging.info(f"Attempting to import GTM Project ID: {gtm_id}")
|
|
result = db_manager.import_gtm_project(gtm_id)
|
|
|
|
if not result:
|
|
logging.error("Import returned None (Project not found or GTM DB issue)")
|
|
return {"error": "GTM Project not found or import failed"}
|
|
|
|
logging.info(f"Successfully imported Project. ID: {result.get('id')}")
|
|
return result
|
|
except Exception as e:
|
|
logging.error(f"Exception during import: {e}", exc_info=True)
|
|
return {"error": str(e)}
|
|
|
|
def list_content_projects(payload):
|
|
logging.info("Executing list_content_projects")
|
|
try:
|
|
projects = db_manager.get_all_content_projects()
|
|
logging.debug(f"Found {len(projects)} Content projects")
|
|
return {"projects": projects}
|
|
except Exception as e:
|
|
logging.error(f"Error listing Content projects: {e}", exc_info=True)
|
|
return {"error": str(e), "projects": []}
|
|
|
|
def load_project_details(payload):
|
|
project_id = payload.get('projectId')
|
|
logging.info(f"Loading project details for ID: {project_id}")
|
|
|
|
try:
|
|
project = db_manager.get_content_project(project_id)
|
|
if not project:
|
|
logging.warning(f"Project ID {project_id} not found")
|
|
return {"error": "Project not found"}
|
|
|
|
assets = db_manager.get_project_assets(project_id)
|
|
project['assets'] = assets
|
|
logging.info(f"Loaded project '{project.get('name')}' with {len(assets)} assets.")
|
|
return project
|
|
except Exception as e:
|
|
logging.error(f"Error loading project: {e}", exc_info=True)
|
|
return {"error": str(e)}
|
|
|
|
def seo_brainstorming(payload):
|
|
project_id = payload.get('projectId')
|
|
lang = payload.get('lang', 'de')
|
|
logging.info(f"Starting SEO Brainstorming for Project ID: {project_id}")
|
|
|
|
try:
|
|
project = db_manager.get_content_project(project_id)
|
|
if not project:
|
|
return {"error": "Project context not found"}
|
|
|
|
gtm_data = project.get('gtm_data_snapshot', {})
|
|
logging.debug(f"Loaded GTM context snapshot size: {len(str(gtm_data))} chars")
|
|
|
|
prompt = r"""
|
|
Basierend auf folgendem GTM-Kontext (Strategie für ein technisches Produkt):
|
|
{gtm_context}
|
|
|
|
AUFGABE:
|
|
Generiere eine strategische Liste von 15 SEO-Keywords.
|
|
1. 5 Short-Tail Fokus-Keywords (z.B. Produktkategorie + 'kaufen/mieten').
|
|
2. 10 Long-Tail Keywords, die spezifische Pain Points oder Usecases adressieren (z.B. 'Kostenreduktion bei Sicherheitsrundgängen').
|
|
|
|
Die Keywords müssen für Entscheider relevant sein (CFO, Head of Security, Operations Manager).
|
|
|
|
Output NUR als JSON Liste von Strings.
|
|
""".format(gtm_context=json.dumps(gtm_data))
|
|
|
|
logging.info("Calling Gemini Flash for keywords...")
|
|
response = call_gemini_flash(prompt, system_instruction=get_copywriter_instruction(lang), json_mode=True)
|
|
logging.info("Gemini response received.")
|
|
|
|
keywords = json.loads(response)
|
|
|
|
db_manager.save_seo_strategy(project_id, {"seed_keywords": keywords})
|
|
logging.info("Keywords saved to DB.")
|
|
return {"keywords": keywords}
|
|
except Exception as e:
|
|
logging.error(f"Error in SEO Brainstorming: {e}", exc_info=True)
|
|
return {"error": str(e)}
|
|
|
|
def generate_section(payload):
|
|
project_id = payload.get('projectId')
|
|
section_key = payload.get('sectionKey')
|
|
manual_content = payload.get('manualContent')
|
|
lang = payload.get('lang', 'de')
|
|
keywords = payload.get('keywords', [])
|
|
|
|
logging.info(f"Generating section '{section_key}' for Project {project_id}")
|
|
|
|
if manual_content:
|
|
logging.info("Saving manual content update.")
|
|
try:
|
|
db_manager.save_content_asset(project_id, 'website_section', section_key, f"Section: {section_key}", manual_content, keywords)
|
|
return {"status": "saved", "sectionKey": section_key}
|
|
except Exception as e:
|
|
logging.error(f"Error saving manual content: {e}", exc_info=True)
|
|
return {"error": str(e)}
|
|
|
|
try:
|
|
project = db_manager.get_content_project(project_id)
|
|
if not project:
|
|
return {"error": "Project context not found"}
|
|
|
|
gtm_data = project.get('gtm_data_snapshot', {})
|
|
category = project.get('category')
|
|
|
|
prompt = r"""
|
|
Erstelle den Website-Inhalt für die Sektion '{section}' eines Produkts in der Kategorie '{cat}'.
|
|
|
|
STRATEGIE-KONTEXT:
|
|
{gtm_context}
|
|
|
|
SEO-KEYWORDS ZU NUTZEN:
|
|
{kws}
|
|
|
|
ANFORDERUNG:
|
|
- Schreibe im Stil eines Senior Copywriters (fachlich fundiert, Challenger Sale).
|
|
- Format: Markdown.
|
|
- Die Sektion muss den Nutzer zur nächsten Aktion (CTA) führen.
|
|
""".format(
|
|
section=section_key,
|
|
cat=category,
|
|
gtm_context=json.dumps(gtm_data),
|
|
kws=json.dumps(keywords)
|
|
)
|
|
|
|
logging.info("Calling Gemini for copy generation...")
|
|
content = call_gemini_flash(prompt, system_instruction=get_copywriter_instruction(lang), json_mode=False)
|
|
logging.info("Copy generated.")
|
|
|
|
db_manager.save_content_asset(project_id, 'website_section', section_key, f"Section: {section_key}", content, keywords)
|
|
|
|
return {"content": content, "sectionKey": section_key}
|
|
except Exception as e:
|
|
logging.error(f"Error generating section: {e}", exc_info=True)
|
|
return {"error": str(e)}
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Content Engine Orchestrator")
|
|
parser.add_argument("--mode", required=True)
|
|
parser.add_argument("--payload_file", help="Path to JSON payload")
|
|
|
|
args = parser.parse_args()
|
|
logging.info(f"Orchestrator called with mode: {args.mode}")
|
|
|
|
payload = {}
|
|
if args.payload_file:
|
|
try:
|
|
with open(args.payload_file, 'r') as f:
|
|
payload = json.load(f)
|
|
logging.debug("Payload loaded successfully.")
|
|
except Exception as e:
|
|
logging.error(f"Failed to load payload file: {e}")
|
|
print(json.dumps({"error": "Failed to load payload"}))
|
|
sys.exit(1)
|
|
|
|
modes = {
|
|
"list_gtm_projects": list_gtm_projects,
|
|
"import_project": import_project,
|
|
"list_content_projects": list_content_projects,
|
|
"load_project": load_project_details,
|
|
"seo_brainstorming": seo_brainstorming,
|
|
"generate_section": generate_section,
|
|
}
|
|
|
|
if args.mode in modes:
|
|
try:
|
|
result = modes[args.mode](payload)
|
|
# DUMP RESULT TO STDOUT
|
|
print(json.dumps(result, ensure_ascii=False))
|
|
except Exception as e:
|
|
logging.critical(f"Unhandled exception in mode execution: {e}", exc_info=True)
|
|
print(json.dumps({"error": f"Internal Error: {str(e)}"}))
|
|
else:
|
|
logging.error(f"Unknown mode: {args.mode}")
|
|
print(json.dumps({"error": f"Unknown mode: {args.mode}"}))
|
|
|
|
if __name__ == "__main__":
|
|
main() |