#!/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 from images import create_responsive_variants, load_image_descriptor 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, ) -> dict[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() if len(img_response.content) < 50_000: log( f"warning: hero image for {date_iso} too small " f"({len(img_response.content)} bytes), discarding" ) return None 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 create_responsive_variants(img_dir, date_iso) 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], site_dir: Path) -> list[dict]: img_dir = site_dir / "img" 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": load_image_descriptor(img_dir, item["date"]) 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_base = { "date": date_iso, "date_human": human_date, "theme": theme, "hero_image": hero_image, "s": llm_json, "site_url": os.environ.get("SITE_URL", "https://mobea.de"), } startup_html = render_template( "startup.html.j2", {**startup_base, "canonical_path": "/"}, ) archiv_day_html = render_template( "startup.html.j2", {**startup_base, "canonical_path": f"/archiv/{date_iso}.html"}, ) archiv_day_path = site_dir / "archiv" / f"{date_iso}.html" write_text_atomic(index_path, startup_html) write_text_atomic(archiv_day_path, archiv_day_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'}") latest_image = next( ( f"/img/{entry['date']}.jpg" for entry in sorted(archive, key=lambda e: e["date"], reverse=True) if entry.get("has_image") ), "/og-default.jpg", ) archiv_html = render_template( "archiv.html.j2", { "entries": build_archiv_entries(archive, site_dir), "site_url": os.environ.get("SITE_URL", "https://mobea.de"), "og_image": latest_image, }, ) 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())