"""
SMA Visual Engine — Playwright Render Pipeline
Erzeugt PNG-Grafiken aus Signal-Daten via HTML-Template + Playwright.
Aufruf:
python3 visual_render.py <signal_id> [1x1|4x5]
python3 visual_render.py --test (5 Testsignale)
"""
import base64
import fcntl
import hashlib
import http.client
import io
import json
import re
import socket
import ssl
import stat
import tempfile
import urllib.error
import urllib.parse
import urllib.request
import sys, os, sqlite3, math, textwrap
from pathlib import Path
from PIL import Image, UnidentifiedImageError
from radar_database import RADAR_DB_PATH
from datetime import datetime, timezone
from jinja2 import Environment
# ── Pfade ─────────────────────────────────────────────────────────────────
BASE_DIR = Path(os.environ.get("SMA_AGENT_DIR", "/opt/struktur/social-media-agent")).resolve()
RADAR_DIR = Path(os.environ.get("SMA_RADAR_DIR", "/opt/struktur/social-media-radar")).resolve()
DB_RADAR = RADAR_DB_PATH
TEMPLATE_DIR = BASE_DIR / "templates"
OUTPUT_DIR = Path(os.environ.get("SMA_VISUALS_DIR", str(BASE_DIR / "generated_visuals"))).resolve()
PHOTO_DIR = Path(os.environ.get("SMA_PHOTO_DIR", str(BASE_DIR / "photo_assets"))).resolve()
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
# visual_gate liegt im Radar-Verzeichnis
sys.path.insert(0, str(RADAR_DIR))
from visual_gate import can_generate_visual
from topic_normalize import normalize_topic
# ── Kategorie-Konfig ──────────────────────────────────────────────────────
CAT_CFG = {
"feuchte": {
"icon": "💧", "label": "Feuchtigkeit im Leerstand",
"bg": "#1c1208", "accent": "#c8402a", "light": "#edc878",
"text": "#ffffff", "cta": "Feuchte-Check anfragen",
},
"wetter_klima": {
"icon": "⛈️", "label": "Wetter-Alarm",
"bg": "#1a1a2e", "accent": "#E63946", "light": "#FFBE0B",
"text": "#ffffff", "cta": "Wetterlage beachten",
},
"schimmel_risiko": {
"icon": "⚠️", "label": "Schimmel-Risiko",
"bg": "#2d0a0a", "accent": "#E63946", "light": "#ff8c94",
"text": "#ffffff", "cta": "Experten kontaktieren",
},
"ferienimmobilie": {
"icon": "🏡", "label": "Ferienimmobilien",
"bg": "#0d3b35", "accent": "#2A9D8F", "light": "#E9C46A",
"text": "#ffffff", "cta": "Immobilie schützen",
},
"mallorca_news": {
"icon": "📰", "label": "Mallorca Aktuell",
"bg": "#1b3344", "accent": "#457B9D", "light": "#a8c5da",
"text": "#ffffff", "cta": "Mehr erfahren",
},
}
DEFAULT_CAT = {
"icon": "📡", "label": "Mallorca Signal",
"bg": "#1b2a3b", "accent": "#4a90d9", "light": "#b8d4f0",
"text": "#ffffff", "cta": "Mehr erfahren",
}
FORMATS = {
"1x1": {"w": 1080, "h": 1080, "template": "visual_square.html"},
"4x5": {"w": 1080, "h": 1350, "template": "visual_45.html"},
}
MAX_PHOTO_BYTES = 25 * 1024 * 1024
MAX_IMAGE_EDGE = 12_000
MAX_IMAGE_PIXELS = 40_000_000
IMAGE_GENERATOR_HOST = "image.pollinations.ai"
IMAGE_GENERATOR_VERSION = 2
DEFAULT_IMAGE_HOSTS = {"estaticos-cdn.prensaiberica.es", IMAGE_GENERATOR_HOST}
IMAGE_HOST_ALLOWLIST = DEFAULT_IMAGE_HOSTS | {
host.strip().lower()
for host in os.environ.get("SMA_IMAGE_HOST_ALLOWLIST", "").split(",")
if host.strip()
}
SIGNAL_ID_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]{0,127}\Z")
def _asset_key(signal_id: object) -> str:
"""Use the full validated ID so unrelated signals never share image assets."""
if not isinstance(signal_id, str) or not SIGNAL_ID_PATTERN.fullmatch(signal_id):
raise ValueError("Ungültige Signal-ID für Bild")
return signal_id
def _validate_public_https_url(url: str) -> None:
"""Reject non-HTTPS and local/private destinations before fetching an image."""
import ipaddress
parsed = urllib.parse.urlparse(url)
if (
parsed.scheme.lower() != "https"
or not parsed.hostname
or parsed.username
or parsed.password
or parsed.port not in (None, 443)
):
raise ValueError("Bildquelle muss eine öffentliche HTTPS-URL auf Port 443 sein")
hostname = parsed.hostname.lower()
if hostname not in IMAGE_HOST_ALLOWLIST:
raise ValueError(f"Bildquellen-Host ist nicht freigegeben: {hostname}")
try:
addresses = {
item[4][0]
for item in socket.getaddrinfo(hostname, 443, type=socket.SOCK_STREAM)
}
except socket.gaierror as exc:
raise ValueError("Bildquelle konnte nicht aufgelöst werden") from exc
if not addresses or any(not ipaddress.ip_address(address).is_global for address in addresses):
raise ValueError("Bildquelle darf nicht auf lokale oder private Netze zeigen")
def _validated_image(data: bytes) -> tuple[str, str]:
"""Validate image bytes and return canonical extension and MIME type."""
if not 0 < len(data) <= MAX_PHOTO_BYTES:
raise ValueError("Bild muss zwischen 1 Byte und 25 MB groß sein")
try:
with Image.open(io.BytesIO(data)) as image:
width, height = image.size
image_format = image.format
if (
width <= 0
or height <= 0
or width > MAX_IMAGE_EDGE
or height > MAX_IMAGE_EDGE
or width * height > MAX_IMAGE_PIXELS
):
raise ValueError("Bildabmessungen überschreiten das sichere Limit")
image.verify()
except (Image.DecompressionBombError, UnidentifiedImageError, SyntaxError, OSError) as exc:
raise ValueError("Bildquelle enthält kein gültiges Bild") from exc
formats = {
"JPEG": (".jpg", "image/jpeg"),
"PNG": (".png", "image/png"),
"WEBP": (".webp", "image/webp"),
}
if width <= 0 or height <= 0 or image_format not in formats:
raise ValueError("Bildquelle muss JPEG, PNG oder WebP sein")
return formats[image_format]
def _install_photo_bytes(signal: dict, data: bytes, photo_dir: Path) -> Path:
"""Validate and atomically cache one canonical topic photo per full signal ID."""
key = _asset_key(signal.get("signal_id"))
extension, _mime = _validated_image(data)
root = Path(photo_dir).resolve()
root.mkdir(parents=True, exist_ok=True)
target = root / f"{key}{extension}"
descriptor, temp_name = tempfile.mkstemp(prefix=f".{key}-", suffix=".tmp", dir=root)
temp_path = Path(temp_name)
try:
with os.fdopen(descriptor, "wb") as output:
output.write(data)
output.flush()
os.fsync(output.fileno())
temp_path.replace(target)
for obsolete in root.glob(f"{key}.*"):
if obsolete != target and obsolete.suffix.lower() in {".jpg", ".jpeg", ".png", ".webp"}:
obsolete.unlink(missing_ok=True)
finally:
temp_path.unlink(missing_ok=True)
return target
def _photo_metadata_path(signal: dict, photo_dir: Path) -> Path:
return Path(photo_dir).resolve() / f".{_asset_key(signal.get('signal_id'))}.photo.json"
def _signal_photo_fingerprint(signal: dict) -> str:
values = {
"topic": signal.get("topic") or "",
"topic_de": signal.get("topic_de") or "",
"category": signal.get("signal_category") or "",
"image_url": signal.get("image_url") or "",
"generator_version": IMAGE_GENERATOR_VERSION,
}
return hashlib.sha256(
json.dumps(values, ensure_ascii=False, sort_keys=True).encode("utf-8")
).hexdigest()
def _read_photo_metadata(signal: dict, photo_dir: Path) -> dict:
try:
path = _photo_metadata_path(signal, photo_dir)
if path.is_symlink() or path.resolve(strict=False).parent != Path(photo_dir).resolve():
return {}
value = json.loads(path.read_text(encoding="utf-8"))
return value if isinstance(value, dict) else {}
except (OSError, ValueError, TypeError):
return {}
def _write_photo_metadata(signal: dict, photo_dir: Path, provenance: str) -> None:
root = Path(photo_dir).resolve()
path = _photo_metadata_path(signal, root)
descriptor, temp_name = tempfile.mkstemp(prefix=path.name, suffix=".tmp", dir=root)
temp_path = Path(temp_name)
value = {"provenance": provenance, "fingerprint": _signal_photo_fingerprint(signal)}
try:
with os.fdopen(descriptor, "w", encoding="utf-8") as output:
json.dump(value, output, ensure_ascii=False, sort_keys=True)
output.flush()
os.fsync(output.fileno())
temp_path.replace(path)
finally:
temp_path.unlink(missing_ok=True)
def select_photo_asset(signal: dict, selected_path: Path, photo_dir: Path = PHOTO_DIR) -> Path:
"""Validate and atomically install an explicitly supplied fallback photo."""
_asset_key(signal.get("signal_id"))
selected = Path(selected_path)
if selected.is_symlink():
raise ValueError("Fotoauswahl darf kein Symlink sein")
try:
descriptor = os.open(selected, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
try:
info = os.fstat(descriptor)
if not stat.S_ISREG(info.st_mode) or not 0 < info.st_size <= MAX_PHOTO_BYTES:
raise ValueError("Fotoauswahl muss eine reguläre Bilddatei bis 25 MB sein")
data = os.read(descriptor, info.st_size + 1)
finally:
os.close(descriptor)
except OSError as exc:
raise ValueError("Fotoauswahl ist keine lesbare reguläre Datei") from exc
installed = _install_photo_bytes(signal, data, photo_dir)
_write_photo_metadata(signal, photo_dir, "manual")
return installed
def load_photo_data_uri(signal: dict, photo_dir: Path = PHOTO_DIR) -> str | None:
"""Load the validated local topic image for one exact signal ID."""
try:
key = _asset_key(signal.get("signal_id"))
except ValueError:
return None
root = Path(photo_dir).resolve()
for extension in (".jpg", ".jpeg", ".png", ".webp"):
asset = root / f"{key}{extension}"
if asset.is_symlink() or asset.resolve(strict=False).parent != root:
continue
try:
descriptor = os.open(asset, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
try:
info = os.fstat(descriptor)
if not stat.S_ISREG(info.st_mode) or not 0 < info.st_size <= MAX_PHOTO_BYTES:
continue
data = os.read(descriptor, info.st_size + 1)
finally:
os.close(descriptor)
_extension, mime = _validated_image(data)
except (OSError, ValueError):
continue
encoded = base64.b64encode(data).decode("ascii")
return f"data:{mime};base64,{encoded}"
return None
class _RejectRedirects(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
raise ValueError("Weiterleitungen für Bildquellen sind nicht erlaubt")
class _PinnedHTTPSConnection(http.client.HTTPSConnection):
"""HTTPS connection pinned to one already validated public address."""
def __init__(self, hostname: str, address: str, *, timeout: float):
super().__init__(hostname, port=443, timeout=timeout, context=ssl.create_default_context())
self._pinned_address = address
def connect(self) -> None:
raw_socket = socket.create_connection(
(self._pinned_address, 443), self.timeout, self.source_address
)
self.sock = self._context.wrap_socket(raw_socket, server_hostname=self.host)
class _PinnedImageResponse:
def __init__(self, connection: _PinnedHTTPSConnection, response, url: str):
self._connection = connection
self._response = response
self._url = url
self.status = response.status
self.headers = response.headers
def read(self, amount: int = -1):
return self._response.read(amount)
def geturl(self) -> str:
return self._url
def __enter__(self):
return self
def __exit__(self, *_args):
self._response.close()
self._connection.close()
def _open_image_request(request: urllib.request.Request, timeout: float):
"""Open an allowlisted URL using a validated, pinned public IP and hostname TLS."""
import ipaddress
url = request.full_url
_validate_public_https_url(url)
parsed = urllib.parse.urlparse(url)
hostname = parsed.hostname or ""
addresses = []
for item in socket.getaddrinfo(hostname, 443, type=socket.SOCK_STREAM):
address = item[4][0]
if not ipaddress.ip_address(address).is_global:
raise ValueError("Bildquelle darf nicht auf lokale oder private Netze zeigen")
if address not in addresses:
addresses.append(address)
if not addresses:
raise ValueError("Bildquelle konnte nicht aufgelöst werden")
path = urllib.parse.urlunparse(("", "", parsed.path or "/", parsed.params, parsed.query, ""))
last_error = None
for address in addresses:
connection = _PinnedHTTPSConnection(hostname, address, timeout=timeout)
try:
connection.request(request.get_method(), path, headers=dict(request.header_items()))
return _PinnedImageResponse(connection, connection.getresponse(), url)
except (OSError, ssl.SSLError, http.client.HTTPException) as exc:
last_error = exc
connection.close()
raise urllib.error.URLError(last_error or "Bildquelle nicht erreichbar")
def _generated_photo_url(signal: dict) -> str:
"""Build a deterministic photorealistic image request from public signal context."""
topic = str(signal.get("image_prompt") or signal.get("topic_de") or signal.get("topic") or "Mallorca").strip()
subject = topic[:600]
scene = "professional indoor building-technology scene"
prompt = (
f"Photorealistic documentary {scene}. Main subject: {subject}. "
"The technical measuring device, ventilation component, sensor, or moisture detail must be the dominant foreground subject. "
"Realistic interior, credible materials, no readable invented values, no people identity, no text, no logos, no landscape, no beach, no coastline, no sea view, no tourist photography."
)
key = _asset_key(signal.get("signal_id"))
seed = int(hashlib.sha256(f"{key}:{signal.get('image_generation_nonce', '')}".encode("utf-8")).hexdigest()[:8], 16) % 2_147_483_647
query = urllib.parse.urlencode({
"width": 1200,
"height": 1200,
"seed": seed,
"nologo": "true",
})
return f"https://{IMAGE_GENERATOR_HOST}/prompt/{urllib.parse.quote(prompt, safe='')}?{query}"
def _download_photo(signal: dict, url: str, root: Path, provenance: str = "source") -> str | None:
"""Download, validate, atomically install, and reload one candidate photo."""
_validate_public_https_url(url)
request = urllib.request.Request(
url,
headers={
"User-Agent": "SMA-Staging-Visual/1.0",
"Accept": "image/jpeg,image/png,image/webp",
},
)
with _open_image_request(request, timeout=45) as response:
if getattr(response, "status", None) != 200:
raise ValueError("Bildquelle muss HTTP 200 liefern")
content_type = response.headers.get("Content-Type", "").split(";", 1)[0].lower()
if content_type not in {"image/jpeg", "image/png", "image/webp"}:
raise ValueError("Bildquelle liefert keinen erlaubten Bildtyp")
length = response.headers.get("Content-Length")
if length and int(length) > MAX_PHOTO_BYTES:
raise ValueError("Bildquelle ist größer als 25 MB")
_validate_public_https_url(response.geturl())
data = response.read(MAX_PHOTO_BYTES + 1)
if len(data) > MAX_PHOTO_BYTES:
raise ValueError("Bildquelle ist größer als 25 MB")
_install_photo_bytes(signal, data, root)
_write_photo_metadata(signal, root, provenance)
return load_photo_data_uri(signal, root)
def ensure_topic_photo(signal: dict, photo_dir: Path = PHOTO_DIR, *, force: bool = False) -> str | None:
"""Automatically acquire or generate and cache a topic-matched photograph."""
existing = load_photo_data_uri(signal, photo_dir)
metadata = _read_photo_metadata(signal, photo_dir)
current_fingerprint = _signal_photo_fingerprint(signal)
image_url = signal.get("image_url")
source_url = image_url.strip() if isinstance(image_url, str) else ""
source_available = bool(source_url)
cache_is_stale = metadata.get("provenance") != "manual" and (
not metadata or metadata.get("fingerprint") != current_fingerprint
)
generated_must_yield_to_source = metadata.get("provenance") == "generated" and source_available
if existing and not force and not cache_is_stale and not generated_must_yield_to_source:
return existing
try:
key = _asset_key(signal.get("signal_id"))
root = Path(photo_dir).resolve()
root.mkdir(parents=True, exist_ok=True)
lock_path = root / f".{key}.lock"
with lock_path.open("a+b") as lock_file:
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
existing = load_photo_data_uri(signal, root)
metadata = _read_photo_metadata(signal, root)
cache_is_stale = metadata.get("provenance") != "manual" and (
not metadata or metadata.get("fingerprint") != current_fingerprint
)
generated_must_yield_to_source = metadata.get("provenance") == "generated" and source_available
if existing and not force and not cache_is_stale and not generated_must_yield_to_source:
return existing
candidates = []
if source_available:
candidates.append((source_url, "source"))
candidates.append((_generated_photo_url(signal), "generated"))
for candidate, provenance in candidates:
try:
acquired = _download_photo(signal, candidate, root, provenance)
if acquired:
return acquired
except (OSError, ValueError, urllib.error.URLError):
continue
return None
except (OSError, ValueError, urllib.error.URLError):
return None
def _headline_size(text: str, fmt: str) -> str:
n = len(text)
if fmt == "1x1":
if n < 45: return "68px"
if n < 70: return "56px"
if n < 95: return "46px"
return "36px"
else:
if n < 45: return "72px"
if n < 70: return "60px"
if n < 95: return "50px"
return "40px"
def _trunc(text: str, max_chars: int) -> str:
if not text:
return ""
text = text.strip()
if len(text) <= max_chars:
return text
return text[:max_chars - 1].rstrip() + "…"
def _source_short(source_name: str) -> str:
if not source_name:
return "Mallorca Signal"
# "Google News: Feuchte Immobilien Mallorca (DE)" → "Google News"
return source_name.split(":")[0].strip()[:25]
# ── Visual Hook Lines (Feed-Impact) ──────────────────────────────────────────
HOOK_LINES = {
"feuchte": "Kondensation\nfrüh erkennen.",
"schimmel_risiko": "Erster Fleck an der Wand —\nda ist schon mehr.",
"wetter_klima": "Nach dem Sturm:\nWas Ihre Immobilie jetzt zeigt.",
"ferienimmobilie": "3 Monate leer —\nund das Problem beginnt.",
"mallorca_news": "Was Mallorca-Eigentümer\njetzt wissen müssen.",
"konkurrenz": "Der Markt bewegt sich —\nwo stehen Sie?",
}
def _hook_line(category: str, topic: str = "") -> str:
if category == "wetter_klima" and any(
term in topic.casefold() for term in ("nebel", "niebla", "fog")
):
return "Sommernebel in Palma —\nwas jetzt wichtig ist."
return HOOK_LINES.get(category, "Mallorca\naktuell.")
def _hook_size(fmt: str) -> str:
return "72px"
def build_context(signal: dict, fmt: str, gate: dict) -> dict:
category = (signal.get("signal_category") or "mallorca_news").lower()
cfg = CAT_CFG.get(category, DEFAULT_CAT)
headline = normalize_topic(
topic_de=signal.get("topic_de") or "",
topic=signal.get("topic") or "",
max_chars=110,
) or "Mallorca Signal"
subline = _trunc(signal.get("briefing_winkel") or "", 120)
nutzen = _trunc(signal.get("briefing_nutzen") or "", 130)
risiko = (signal.get("briefing_risiko") or "").lower()
risk_flag = bool(risiko) and not risiko.startswith("kein ")
urgency_raw = signal.get("urgency_level") or 3
urgency = max(1, min(5, math.ceil(urgency_raw / 2)))
created = signal.get("created_at") or ""
date = created[:10] if created else datetime.now(timezone.utc).strftime("%Y-%m-%d")
freigabe = signal.get("briefing_freigabe") or ""
return {
"category_class": gate["category_class"],
"bg_color": cfg["bg"],
"text_color": cfg["text"],
"accent_color": cfg["accent"],
"accent_light": cfg["light"],
"cat_icon": cfg["icon"],
"cat_label": cfg["label"],
"headline": headline,
"headline_size": _headline_size(headline, fmt),
"subline": subline,
"nutzen": nutzen,
"risk_flag": risk_flag,
"urgency": urgency,
"date": date,
"source_short": _source_short(signal.get("source_name") or ""),
"cta_text": cfg["cta"],
"freigabe": freigabe,
"hook_line": _hook_line(category, headline),
"hook_size": _hook_size(fmt),
}
def render_template_html(raw_html: str, context: dict) -> str:
"""Render browser HTML with escaping for every upstream-derived value."""
return Environment(autoescape=True).from_string(raw_html).render(**context)
def render_signal(signal: dict, fmt: str = "1x1", *, force_new: bool = False) -> dict:
"""
Erzeugt PNG fuer ein Signal.
Returns: {"success": bool, "path": str|None, "reason": str}
"""
sid = signal.get("signal_id") or "unknown"
try:
asset_key = _asset_key(sid)
except ValueError as exc:
return {"success": False, "path": None, "reason": str(exc)}
# Gate pruefen
gate = can_generate_visual(signal)
if not gate["allowed"]:
return {"success": False, "path": None, "reason": gate["reason"]}
if fmt not in FORMATS:
return {"success": False, "path": None, "reason": f"Unbekanntes Format: {fmt}"}
photo_data_uri = ensure_topic_photo(signal, PHOTO_DIR, force=force_new)
if not photo_data_uri:
return {
"success": False,
"path": None,
"reason": (
f"Automatische thematische Bildbeschaffung fehlgeschlagen: {asset_key}. "
"Das Signal enthält keine erreichbare, gültige Bildquelle."
),
}
fmt_cfg = FORMATS[fmt]
tmpl_path = TEMPLATE_DIR / fmt_cfg["template"]
if not tmpl_path.exists():
return {"success": False, "path": None, "reason": f"Template nicht gefunden: {tmpl_path}"}
# Kontext aufbauen
ctx = build_context(signal, fmt, gate)
ctx["photo_data_uri"] = photo_data_uri
ctx["ai_generated"] = _read_photo_metadata(signal, PHOTO_DIR).get("provenance") == "generated"
# Template rendern
raw_html = tmpl_path.read_text(encoding="utf-8")
html = render_template_html(raw_html, ctx)
# Temp-HTML schreiben
descriptor, tmp_name = tempfile.mkstemp(
prefix=f".{asset_key}_{fmt}_", suffix=".html", dir=OUTPUT_DIR
)
os.close(descriptor)
tmp_html = Path(tmp_name)
tmp_html.write_text(html, encoding="utf-8")
# Output-Pfad
out_path = OUTPUT_DIR / f"{asset_key}_{fmt}.png"
# Playwright rendert pro Aufruf in eine eigene Datei; erst das vollständige PNG
# wird atomar unter dem stabilen Ausgabe-Namen veröffentlicht.
screenshot_fd, screenshot_name = tempfile.mkstemp(
prefix=f".{asset_key}_{fmt}_", suffix=".png", dir=OUTPUT_DIR
)
os.close(screenshot_fd)
screenshot_tmp = Path(screenshot_name)
try:
from playwright.sync_api import sync_playwright
with sync_playwright() as p:
browser = p.chromium.launch(args=["--no-sandbox", "--disable-dev-shm-usage"])
page = browser.new_page(
viewport={"width": fmt_cfg["w"], "height": fmt_cfg["h"]}
)
page.goto(f"file://{tmp_html.resolve()}", wait_until="networkidle")
page.screenshot(path=str(screenshot_tmp), full_page=False)
browser.close()
screenshot_tmp.replace(out_path)
return {"success": True, "path": str(out_path), "reason": gate["reason"]}
except Exception as e:
return {"success": False, "path": None, "reason": f"Render-Fehler: {e}"}
finally:
tmp_html.unlink(missing_ok=True)
screenshot_tmp.unlink(missing_ok=True)
def render_by_id(signal_id: str, fmt: str = "1x1") -> dict:
"""Laedt ein exakt bezeichnetes Signal aus der DB und rendert."""
try:
_asset_key(signal_id)
except ValueError as exc:
return {"success": False, "path": None, "reason": str(exc)}
conn = sqlite3.connect(str(DB_RADAR))
conn.row_factory = sqlite3.Row
row = conn.execute(
"SELECT * FROM signals WHERE signal_id = ?",
(signal_id,)
).fetchone()
conn.close()
if not row:
return {"success": False, "path": None, "reason": f"Signal nicht gefunden: {signal_id}"}
return render_signal(dict(row), fmt)
# ── CLI ───────────────────────────────────────────────────────────────────
if __name__ == "__main__":
if "--test" in sys.argv:
# 5 Testsignale: eine pro Kategorie + 1 blocked
import sqlite3 as _sq
conn = _sq.connect(str(DB_RADAR))
conn.row_factory = _sq.Row
test_cases = [
("wetter_klima", "signal_category='wetter_klima' AND mallorca_relevance>0 AND radar_score>=31 AND (expires_at IS NULL OR expires_at > datetime('now'))", False),
("feuchte", "signal_category='feuchte' AND mallorca_relevance>0 AND radar_score>=55 AND (expires_at IS NULL OR expires_at > datetime('now'))", False),
("mallorca_news", "signal_category='mallorca_news' AND mallorca_relevance>0 AND radar_score>=31 AND (expires_at IS NULL OR expires_at > datetime('now'))", False),
("schimmel erlaubt","signal_category='schimmel_risiko' AND mallorca_relevance>0 AND briefing_freigabe!='verwerfen'", False),
("schimmel blocked","briefing_freigabe='verwerfen' OR mallorca_relevance=0", True),
]
print("=" * 60)
print("SMA VISUAL ENGINE — TEST (5 Signale)")
print("=" * 60)
passed = 0
for label, where, expect_blocked in test_cases:
row = conn.execute(f"SELECT * FROM signals WHERE {where} LIMIT 1").fetchone()
if not row:
print(f"[SKIP] {label}: kein Signal gefunden")
continue
sig = dict(row)
result = render_signal(sig, "1x1")
ok = result["success"] == (not expect_blocked)
status = "PASS" if ok else "FAIL"
passed += ok
sid = sig.get("signal_id", "")[:8]
print(f"[{status}] {label:<22} | {sid}... | {result['reason'][:55]}")
if result["success"]:
# auch 4x5
r2 = render_signal(sig, "4x5")
print(f" 4x5: {'OK' if r2['success'] else 'FAIL'} | {r2.get('path','')}")
conn.close()
print(f"\nErgebnis: {passed}/5 PASS")
print(f"Output: {OUTPUT_DIR}")
elif len(sys.argv) >= 2:
sid = sys.argv[1]
fmt = sys.argv[2] if len(sys.argv) >= 3 else "1x1"
res = render_by_id(sid, fmt)
print(res)
else:
print("Usage: python3 visual_render.py <signal_id> [1x1|4x5]")
print(" python3 visual_render.py --test")