#!/usr/bin/env python3
"""Article Index Server — with IDs, meta extraction, image upload, and multi-threading.
v2: Added 'Markierte veröffentlichen' (Joomla API publish), article-image association,
and automatic VG-Wort tracking pixel injection.
v3: Added NanoBanana Image Generator (/nanobanana)."""
import os
import re
import glob
import json
import base64
import hashlib
import socket
import urllib.parse
import urllib.request
import io
import email.parser
import tempfile
import subprocess
import time
from http.server import HTTPServer, BaseHTTPRequestHandler
from datetime import datetime
from socketserver import ThreadingMixIn
WORKSPACE = "/home/hermes/workspace"
MEDIA_DIR = os.path.join(WORKSPACE, "media")
os.makedirs(MEDIA_DIR, exist_ok=True)
PORT = int(os.environ.get("ARTICLE_SERVER_PORT", "8890"))
ACCESS_TOKEN = "br3ttsp1el-n3ws-2026"
USERNAME = "DK-Adminchef"
PASSWORD = "Mcik7%vbdhXa_"
# ── BSN Article CSS (aus Joomla bsn-custom.css) ───────────────────────────
# Wird in Artikel-HTML injiziert, damit bsn-Klassen auch auf dem
# Article-Server gerendert werden (nicht nur im Joomla-Template).
BSN_ARTICLE_CSS = """
/* ── Basis-Typografie (Article-Server) ── */
body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, Helvetica, Arial, sans-serif; font-size: 16px; line-height: 1.7; color: #3B3B3C; background: #fff; max-width: 720px; margin: 0 auto; padding: 2rem 1.5rem; -webkit-font-smoothing: antialiased; }
article { display: block; }
article h1, article h1.title { font-size: 2rem; font-weight: 800; color: #171D4F; margin: 0 0 0.5rem 0; line-height: 1.2; letter-spacing: -0.02em; }
article h2 { font-size: 1.4rem; font-weight: 700; color: #171D4F; margin: 2rem 0 0.75rem; }
article h3 { font-size: 1.15rem; font-weight: 700; color: #20228A; margin: 1.5rem 0 0.5rem; }
article p { margin: 0 0 1rem 0; }
article a { color: #20228A; text-decoration: underline; text-underline-offset: 2px; }
article a:hover { color: #161e6e; }
article ul, article ol { margin: 0.5rem 0 1rem 1.5rem; }
article li { margin-bottom: 0.4rem; }
article img { max-width: 100%; height: auto; border-radius: 8px; margin: 1.5rem 0; }
article hr { border: none; border-top: 1px solid #e0e0e0; margin: 2rem 0; }
/* ── Meta / Dachzeile ── */
.bsn-meta, .meta, p.meta { font-size: 0.85rem; color: #999; margin-bottom: 1.5rem; }
.bsn-dachzeile, .dachzeile, p.dachzeile { font-size: 1.05rem; color: #555; margin-bottom: 1.5rem; font-weight: 500; }
/* ── Infobox ── */
.bsn-infobox { background: #f2f2f2; border: 1px solid #d0d8e0; border-left: 4px solid #20228A; padding: 1rem 1.5rem; margin: 1.5rem 0; border-radius: 16px; font-size: 0.95rem; }
.bsn-infobox h3 { margin-top: 0; color: #20228A; font-family: Arial, Helvetica, sans-serif; font-size: 1.15rem; }
.bsn-infobox table { width: 100%; border-collapse: collapse; }
.bsn-infobox td { padding: 0.3rem 0.5rem; border-bottom: 1px solid #dce4ec; }
.bsn-infobox tr:last-child td { border-bottom: none; }
.bsn-infobox td:first-child { font-weight: 600; color: #171D4F; width: 160px; white-space: nowrap; }
/* ── Tabellen ── */
.bsn-table { width: 100%; border-collapse: collapse; margin: 1rem 0; }
.bsn-table td { padding: 0.5rem 0.8rem; border-bottom: 1px solid #e8e8f0; }
.bsn-table tr:last-child td { border-bottom: none; }
.bsn-table td:first-child { font-weight: 600; color: #171D4F; width: 30%; }
/* ── Gamecard ── */
.bsn-gamecard { background: #f2f2f2; border: 1px solid #e0e3e8; border-left: 4px solid #20228A; padding: 1rem 1.5rem; margin: 1.2rem 0; border-radius: 16px; transition: transform 0.2s ease, box-shadow 0.2s ease; }
.bsn-gamecard h3 { margin-top: 0; color: #171D4F; font-family: Arial, Helvetica, sans-serif; font-size: 1.15rem; }
.bsn-gamecard .game-meta { font-size: 0.85rem; color: #888; margin-bottom: 0.5rem; }
@media (hover: hover) { .bsn-gamecard:hover { transform: translateY(-3px); box-shadow: 0 6px 20px rgba(32,34,138,0.12); border-color: #c8cde0; } }
/* ── Quellen / Recherche ── */
.bsn-quellen { margin-top: 2rem; font-size: 0.9rem; }
.bsn-quellen h3 { color: #171D4F; font-family: Arial, Helvetica, sans-serif; font-size: 1.1rem; margin-bottom: 0.5rem; }
.bsn-quellen ul { padding-left: 1.3rem; margin: 0; }
.bsn-quellen li { margin-bottom: 0.3rem; }
.bsn-recherche, .recherche, p.bsn-recherche { margin-top: 2.5rem; padding-top: 1rem; border-top: 1px solid #ddd; font-size: 0.9rem; color: #777; }
/* ── Blockquote ── */
blockquote { border-left: 4px solid #20228A; background: #f8f9ff; padding: 0.8rem 1.2rem; margin: 1.2rem 0; color: #3B3B3C; font-style: italic; border-radius: 0 8px 8px 0; }
blockquote p { margin-bottom: 0.25rem; }
/* ── System-Readmore (Joomla) ── */
hr#system-readmore { border: none; border-top: 2px dashed #e0e0e0; margin: 1.5rem 0; }
/* ── Quellen (alte Klasse) ── */
.quellen { margin-top: 2rem; font-size: 0.9rem; }
.quellen h3, .quellen strong { color: #171D4F; font-size: 1.1rem; display: block; margin-bottom: 0.5rem; }
.quellen ol, .quellen ul { padding-left: 1.3rem; margin: 0.5rem 0; }
.quellen li { margin-bottom: 0.3rem; }
@media (prefers-color-scheme: dark) {
body { background: #1a1a2e; color: #c8c8d8; }
article h1, article h2 { color: #e0e0f0; }
article h3 { color: #6b6bcc; }
.bsn-infobox, .bsn-gamecard { background: #252530; border-color: #3a3a45; }
.bsn-infobox td, .bsn-table td { border-bottom-color: #3a3a45; }
.bsn-infobox td:first-child, .bsn-table td:first-child { color: #a0a0c0; }
.bsn-infobox h3 { color: #6b6bcc; }
.bsn-gamecard h3 { color: #c8c8d8; }
.bsn-gamecard .game-meta { color: #888; }
.bsn-quellen h3, .quellen h3, .quellen strong { color: #c8c8d8; }
.bsn-quellen a, .quellen a { color: #6b6bcc; }
.bsn-meta, .meta, p.meta { color: #777; }
.bsn-recherche, .recherche, p.bsn-recherche { color: #888; border-top-color: #3a3a45; }
.bsn-dachzeile, .dachzeile, p.dachzeile { color: #999; border-bottom-color: #3a3a45; }
.bsn-gamecard:hover { box-shadow: 0 6px 20px rgba(100,100,200,0.15); border-color: #505068; }
blockquote { background: #252530; border-left-color: #6b6bcc; color: #c0c0d0; }
hr#system-readmore { border-top-color: #3a3a45; }
hr { border-top-color: #3a3a45; }
}
"""
# ── Joomla API Configuration ──────────────────────────────────────────────
JOOMLA_API_URL = "https://www.brettspiel-news.de/api/index.php/v1/content/articles"
# JOOMLA_TOKEN is read from environment or from a token file
JOOMLA_TOKEN = os.environ.get("JOOMLA_API_TOKEN", "")
if not JOOMLA_TOKEN:
token_paths = [
"/tmp/alicia_skills/KEYS.txt",
"/home/hermes/.hermes/joomla_token.txt",
]
for tp in token_paths:
if os.path.exists(tp):
with open(tp, "r") as f:
for line in f:
line = line.strip()
if line and not line.startswith("#"):
JOOMLA_TOKEN = line
break
if JOOMLA_TOKEN:
break
# ── VG-Wort Configuration ─────────────────────────────────────────────────
VG_WORT_COUNTER_ID = os.environ.get("VG_WORT_COUNTER_ID", "")
VG_WORT_PIXEL_HTML = (
f'
'
) if VG_WORT_COUNTER_ID else ""
# ── Article-Image Mapping ─────────────────────────────────────────────────
IMAGE_MAP_FILE = os.path.join(WORKSPACE, "article_images.json")
MONTHS = {
"Januar": 1, "Februar": 2, "März": 3, "April": 4,
"Mai": 5, "Juni": 6, "Juli": 7, "August": 8,
"September": 9, "Oktober": 10, "November": 11, "Dezember": 12
}
# ── Gemini Flash Image API Configuration ────────────────────────────────────
GEMINI_IMAGE_MODEL = "gemini-2.5-flash-image"
GEMINI_API_URL = (
"https://generativelanguage.googleapis.com/v1beta/"
"models/" + GEMINI_IMAGE_MODEL + ":generateContent"
)
GEMINI_KEY_SECRET = "bsn/google-imagen-key"
def _get_gemini_key() -> str:
"""Google Gemini API-Key aus bsn-secrets lesen."""
try:
r = subprocess.run(
["/home/hermes/bin/bsn-secrets", "get", GEMINI_KEY_SECRET],
capture_output=True, text=True, timeout=5
)
return r.stdout.strip()
except Exception:
return ""
def _detect_image_mime(image_bytes: bytes) -> str:
"""MIME-Type anhand Magic-Bytes erkennen."""
if len(image_bytes) < 8:
return "image/jpeg" # Fallback
if image_bytes[:4] == b"\x89PNG":
return "image/png"
if image_bytes[:2] == b"\xff\xd8":
return "image/jpeg"
if image_bytes[:4] == b"RIFF" and image_bytes[8:12] == b"WEBP":
return "image/webp"
if image_bytes[:4] in (b"ftyp", b"heic", b"heix", b"hevc", b"hevx"):
return "image/heic"
return "image/jpeg" # Fallback
def _gemini_generate_image(
prompt: str,
image_base64: str | None,
aspect_ratio: str,
resolution: str
) -> dict:
"""Gemini Flash Image aufrufen — Text-to-Image und Image-to-Image.
Args:
prompt: Text-Prompt.
image_base64: Optionales Referenzbild als Base64 (Image-to-Image).
aspect_ratio: Seitenverhältnis (z.B. "16:9").
resolution: Behält Kompatibilität (von Gemini ignoriert).
Returns:
dict mit "image_base64" (PNG) oder "error".
"""
import base64 as b64
key = _get_gemini_key()
if not key:
return {"error": "Kein Gemini API-Key"}
parts = [{"text": f"{prompt}\n\nAspect ratio: {aspect_ratio}."}]
if image_base64:
# MIME-Type aus den tatsächlichen Bilddaten erkennen
raw_bytes = b64.b64decode(image_base64)
mime_type = _detect_image_mime(raw_bytes)
parts.append({
"inlineData": {
"mimeType": mime_type,
"data": image_base64
}
})
body = json.dumps({
"contents": [{"parts": parts}],
"generationConfig": {
"responseModalities": ["image", "text"]
}
}).encode()
req = urllib.request.Request(GEMINI_API_URL, data=body, method="POST")
req.add_header("x-goog-api-key", key)
req.add_header("Content-Type", "application/json")
try:
with urllib.request.urlopen(req, timeout=45) as resp:
result = json.loads(resp.read())
candidates = result.get("candidates") or []
for cand in candidates:
for part in cand.get("content", {}).get("parts", []):
inline = part.get("inlineData")
if inline and inline.get("data"):
return {"image_base64": inline["data"]}
return {"error": "Kein Bild in Gemini-Antwort"}
except urllib.error.HTTPError as e:
return {"error": f"HTTP {e.code}: {e.read().decode()[:300]}"}
def _get_unpublished_articles() -> list[dict]:
"""Liste aller unveröffentlichten Artikel (ohne joomla-id)."""
articles = []
for f in sorted(glob.glob(os.path.join(WORKSPACE, "*.html")), key=os.path.getmtime, reverse=True):
name = os.path.basename(f)
if any(re.search(p, name) for p in SKIP_PATTERNS):
continue
try:
with open(f, "r") as fh:
content = fh.read(5000)
has_joomla_id = '(.*?)', content)
title = title_match.group(1).strip() if title_match else name
articles.append({"filename": name, "title": title})
except Exception:
pass
return articles[:50] # Max 50
SKIP_PATTERNS = [
r'^ddg\\d*\\.html', r'^google', r'^bgg_(search|thread|game|price)',
r'^bk_', r'^backerkit', r'^brave_search', r'^fb_post', r'^reddit',
r'^rp_mg', r'^rp_gladbach', r'^spiegel', r'^sportschau', r'^transfermarkt',
r'^tm_', r'^kicker', r'^brettspielbox', r'^bsg_', r'^bbb_',
r'^envyborn', r'^brettspiel_news_newsletter', r'^boardgamewire',
r'^ddg_', r'^blocked', r'^error', r'^just.a.moment',
r'^duckduckgo', r'^update.regarding',
r'kennerspiele_deals', r'kombi_deals', r'top_deals',
r'kinderspiele_deals', r'amazon_deals', r'amazon_primeday',
]
# ── Image Mapping Helpers ─────────────────────────────────────────────────
def get_image_map() -> dict[str, list[str]]:
"""Load article→image mapping from JSON."""
if os.path.exists(IMAGE_MAP_FILE):
with open(IMAGE_MAP_FILE, "r") as f:
return json.load(f)
return {}
def save_image_map(mapping: dict[str, list[str]]) -> None:
"""Save article→image mapping to JSON."""
with open(IMAGE_MAP_FILE, "w") as f:
json.dump(mapping, f, indent=2)
def find_orphan_images() -> list[str]:
"""Find images in workspace that are NOT in article_images.json."""
import glob as _glob
existing = set()
for paths in get_image_map().values():
for p in paths:
existing.add(os.path.abspath(p))
orphans = []
for ext in ("*.jpg", "*.jpeg", "*.png", "*.webp", "*.gif"):
for p in _glob.iglob(os.path.join(WORKSPACE, "**", ext), recursive=True):
# Skip bsn-chatbot, media dir (already managed), __pycache__, .git, venv
if "/bsn-chatbot/" in p or "/__pycache__/" in p or "/.git/" in p or "/venv/" in p:
continue
if "/image_cache/" in p:
continue
abspath = os.path.abspath(p)
if abspath not in existing:
orphans.append(abspath)
return sorted(orphans)
def is_article(filename: str) -> bool:
for pattern in SKIP_PATTERNS:
if re.search(pattern, filename, re.IGNORECASE):
return False
return filename.endswith(".html")
def extract_meta(filepath: str) -> dict | None:
"""Extract title, date, meta description, keywords, and Joomla ID from an article."""
with open(filepath, "r", encoding="utf-8") as f:
content = f.read()
title_match = re.search(r'
([^<]+)', content)
if not title_match:
return None
title = title_match.group(1)
date_match = re.search(r'[^<]*?(\d{1,2}\.\s*(?:Januar|Februar|März|April|Mai|Juni|Juli|August|September|Oktober|November|Dezember)\s*\d{4})', content)
if not date_match:
return None
date_str = date_match.group(1)
try:
parts = date_str.replace(".", "").split()
day, month_name, year = int(parts[0]), parts[1], int(parts[2])
month = MONTHS.get(month_name, 1)
sort_key = f"{year:04d}-{month:02d}-{day:02d}"
display_date = f"{day}.{month_name} {year}"
except (ValueError, IndexError):
return None
meta_desc = ""
desc_match = re.search(r' list[dict]:
articles = []
for f in glob.glob(os.path.join(WORKSPACE, "*.html")):
filename = os.path.basename(f)
if filename == "index.html":
continue
if not is_article(filename):
continue
meta = extract_meta(f)
if meta:
articles.append(meta)
articles.sort(key=lambda a: (a["date_sort"], a["filename"]), reverse=True)
for i, a in enumerate(articles, 1):
a["local_id"] = i
return articles
def build_index(articles: list[dict]) -> str:
by_date_sort: dict[str, list[dict]] = {}
for a in articles:
by_date_sort.setdefault(a["date_sort"], []).append(a)
month_names = {"01":"Januar","02":"Februar","03":"März","04":"April","05":"Mai","06":"Juni",
"07":"Juli","08":"August","09":"September","10":"Oktober","11":"November","12":"Dezember"}
# Read image map for badge display
img_map = get_image_map()
items_html = ""
for sort_key in sorted(by_date_sort.keys(), reverse=True):
day_articles = by_date_sort[sort_key]
try:
y, m, d = sort_key.split("-")
nice_date = f"{int(d)}. {month_names.get(m, m)} {y}"
except ValueError:
nice_date = day_articles[0]["date_display"]
items_html += f'
\n'
items_html += f'\n'
for a in day_articles:
jid = a.get("joomla_id", "")
local_id = a.get("local_id", "")
id_str = f"#{jid}" if jid else f"#{local_id}"
id_class = "id-joomla" if jid else "id-local"
meta_desc = a.get("meta_desc", "")
meta_tags = a.get("meta_tags", "")
meta_preview = ""
if meta_desc or meta_tags:
meta_parts = []
if meta_desc:
meta_parts.append(meta_desc[:120] + ("…" if len(meta_desc) > 120 else ""))
if meta_tags:
meta_parts.append(f'
🏷 {meta_tags[:60]}{"…" if len(meta_tags) > 60 else ""}')
meta_preview = '
' + " · ".join(meta_parts) + '
'
# Image badge
image_badge = ""
image_ok = False
if a["filename"] in img_map and img_map[a["filename"]]:
img_count = len(img_map[a["filename"]])
image_badge = f'
🖼{img_count}'
image_ok = True
# Readiness indicators
meta_desc_ok = bool(meta_desc)
meta_tags_ok = bool(meta_tags)
all_ready = meta_desc_ok and meta_tags_ok and image_ok
def ready_dot(ok: bool, title: str) -> str:
cls = "ready-ok" if ok else "ready-missing"
symbol = "●" if ok else "○"
return f'
{symbol}'
readiness = (
ready_dot(meta_desc_ok, "Meta-Description") +
ready_dot(meta_tags_ok, "Keywords") +
ready_dot(image_ok, "Bild")
)
items_html += f"""
\n"""
items_html += "
\n"
count = len(articles)
has_token = bool(JOOMLA_TOKEN)
token_status = "🟢 API-Token vorhanden" if has_token else "🔴 Kein API-Token — Veröffentlichung deaktiviert"
publish_disabled = "" if has_token else "disabled"
publish_title = "" if has_token else ' title="Kein Joomla API-Token konfiguriert (JOOMLA_API_TOKEN env oder /tmp/alicia_skills/KEYS.txt)"'
return f"""
Artikel-Übersicht · brettspiel-news.de
📂 Rubriken (Kategorien) — zum Nachschlagen · 📌 Öffentliche = 3 Tage Hauptartikel
61 Korrektur 🔒 Redaktion
9 Nachrichten
└ 64 Angebot
└ 65 Crowdfunding
└ 66 Branche
└ 67 Neuerscheinungen
└ 68 Ankündigungen
└ 69 Zubehör
8 Brettspieltest
11 Magazin
10 Video
55 Podcast
0 markiert
📷 Bild hochladen
Ziehe Bilder hierher oder klicke zum Auswählen.
Das Bild wird automatisch dem ersten markierten Artikel zugeordnet.
Oder nutze das 📷 Icon direkt neben dem gewünschten Artikel.
{items_html}
"""
class ThreadingArticleServer(ThreadingMixIn, HTTPServer):
"""HTTPServer with threading support — handles multiple requests concurrently."""
daemon_threads = True
allow_reuse_address = True
class ArticleHandler(BaseHTTPRequestHandler):
timeout = 30
def _check_auth(self) -> bool:
auth_header = self.headers.get("Authorization", "")
if not auth_header.startswith("Basic "):
self._send_auth_required()
return False
try:
creds = base64.b64decode(auth_header[6:]).decode("utf-8")
username, password = creds.split(":", 1)
except Exception:
self._send_auth_required()
return False
if username == USERNAME and password == PASSWORD:
return True
self._send_auth_required()
return False
def _send_auth_required(self):
self.send_response(401)
self.send_header("WWW-Authenticate", 'Basic realm="Artikel-Archiv"')
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
html = """
Zugriff verweigert
Zugriff verweigert
Bitte einloggen, um die Artikel-Übersicht zu sehen.
"""
self.wfile.write(html.encode("utf-8"))
def _check_token(self) -> bool:
parsed = urllib.parse.urlparse(self.path)
params = urllib.parse.parse_qs(parsed.query)
key = params.get("key", [None])[0]
if key == ACCESS_TOKEN:
return True
self.send_response(403)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
html = "Zugriff verweigertZugriff verweigert
Diese Seite ist nicht öffentlich.
"
self.wfile.write(html.encode("utf-8"))
return False
def do_GET(self):
try:
self._do_GET_impl()
except (ConnectionResetError, BrokenPipeError, socket.timeout):
pass
except Exception as e:
try:
self.send_response(500)
self.end_headers()
self.wfile.write(b"Internal Server Error")
except Exception:
pass
print(f"[ERROR] {self.client_address} - {e}", flush=True)
def _do_GET_impl(self):
if not self._check_token():
return
parsed = urllib.parse.urlparse(self.path)
path = parsed.path
if path == "/" or path == "/index.html":
articles = get_articles()
html = build_index(articles)
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
self.wfile.write(html.encode("utf-8"))
elif path == "/nanobanana":
html = self._build_nanobanana_page()
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
self.wfile.write(html.encode("utf-8"))
elif path == "/nanobanana/articles":
self._handle_nanobanana_articles()
elif path.startswith("/media/"):
rel = path[len("/"):]
filepath = os.path.normpath(os.path.join(WORKSPACE, rel))
if os.path.isfile(filepath) and filepath.startswith(WORKSPACE):
with open(filepath, "rb") as f:
data = f.read()
self.send_response(200)
self.send_header("Content-Length", str(len(data)))
ext = os.path.splitext(filepath)[1].lower()
mime_map = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png",
".webp": "image/webp", ".gif": "image/gif", ".svg": "image/svg+xml"}
self.send_header("Content-Type", mime_map.get(ext, "application/octet-stream"))
self.send_header("Cache-Control", "public, max-age=86400")
self.end_headers()
self.wfile.write(data)
else:
self.send_response(404)
self.end_headers()
elif path.startswith("/"):
filepath = os.path.join(WORKSPACE, path.lstrip("/"))
if os.path.isfile(filepath) and filepath.startswith(WORKSPACE):
with open(filepath, "rb") as f:
data = f.read()
if filepath.endswith(".html"):
# BSN-CSS in Artikel injizieren (vor )
html_str = data.decode("utf-8", errors="replace")
css_block = f""
if "" in html_str:
html_str = html_str.replace("", f"{css_block}\n", 1)
elif "\n{css_block}{html_str}"
data = html_str.encode("utf-8")
self.send_response(200)
self.send_header("Content-Length", str(len(data)))
if filepath.endswith(".html"):
self.send_header("Content-Type", "text/html; charset=utf-8")
elif filepath.endswith(".css"):
self.send_header("Content-Type", "text/css")
elif filepath.endswith(".js"):
self.send_header("Content-Type", "application/javascript")
else:
self.send_header("Content-Type", "application/octet-stream")
self.end_headers()
self.wfile.write(data)
else:
self.send_response(404)
self.end_headers()
self.wfile.write(b"Not found")
else:
self.send_response(404)
self.end_headers()
def do_POST(self):
try:
self._do_POST_impl()
except Exception as e:
print(f"[POST ERROR] {e}", flush=True)
try:
self._json_reply(500, {"ok": False, "error": str(e)})
except Exception:
pass
def _do_POST_impl(self):
parsed = urllib.parse.urlparse(self.path)
path = parsed.path
if path == "/upload":
if not self._check_token():
return
self._handle_upload()
elif path == "/assign-image":
if not self._check_token():
return
self._handle_assign_image()
elif path == "/nanobanana/generate":
if not self._check_token():
return
self._handle_image_generate()
elif path == "/nanobanana/articles":
if not self._check_token():
return
self._handle_nanobanana_articles()
elif path == "/nanobanana/assign":
if not self._check_token():
return
self._handle_nanobanana_assign()
elif path == "/delete":
if not self._check_auth():
return
self._handle_delete()
elif path == "/publish":
if not self._check_auth():
return
self._handle_publish()
else:
self.send_response(404)
self.end_headers()
# ═══════════════════════════════════════════════════════════════════
# Imagen Handler (Google Imagen — synchron, kein Polling)
# ═══════════════════════════════════════════════════════════════════
def _handle_image_generate(self):
"""Bildgenerierung via Google Imagen (synchron, Base64→JPEG)."""
import base64 as b64
content_type = self.headers.get("Content-Type", "")
if "multipart/form-data" not in content_type:
self._json_reply(400, {"ok": False, "error": "Expected multipart/form-data"})
return
boundary = content_type.split("boundary=")[-1].strip()
if not boundary:
self._json_reply(400, {"ok": False, "error": "Kein Boundary"})
return
length = int(self.headers.get("Content-Length", 0))
raw = self.rfile.read(length)
# Multipart manuell parsen (Python 3.13 hat kein cgi)
parts = raw.split(b"--" + boundary.encode())
prompt = ""
aspect_ratio = "16:9"
resolution = "1K"
image_base64 = None
for part in parts:
if b"Content-Disposition" not in part:
continue
headers_end = part.find(b"\r\n\r\n")
if headers_end == -1:
continue
headers_raw = part[:headers_end].decode("utf-8", errors="ignore")
content = part[headers_end + 4:].rstrip(b"\r\n")
if 'name="prompt"' in headers_raw:
prompt = content.decode("utf-8", errors="ignore")
elif 'name="aspectRatio"' in headers_raw:
aspect_ratio = content.decode("utf-8", errors="ignore")
elif 'name="resolution"' in headers_raw:
resolution = content.decode("utf-8", errors="ignore")
elif 'name="image"' in headers_raw and len(content) > 100:
# Referenzbild als Base64 für Image-to-Image
image_base64 = b64.b64encode(content).decode()
if not prompt:
self._json_reply(400, {"ok": False, "error": "Prompt fehlt"})
return
# Gemini Flash Image aufrufen (synchron!)
result = _gemini_generate_image(prompt, image_base64, aspect_ratio, resolution)
if "error" in result:
self._json_reply(500, {"ok": False, "error": result["error"]})
return
# Base64-Bild (PNG) dekodieren
b64_data = result.get("image_base64", "")
if not b64_data:
self._json_reply(500, {"ok": False, "error": "Leeres Bild von Gemini"})
return
img_bytes = b64.b64decode(b64_data)
# Im media-Ordner speichern
now = datetime.now()
month_dir = os.path.join(MEDIA_DIR, now.strftime("%Y"), now.strftime("%m"))
os.makedirs(month_dir, exist_ok=True)
base_name = f"gemini_{now.strftime('%H%M%S')}"
ext = ".png" # Gemini liefert PNG
filepath = os.path.join(month_dir, base_name + ext)
with open(filepath, "wb") as f:
f.write(img_bytes)
# Optional: PIL-Konvertierung zu JPEG (1920x1080 @ 72 DPI)
try:
from PIL import Image
img = Image.open(filepath)
img = img.convert("RGB")
img = img.resize((1920, 1080), Image.LANCZOS)
jpg_path = os.path.join(month_dir, base_name + ".jpg")
img.save(jpg_path, "JPEG", dpi=(72, 72), quality=92)
filepath = jpg_path
ext = ".jpg"
except Exception:
pass # PNG behalten wenn PIL fehlschlägt
img_url = f"/media/{now.strftime('%Y')}/{now.strftime('%m')}/{base_name}{ext}?key={ACCESS_TOKEN}"
self._json_reply(200, {
"ok": True,
"imageUrl": img_url,
"filepath": filepath,
})
def _handle_nanobanana_articles(self):
"""Liste unveröffentlichter Artikel als JSON."""
articles = _get_unpublished_articles()
self._json_reply(200, {"ok": True, "articles": articles})
def _handle_nanobanana_assign(self):
"""Bild einem Artikel zuordnen."""
content_type = self.headers.get("Content-Type", "")
if "application/json" not in content_type:
self._json_reply(400, {"ok": False, "error": "Expected JSON"})
return
length = int(self.headers.get("Content-Length", 0))
raw = self.rfile.read(length)
try:
data = json.loads(raw)
except json.JSONDecodeError:
self._json_reply(400, {"ok": False, "error": "Invalid JSON"})
return
article = data.get("article", "")
image_url = data.get("imageUrl", "")
if not article or not image_url:
self._json_reply(400, {"ok": False, "error": "article und imageUrl erforderlich"})
return
# Bild herunterladen oder lokal lesen
try:
if image_url.startswith("/media/"):
# Lokales Bild — direkt von Platte lesen
local_path = os.path.join(WORKSPACE, image_url.lstrip("/"))
if not os.path.isfile(local_path):
self._json_reply(500, {"ok": False, "error": f"Lokales Bild nicht gefunden: {local_path}"})
return
with open(local_path, "rb") as lf:
img_data = lf.read()
else:
req = urllib.request.Request(image_url)
req.add_header("User-Agent", "Mozilla/5.0")
with urllib.request.urlopen(req, timeout=30) as resp:
img_data = resp.read()
except Exception as e:
self._json_reply(500, {"ok": False, "error": f"Download fehlgeschlagen: {e}"})
return
# In media/YYYY/MM/ speichern
now = datetime.now()
month_dir = os.path.join(MEDIA_DIR, now.strftime("%Y"), now.strftime("%m"))
os.makedirs(month_dir, exist_ok=True)
ts = now.strftime("%H%M%S")
fname = f"nanobanana_{ts}.jpg"
fpath = os.path.join(month_dir, fname)
with open(fpath, "wb") as f:
f.write(img_data)
# article_images.json updaten
img_map = get_image_map()
if article not in img_map:
img_map[article] = []
img_map[article].append(fpath)
save_image_map(img_map)
self._json_reply(200, {"ok": True, "path": fpath, "filename": fname})
def _build_nanobanana_page(self) -> str:
"""Baut die NanoBanana Editor HTML-Seite."""
articles = _get_unpublished_articles()
article_options = ""
for a in articles[:30]:
article_options += f'\n'
return NANOBANANA_HTML.format(
key=ACCESS_TOKEN,
article_options=article_options,
)
# ═══════════════════════════════════════════════════════════════════
# End NanoBanana Handler
# ═══════════════════════════════════════════════════════════════════
def _handle_upload(self):
"""Handle multipart file upload — now accepts 'article' field for association."""
content_type = self.headers.get("Content-Type", "")
if "multipart/form-data" not in content_type:
self._json_reply(400, {"ok": False, "error": "Expected multipart/form-data"})
return
content_length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(content_length)
boundary = None
for part in content_type.split(";"):
part = part.strip()
if part.startswith("boundary="):
boundary = part[9:].strip('"').strip("'").strip()
break
if not boundary:
self._json_reply(400, {"ok": False, "error": "Kein Boundary in Content-Type"})
return
b_boundary = boundary.encode()
parts = body.split(b"--" + b_boundary)
file_data = None
filename = None
article_file = None
for part in parts:
if b"Content-Disposition" not in part:
continue
header_end = part.find(b"\r\n\r\n")
if header_end == -1:
continue
headers = part[:header_end].decode("utf-8", errors="replace")
if b"filename=" in part:
fn_match = re.search(r'filename="([^"]*)"', headers)
if fn_match:
filename = os.path.basename(fn_match.group(1))
file_data = part[header_end + 4:]
if file_data:
file_data = file_data.rstrip(b"\r\n").rstrip(b"-")
elif 'name="article"' in headers:
article_file = part[header_end + 4:].decode("utf-8", errors="replace").strip()
if article_file:
article_file = article_file.rstrip("\r\n").rstrip("-")
if not file_data or not filename:
self._json_reply(400, {"ok": False, "error": "Keine Datei gefunden"})
return
ext = os.path.splitext(filename)[1].lower()
if ext not in (".jpg", ".jpeg", ".png", ".webp", ".gif", ".svg"):
self._json_reply(400, {"ok": False, "error": f"Nicht erlaubtes Format: {ext}. Erlaubt: jpg, png, webp, gif, svg"})
return
now = datetime.now()
month_dir = os.path.join(MEDIA_DIR, now.strftime("%Y"), now.strftime("%m"))
os.makedirs(month_dir, exist_ok=True)
base, ext = os.path.splitext(filename)
safe_base = re.sub(r'[^a-zA-Z0-9_-]', '_', base)
unique_name = f"{safe_base}_{now.strftime('%H%M%S')}{ext}"
filepath = os.path.join(month_dir, unique_name)
with open(filepath, "wb") as f:
f.write(file_data)
# ── Article-Image Association ──────────────────────────────────────
association_msg = ""
if article_file:
img_map = get_image_map()
if article_file not in img_map:
img_map[article_file] = []
img_map[article_file].append(filepath)
save_image_map(img_map)
association_msg = f" → zugeordnet zu: {article_file}"
rel_path = os.path.relpath(filepath, WORKSPACE)
self._json_reply(200, {
"ok": True,
"filename": unique_name,
"path": filepath,
"url": f"MEDIA:{filepath}",
"html_src": f"/media/{now.strftime('%Y')}/{now.strftime('%m')}/{unique_name}",
"article": article_file,
})
def _handle_assign_image(self):
"""Associate an existing image file with an article."""
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length)
try:
data = json.loads(body)
image_path = data.get("image", "")
article_file = data.get("article", "")
except json.JSONDecodeError:
self._json_reply(400, {"ok": False, "error": "Invalid JSON"})
return
if not image_path or not os.path.exists(image_path):
self._json_reply(400, {"ok": False, "error": f"Bild nicht gefunden: {image_path}"})
return
if not article_file:
self._json_reply(400, {"ok": False, "error": "Kein Artikel angegeben"})
return
img_map = get_image_map()
if article_file not in img_map:
img_map[article_file] = []
if image_path not in img_map[article_file]:
img_map[article_file].append(image_path)
save_image_map(img_map)
self._json_reply(200, {"ok": True, "image": image_path, "article": article_file})
def _handle_delete(self):
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length)
try:
data = json.loads(body)
files = data.get("files", [])
except json.JSONDecodeError:
self._json_reply(400, {"ok": False, "error": "Invalid JSON"})
return
deleted = 0
errors = []
for fname in files:
fname = os.path.basename(fname)
if not fname.endswith(".html"):
errors.append(f"{fname}: not an HTML file")
continue
filepath = os.path.join(WORKSPACE, fname)
if not os.path.exists(filepath):
errors.append(f"{fname}: not found")
continue
try:
os.remove(filepath)
deleted += 1
except OSError as e:
errors.append(f"{fname}: {e}")
self._json_reply(200, {"ok": True, "deleted": deleted, "errors": errors})
def _handle_publish(self):
"""Publish selected articles via Joomla API with VG-Wort pixel injection
and automatic cleanup of associated images.
SAFETY GATE (Daniel 18.06.): Ohne 'live': true im Request-Payload
gehen ALLE Artikel auf Korrektur (catid=61, state=0) — niemals live.
"""
if not JOOMLA_TOKEN:
self._json_reply(400, {
"ok": False,
"error": "Kein Joomla API-Token konfiguriert. "
"Bitte JOOMLA_API_TOKEN Umgebungsvariable setzen "
"oder /tmp/alicia_skills/KEYS.txt anlegen."
})
return
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length)
try:
data = json.loads(body)
files = data.get("files", [])
allow_live = data.get("live", False) # SAFETY GATE: Nur wenn explizit true
rubric = data.get("rubric", 0) # 0 = nicht gesetzt
except json.JSONDecodeError:
self._json_reply(400, {"ok": False, "error": "Invalid JSON"})
return
published = 0
published_ids = []
images_deleted_total = 0
errors = []
img_map = get_image_map()
# Safety log
mode = "🔴 LIVE (state=1)" if allow_live else "🟡 KORREKTUR (state=0, Redaktion-only)"
print(f"[PUBLISH] {len(files)} Artikel, Modus: {mode}", flush=True)
# Try importing requests
try:
import requests
except ImportError:
self._json_reply(500, {
"ok": False,
"error": "Python 'requests' Modul nicht installiert. "
"Bitte ausführen: /home/hermes/venv/bin/pip install requests"
})
return
for fname in files:
fname = os.path.basename(fname)
if not fname.endswith(".html"):
errors.append(f"{fname}: not an HTML file")
continue
filepath = os.path.join(WORKSPACE, fname)
if not os.path.exists(filepath):
errors.append(f"{fname}: not found")
continue
try:
with open(filepath, "r", encoding="utf-8") as f:
html_content = f.read()
# ── Extract article metadata ───────────────────────────────
title_match = re.search(r'([^<]+)', html_content)
if not title_match:
errors.append(f"{fname}: kein gefunden")
continue
title = title_match.group(1)
# ── Readiness-Warnung (soft, blockiert nicht) ───────────
readiness_warnings = []
meta_desc_match = re.search(
r' or )
body_match = re.search(r']*>(.*?)', html_content, re.DOTALL)
if not body_match:
# Fallback: use everything between and