Adds website theme serialization system

backend/app/theme_export.py

@@ -0,0 +1,134 @@
+"""Theme export: read StationConfig from the DB and dump a theme.json file.
+
+This module provides functions to:
+- Build a clean, consumer-friendly JSON structure from StationConfig.
+- Write it atomically to disk (for scp by build machines).
+- Check whether the file exists and return its metadata.
+"""
+
+import json
+import logging
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict
+
+from app.config import settings
+from app.database import async_session
+from app.models import StationConfig
+from sqlalchemy import select
+
+logger = logging.getLogger(__name__)
+
+
+def resolve_image_url(base_url: str, relative_url: str) -> str | None:
+    """Convert a relative image URL to an absolute one.
+
+    If the URL is already absolute (starts with http:// or https://), return as-is.
+    Returns None for empty strings.
+    """
+    if not relative_url:
+        return None
+    if relative_url.startswith(("http://", "https://")):
+        return relative_url
+    path = relative_url.lstrip("/")
+    base = base_url.rstrip("/")
+    return f"{base}/{path}"
+
+
+def build_theme_dict(config: StationConfig) -> Dict[str, Any]:
+    """Build the theme.json dict from a StationConfig ORM object."""
+    base_url = settings.WEBSITE_URL
+
+    images: Dict[str, str] = {}
+    for key, field in [
+        ("logo", config.logo_url),
+        ("hero_background", config.hero_background_url),
+        ("hero_icon", config.hero_icon_url),
+        ("hero_divider", config.hero_divider_url),
+    ]:
+        resolved = resolve_image_url(base_url, field)
+        if resolved:
+            images[key] = resolved
+
+    return {
+        "version": "1.0",
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "station": {
+            "callsign": config.callsign,
+            "name_primary": config.name_primary,
+            "name_secondary": config.name_secondary,
+            "tagline": config.tagline,
+            "frequency": config.frequency,
+        },
+        "colors": {
+            "primary": config.color_primary,
+            "primary_light": config.color_primary_light,
+            "primary_dark": config.color_primary_dark,
+            "primary_muted": config.color_primary_muted,
+            "accent": config.color_accent,
+            "accent_light": config.color_accent_light,
+            "accent_dark": config.color_accent_dark,
+            "accent_muted": config.color_accent_muted,
+            "background": config.color_background,
+            "text": config.color_text,
+            "success": config.color_success,
+            "danger": config.color_danger,
+            "info": config.color_info,
+            "white": config.color_white,
+            "light": config.color_light,
+            "medium": config.color_medium,
+            "black": config.color_black,
+        },
+        "images": images,
+    }
+
+
+async def write_theme_json() -> Dict[str, Any]:
+    """Read the current station config and write theme.json to disk.
+
+    Uses atomic write (write to .tmp, then os.replace) to prevent
+    reading a partially-written file.
+
+    Returns the theme dict that was written.
+    """
+    async with async_session() as session:
+        result = await session.execute(
+            select(StationConfig).where(StationConfig.key == "default")
+        )
+        config = result.scalar_one_or_none()
+        if config is None:
+            raise RuntimeError("Station config not found in database")
+
+        theme = build_theme_dict(config)
+
+    # Atomic write
+    path = Path(settings.THEME_JSON_PATH)
+    tmp_path = path.with_suffix(".tmp")
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+
+    tmp_path.write_text(json.dumps(theme, indent=2) + "\n", encoding="utf-8")
+    tmp_path.replace(path)
+
+    return theme
+
+
+def get_theme_status() -> Dict[str, Any]:
+    """Check if theme.json exists on disk and return its metadata."""
+    path = Path(settings.THEME_JSON_PATH)
+    if not path.exists():
+        return {"exists": False, "path": str(path)}
+
+    stat = path.stat()
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+        generated_at = data.get("generated_at")
+    except Exception:
+        generated_at = None
+
+    return {
+        "exists": True,
+        "path": str(path),
+        "generated_at": generated_at,
+        "size": stat.st_size,
+    }