Frank Schwenk
f6e901bad8
Speichert OpenRouter- und SDXL-Prompts pro Archiv-Eintrag und zeigt sie auf der Übersichtsseite in aufklappbaren Details an. Co-authored-by: Cursor <cursoragent@cursor.com>
372 lines
12 KiB
Python
372 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""Daily mobea.de startup page generator."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import random
|
||
import re
|
||
import sys
|
||
from datetime import date, datetime, timezone
|
||
from pathlib import Path
|
||
|
||
import requests
|
||
from jinja2 import Environment, FileSystemLoader
|
||
|
||
import topics
|
||
|
||
GERMAN_MONTHS = [
|
||
"Januar",
|
||
"Februar",
|
||
"März",
|
||
"April",
|
||
"Mai",
|
||
"Juni",
|
||
"Juli",
|
||
"August",
|
||
"September",
|
||
"Oktober",
|
||
"November",
|
||
"Dezember",
|
||
]
|
||
|
||
THEMES = ["corporate", "neon", "mystic", "brutalist", "pastel"]
|
||
|
||
REQUIRED_LLM_KEYS = [
|
||
"backronym_words",
|
||
"tagline",
|
||
"subtitle",
|
||
"pitch",
|
||
"features",
|
||
"pricing",
|
||
"testimonials",
|
||
"faq",
|
||
"cta",
|
||
"stats",
|
||
"image_prompt",
|
||
"image_negative_prompt",
|
||
]
|
||
|
||
LLM_SCHEMA = """{
|
||
"backronym_words": ["Modulare", "Orbitale", "Bewusstseins-", "Energie-", "Allianz"],
|
||
"tagline": "kurzer knackiger Claim",
|
||
"subtitle": "1-2 Sätze Hero-Untertitel",
|
||
"pitch": "Absatz 'Über uns', 3-5 Sätze, völlig ernst im Ton",
|
||
"features": [{"icon": "🔮", "title": "...", "text": "1-2 Sätze"}],
|
||
"pricing": [{"name": "Starter", "price": "9,99 €", "period": "/Monat", "features": ["...","...","..."], "cta": "Jetzt starten"}],
|
||
"testimonials": [{"quote": "...", "author": "Vorname N.", "role": "Beruf, Stadt"}],
|
||
"faq": [{"q": "...", "a": "..."}],
|
||
"cta": "Haupt-Button-Text",
|
||
"stats": [{"value": "98,3%", "label": "..."}],
|
||
"image_prompt": "english SDXL prompt for the hero image, no text in image",
|
||
"image_negative_prompt": "english negative prompt"
|
||
}"""
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
|
||
|
||
def log(message: str) -> None:
|
||
ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
print(f"{ts} {message}", flush=True)
|
||
|
||
|
||
def date_human(day: date) -> str:
|
||
return f"{day.day}. {GERMAN_MONTHS[day.month - 1]} {day.year}"
|
||
|
||
|
||
def load_archive(data_dir: Path) -> list[dict]:
|
||
path = data_dir / "archive.json"
|
||
if not path.exists():
|
||
return []
|
||
try:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
if isinstance(data, list):
|
||
return data
|
||
log(f"warning: {path} is not a list, starting fresh")
|
||
except (json.JSONDecodeError, OSError) as exc:
|
||
log(f"warning: could not read {path}: {exc}, starting fresh")
|
||
return []
|
||
|
||
|
||
def save_json_atomic(path: Path, data: object) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
tmp = path.with_suffix(path.suffix + ".tmp")
|
||
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
os.replace(tmp, path)
|
||
|
||
|
||
def write_text_atomic(path: Path, content: str) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
tmp = path.with_suffix(path.suffix + ".tmp")
|
||
tmp.write_text(content, encoding="utf-8")
|
||
os.replace(tmp, path)
|
||
|
||
|
||
def strip_json_fences(text: str) -> str:
|
||
stripped = text.strip()
|
||
if stripped.startswith("```"):
|
||
stripped = re.sub(r"^```(?:json)?\s*", "", stripped)
|
||
stripped = re.sub(r"\s*```$", "", stripped)
|
||
return stripped.strip()
|
||
|
||
|
||
def validate_llm_json(data: dict) -> None:
|
||
missing = [key for key in REQUIRED_LLM_KEYS if key not in data]
|
||
if missing:
|
||
raise ValueError(f"missing required keys: {', '.join(missing)}")
|
||
|
||
if len(data.get("backronym_words", [])) != 5:
|
||
log("warning: backronym_words should contain exactly 5 words")
|
||
|
||
features = data.get("features", [])
|
||
if not (3 <= len(features) <= 4):
|
||
log(f"warning: expected 3-4 features, got {len(features)}")
|
||
|
||
pricing = data.get("pricing", [])
|
||
if len(pricing) != 3:
|
||
log(f"warning: expected exactly 3 pricing tiers, got {len(pricing)}")
|
||
elif pricing[-1].get("name") != "Enterprise":
|
||
log("warning: last pricing tier should be named 'Enterprise'")
|
||
elif pricing[-1].get("price") != "Auf Anfrage":
|
||
log("warning: Enterprise tier price should be 'Auf Anfrage'")
|
||
elif pricing[-1].get("cta") != "Kontaktieren Sie uns":
|
||
log("warning: Enterprise tier cta should be 'Kontaktieren Sie uns'")
|
||
|
||
testimonials = data.get("testimonials", [])
|
||
if not (2 <= len(testimonials) <= 3):
|
||
log(f"warning: expected 2-3 testimonials, got {len(testimonials)}")
|
||
|
||
faq = data.get("faq", [])
|
||
if not (3 <= len(faq) <= 4):
|
||
log(f"warning: expected 3-4 faq entries, got {len(faq)}")
|
||
|
||
stats = data.get("stats", [])
|
||
if len(stats) != 3:
|
||
log(f"warning: expected exactly 3 stats, got {len(stats)}")
|
||
|
||
|
||
def build_system_prompt() -> str:
|
||
return (
|
||
"Du bist Texter für eine Satire-Website namens mobea.de – "
|
||
"'das Startup, das jeden Tag ein anderes Startup ist'. "
|
||
"Schreibe ausschließlich auf Deutsch (außer image_prompt und image_negative_prompt). "
|
||
"Der Humor entsteht durch absurde Inhalte in völlig ernstem Startup-Marketing-Ton. "
|
||
"Antworte ausschließlich mit validem JSON nach diesem Schema:\n"
|
||
f"{LLM_SCHEMA}\n\n"
|
||
"Pflichtregeln: backronym_words = genau 5 Wörter, deren Anfangsbuchstaben "
|
||
"zusammen MOBEA ergeben; 3-4 features; genau 3 pricing tiers, der letzte "
|
||
"name 'Enterprise', price 'Auf Anfrage', cta 'Kontaktieren Sie uns'; "
|
||
"2-3 testimonials; 3-4 faq; genau 3 stats."
|
||
)
|
||
|
||
|
||
def build_user_prompt(selected_topics: list[dict]) -> str:
|
||
lines = ["Themen des Tages:"]
|
||
for topic in selected_topics:
|
||
lines.append(f"- {topic['name']}: {topic['hint']}")
|
||
lines.append(
|
||
"Erfinde ein völlig neues, absurd-satirisches Startup im todernsten "
|
||
"Marketing-Ton. Alle Texte müssen wie echtes Startup-Deutsch klingen."
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
|
||
def call_openrouter(api_key: str, model: str, selected_topics: list[dict]) -> dict:
|
||
payload = {
|
||
"model": model,
|
||
"temperature": 0.9,
|
||
"response_format": {"type": "json_object"},
|
||
"messages": [
|
||
{"role": "system", "content": build_system_prompt()},
|
||
{"role": "user", "content": build_user_prompt(selected_topics)},
|
||
],
|
||
}
|
||
response = requests.post(
|
||
"https://openrouter.ai/api/v1/chat/completions",
|
||
headers={
|
||
"Authorization": f"Bearer {api_key}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
json=payload,
|
||
timeout=120,
|
||
)
|
||
response.raise_for_status()
|
||
body = response.json()
|
||
content = body["choices"][0]["message"]["content"]
|
||
data = json.loads(strip_json_fences(content))
|
||
validate_llm_json(data)
|
||
return data
|
||
|
||
|
||
def generate_hero_image(
|
||
api_key: str,
|
||
site_dir: Path,
|
||
date_iso: str,
|
||
prompt: str,
|
||
negative_prompt: str,
|
||
) -> str | None:
|
||
try:
|
||
response = requests.post(
|
||
"https://gateway.pixazo.ai/getImage/v1/getSDXLImage",
|
||
headers={
|
||
"Content-Type": "application/json",
|
||
"Cache-Control": "no-cache",
|
||
"Ocp-Apim-Subscription-Key": api_key,
|
||
},
|
||
json={
|
||
"prompt": prompt,
|
||
"negative_prompt": negative_prompt,
|
||
"height": 1024,
|
||
"width": 1024,
|
||
"num_steps": 20,
|
||
"guidance_scale": 5,
|
||
},
|
||
timeout=180,
|
||
)
|
||
response.raise_for_status()
|
||
body = response.json()
|
||
image_url = body.get("imageUrl") or body.get("output")
|
||
if not image_url:
|
||
log("warning: Pixazo response missing imageUrl/output")
|
||
return None
|
||
|
||
img_response = requests.get(image_url, timeout=120)
|
||
img_response.raise_for_status()
|
||
|
||
img_dir = site_dir / "img"
|
||
img_dir.mkdir(parents=True, exist_ok=True)
|
||
img_path = img_dir / f"{date_iso}.jpg"
|
||
img_path.write_bytes(img_response.content)
|
||
return f"/img/{date_iso}.jpg"
|
||
except (requests.RequestException, OSError, KeyError, TypeError, ValueError) as exc:
|
||
log(f"warning: hero image generation failed: {exc}")
|
||
return None
|
||
|
||
|
||
def render_template(name: str, context: dict) -> str:
|
||
env = Environment(
|
||
loader=FileSystemLoader(SCRIPT_DIR / "templates"),
|
||
autoescape=True,
|
||
)
|
||
return env.get_template(name).render(**context)
|
||
|
||
|
||
def build_archiv_entries(archive: list[dict]) -> list[dict]:
|
||
entries = []
|
||
for item in sorted(archive, key=lambda e: e["date"], reverse=True):
|
||
prompts = item.get("prompts")
|
||
entries.append(
|
||
{
|
||
"date": item["date"],
|
||
"date_human": item["date_human"],
|
||
"backronym": item["backronym"],
|
||
"tagline": item["tagline"],
|
||
"url": f"/archiv/{item['date']}.html",
|
||
"image": f"/img/{item['date']}.jpg" if item.get("has_image") else None,
|
||
"topics": item.get("topics", []),
|
||
"prompts": prompts if isinstance(prompts, dict) else None,
|
||
}
|
||
)
|
||
return entries
|
||
|
||
|
||
def main() -> int:
|
||
api_key = os.environ.get("OPENROUTER_API_KEY")
|
||
model = os.environ.get("OPENROUTER_MODEL")
|
||
pixazo_key = os.environ.get("PIXAZO_API_KEY")
|
||
site_dir = Path(os.environ.get("SITE_DIR", "/site"))
|
||
data_dir = Path(os.environ.get("DATA_DIR", "/data"))
|
||
generate_hour = int(os.environ.get("GENERATE_HOUR_UTC", "5"))
|
||
|
||
if not api_key:
|
||
log("error: OPENROUTER_API_KEY is required")
|
||
return 1
|
||
if not model:
|
||
log("error: OPENROUTER_MODEL is required")
|
||
return 1
|
||
|
||
now = datetime.now(timezone.utc)
|
||
today = now.date()
|
||
date_iso = today.isoformat()
|
||
index_path = site_dir / "index.html"
|
||
|
||
archive = load_archive(data_dir)
|
||
if any(entry.get("date") == date_iso for entry in archive) and index_path.exists():
|
||
log("up to date")
|
||
return 0
|
||
|
||
if index_path.exists() and now.hour < generate_hour:
|
||
log(f"waiting until {generate_hour:02d}:00 UTC (current hour: {now.hour})")
|
||
return 0
|
||
|
||
selected_topics = topics.pick_topics(today)
|
||
theme = random.Random(f"theme-{date_iso}").choice(THEMES)
|
||
log(f"generating edition for {date_iso} theme={theme} topics={[t['name'] for t in selected_topics]}")
|
||
|
||
try:
|
||
llm_json = call_openrouter(api_key, model, selected_topics)
|
||
except (requests.RequestException, json.JSONDecodeError, KeyError, ValueError) as exc:
|
||
log(f"error: OpenRouter call failed: {exc}")
|
||
return 1
|
||
|
||
hero_image = None
|
||
if pixazo_key:
|
||
hero_image = generate_hero_image(
|
||
pixazo_key,
|
||
site_dir,
|
||
date_iso,
|
||
llm_json["image_prompt"],
|
||
llm_json["image_negative_prompt"],
|
||
)
|
||
else:
|
||
log("PIXAZO_API_KEY not set, skipping hero image")
|
||
|
||
human_date = date_human(today)
|
||
startup_context = {
|
||
"date": date_iso,
|
||
"date_human": human_date,
|
||
"theme": theme,
|
||
"hero_image": hero_image,
|
||
"s": llm_json,
|
||
}
|
||
startup_html = render_template("startup.html.j2", startup_context)
|
||
|
||
archiv_day_path = site_dir / "archiv" / f"{date_iso}.html"
|
||
write_text_atomic(index_path, startup_html)
|
||
write_text_atomic(archiv_day_path, startup_html)
|
||
log(f"wrote {index_path} and {archiv_day_path}")
|
||
|
||
archive_entry = {
|
||
"date": date_iso,
|
||
"date_human": human_date,
|
||
"backronym": " ".join(llm_json["backronym_words"]),
|
||
"tagline": llm_json["tagline"],
|
||
"theme": theme,
|
||
"topics": [topic["name"] for topic in selected_topics],
|
||
"has_image": hero_image is not None,
|
||
"prompts": {
|
||
"content_system": build_system_prompt(),
|
||
"content_user": build_user_prompt(selected_topics),
|
||
"image_prompt": llm_json["image_prompt"],
|
||
"image_negative_prompt": llm_json["image_negative_prompt"],
|
||
},
|
||
}
|
||
archive = [entry for entry in archive if entry.get("date") != date_iso]
|
||
archive.append(archive_entry)
|
||
save_json_atomic(data_dir / "archive.json", archive)
|
||
log(f"updated {data_dir / 'archive.json'}")
|
||
|
||
archiv_html = render_template(
|
||
"archiv.html.j2",
|
||
{"entries": build_archiv_entries(archive)},
|
||
)
|
||
write_text_atomic(site_dir / "archiv" / "index.html", archiv_html)
|
||
log(f"wrote {site_dir / 'archiv' / 'index.html'}")
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|