280 lines
9.6 KiB
Python
280 lines
9.6 KiB
Python
"""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.
|
|
- Embed actual image bytes as base64 data URIs for offline build use.
|
|
- Write it atomically to disk (for scp by build machines).
|
|
- Check whether the file exists and return its metadata.
|
|
"""
|
|
|
|
import base64
|
|
import json
|
|
import logging
|
|
import re
|
|
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, StorageBlob
|
|
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 _detect_mime_from_extension(path: str) -> str:
|
|
"""Guess MIME type from file extension."""
|
|
ext = Path(path).suffix.lower()
|
|
return {
|
|
".svg": "image/svg+xml",
|
|
".png": "image/png",
|
|
".jpg": "image/jpeg",
|
|
".jpeg": "image/jpeg",
|
|
".webp": "image/webp",
|
|
".gif": "image/gif",
|
|
".ico": "image/x-icon",
|
|
".avif": "image/avif",
|
|
}.get(ext, "application/octet-stream")
|
|
|
|
|
|
async def fetch_image_bytes(url: str) -> tuple[bytes, str] | None:
|
|
"""Fetch image bytes from any source the station config uses.
|
|
|
|
Resolves three URL patterns:
|
|
- /api/storage/{blob_id} → read from StorageBlob in DB
|
|
- Relative paths (e.g. svg/logo.svg) → read from STATIC_FILES_DIR on disk
|
|
- Absolute HTTP(S) URLs → fetch via httpx
|
|
|
|
Returns (raw_bytes, mime_type) or None on failure.
|
|
"""
|
|
if not url:
|
|
return None
|
|
|
|
# Database blob
|
|
blob_match = re.match(r"^/api/storage/([a-f0-9]+)$", url)
|
|
if blob_match:
|
|
blob_id = blob_match.group(1)
|
|
async with async_session() as blob_session:
|
|
result = await blob_session.execute(
|
|
select(StorageBlob).where(StorageBlob.blob_id == blob_id)
|
|
)
|
|
blob = result.scalar_one_or_none()
|
|
if blob:
|
|
return (blob.data, blob.content_type)
|
|
return None
|
|
|
|
# Static file on disk or internal upstream HTTP
|
|
if not url.startswith(("http://", "https://")):
|
|
# Try local filesystem first
|
|
candidates: list[Path] = [Path(settings.STATIC_FILES_DIR)]
|
|
# Dev fallback: if the configured path (production nginx root) doesn't
|
|
# contain the file, try common Angular dist/public locations.
|
|
if not (Path(settings.STATIC_FILES_DIR) / url.lstrip("/")).is_file():
|
|
candidates.extend([
|
|
Path(__file__).resolve().parents[2] / "dist" / "radio-station" / "browser",
|
|
Path(__file__).resolve().parents[2] / "public",
|
|
])
|
|
|
|
for static_dir in candidates:
|
|
if not static_dir.is_dir():
|
|
continue
|
|
file_path = (static_dir / url.lstrip("/")).resolve()
|
|
# Safety: must be inside static_dir
|
|
try:
|
|
file_path.relative_to(static_dir.resolve())
|
|
except ValueError:
|
|
continue
|
|
if file_path.is_file():
|
|
data = file_path.read_bytes()
|
|
mime = _detect_mime_from_extension(url)
|
|
return (data, mime)
|
|
|
|
# Not on local disk — try internal upstream (Docker/K8s) or public URL
|
|
upstream_base = settings.STATIC_UPSTREAM_URL or settings.WEBSITE_URL
|
|
fetch_url = f"{upstream_base.rstrip('/')}/{url.lstrip('/')}"
|
|
|
|
try:
|
|
import httpx
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.get(fetch_url, follow_redirects=True, timeout=10)
|
|
if resp.status_code == 200:
|
|
content_type = resp.headers.get("content-type", "application/octet-stream")
|
|
return (resp.content, content_type)
|
|
except Exception as e:
|
|
logger.warning("Failed to fetch image %s from %s: %s", url, fetch_url, e)
|
|
return None
|
|
|
|
# Remote HTTP fetch
|
|
try:
|
|
import httpx
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.get(url, follow_redirects=True, timeout=10)
|
|
if resp.status_code == 200:
|
|
content_type = resp.headers.get("content-type", "application/octet-stream")
|
|
return (resp.content, content_type)
|
|
except Exception as e:
|
|
logger.warning("Failed to fetch remote image %s: %s", url, e)
|
|
return None
|
|
|
|
|
|
def build_theme_dict(config: StationConfig) -> Dict[str, Any]:
|
|
"""Build the theme.json dict from a StationConfig ORM object.
|
|
|
|
Returns the theme structure with URL-based images. For embedded
|
|
(base64) images, use build_theme_with_embedded_images() instead.
|
|
"""
|
|
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,
|
|
},
|
|
"streams": {
|
|
"url": config.stream_url or None,
|
|
"metadata_url": config.stream_metadata_url or None,
|
|
},
|
|
"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 build_theme_with_embedded_images(
|
|
config: StationConfig,
|
|
) -> Dict[str, Any]:
|
|
"""Build theme dict with images embedded as base64 data URIs.
|
|
|
|
Fetches actual image bytes from static files, database blobs, or
|
|
remote URLs. Failed fetches are logged and omitted (graceful).
|
|
The original URL-based `images` section is kept for reference.
|
|
"""
|
|
theme = build_theme_dict(config)
|
|
|
|
images_embedded: Dict[str, Dict[str, str]] = {}
|
|
image_fields = [
|
|
("logo", config.logo_url),
|
|
("hero_background", config.hero_background_url),
|
|
("hero_icon", config.hero_icon_url),
|
|
("hero_divider", config.hero_divider_url),
|
|
]
|
|
|
|
for key, url in image_fields:
|
|
result = await fetch_image_bytes(url)
|
|
if result is None:
|
|
logger.warning("Could not fetch image for %s (%s) — skipping embed", key, url)
|
|
continue
|
|
data, mime_type = result
|
|
b64 = base64.b64encode(data).decode("ascii")
|
|
images_embedded[key] = {
|
|
"content_type": mime_type,
|
|
"size": len(data),
|
|
"data": b64,
|
|
}
|
|
|
|
theme["images_embedded"] = images_embedded
|
|
return theme
|
|
|
|
|
|
async def write_theme_json() -> Dict[str, Any]:
|
|
"""Read the current station config and write theme.json to disk.
|
|
|
|
Fetches actual image bytes and embeds them as base64 data URIs
|
|
in an `images_embedded` section for offline build use.
|
|
|
|
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 = await build_theme_with_embedded_images(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,
|
|
}
|