import os import warnings from io import BytesIO from typing import Any, Dict from fastapi import APIRouter, HTTPException, UploadFile, File from fastapi.responses import FileResponse from PIL import Image, ImageDraw, ImageFont router = APIRouter(prefix="/branding", tags=["branding"]) _BRANDING_DIR = os.path.join(os.getcwd(), "data", "branding") _LOGO_PATH = os.path.join(_BRANDING_DIR, "logo.png") _FAVICON_PATH = os.path.join(_BRANDING_DIR, "favicon.ico") _BUNDLED_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "assets", "branding")) _BUNDLED_LOGO_PATH = os.path.join(_BUNDLED_DIR, "logo.png") _BUNDLED_FAVICON_PATH = os.path.join(_BUNDLED_DIR, "favicon.ico") _BRANDING_SOURCE = os.getenv("BRANDING_SOURCE", "bundled").lower() _MAX_UPLOAD_BYTES = 5 * 1024 * 1024 _MAX_IMAGE_PIXELS = 25_000_000 _ALLOWED_IMAGE_TYPES = {"image/png", "image/jpeg", "image/webp"} _ALLOWED_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp"} def _ensure_branding_dir() -> None: os.makedirs(_BRANDING_DIR, exist_ok=True) def _resize_image(image: Image.Image, max_size: int = 300) -> Image.Image: image = image.convert("RGBA") image.thumbnail((max_size, max_size)) return image def _load_font(size: int) -> ImageFont.ImageFont: candidates = [ "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", ] for path in candidates: if os.path.exists(path): try: return ImageFont.truetype(path, size) except OSError: continue return ImageFont.load_default() def _ensure_default_branding() -> None: if os.path.exists(_LOGO_PATH) and os.path.exists(_FAVICON_PATH): return _ensure_branding_dir() if not os.path.exists(_LOGO_PATH) and os.path.exists(_BUNDLED_LOGO_PATH): try: with open(_BUNDLED_LOGO_PATH, "rb") as source, open(_LOGO_PATH, "wb") as target: target.write(source.read()) except OSError: pass if not os.path.exists(_FAVICON_PATH) and os.path.exists(_BUNDLED_FAVICON_PATH): try: with open(_BUNDLED_FAVICON_PATH, "rb") as source, open(_FAVICON_PATH, "wb") as target: target.write(source.read()) except OSError: pass if not os.path.exists(_LOGO_PATH): image = Image.new("RGBA", (300, 300), (12, 18, 28, 255)) draw = ImageDraw.Draw(image) font = _load_font(160) text = "M" box = draw.textbbox((0, 0), text, font=font) text_w = box[2] - box[0] text_h = box[3] - box[1] draw.text( ((300 - text_w) / 2, (300 - text_h) / 2 - 6), text, font=font, fill=(255, 255, 255, 255), ) image.save(_LOGO_PATH, format="PNG") if not os.path.exists(_FAVICON_PATH): favicon = Image.open(_LOGO_PATH).copy() favicon.thumbnail((64, 64)) try: favicon.save(_FAVICON_PATH, format="ICO", sizes=[(32, 32), (64, 64)]) except OSError: favicon.save(_FAVICON_PATH, format="ICO") def _resolve_branding_paths() -> tuple[str, str]: if _BRANDING_SOURCE == "data": _ensure_default_branding() return _LOGO_PATH, _FAVICON_PATH if os.path.exists(_BUNDLED_LOGO_PATH) and os.path.exists(_BUNDLED_FAVICON_PATH): return _BUNDLED_LOGO_PATH, _BUNDLED_FAVICON_PATH _ensure_default_branding() return _LOGO_PATH, _FAVICON_PATH @router.get("/logo.png") async def branding_logo() -> FileResponse: logo_path, _ = _resolve_branding_paths() if not os.path.exists(logo_path): raise HTTPException(status_code=404, detail="Logo not found") headers = {"Cache-Control": "no-store"} return FileResponse(logo_path, media_type="image/png", headers=headers) @router.get("/favicon.ico") async def branding_favicon() -> FileResponse: _, favicon_path = _resolve_branding_paths() if not os.path.exists(favicon_path): raise HTTPException(status_code=404, detail="Favicon not found") headers = {"Cache-Control": "no-store"} return FileResponse(favicon_path, media_type="image/x-icon", headers=headers) async def save_branding_image(file: UploadFile) -> Dict[str, Any]: content_type = str(file.content_type or "").lower() extension = os.path.splitext(str(file.filename or ""))[1].lower() if content_type not in _ALLOWED_IMAGE_TYPES or extension not in _ALLOWED_IMAGE_EXTENSIONS: raise HTTPException(status_code=400, detail="Upload a PNG, JPEG, or WebP image.") content = await file.read(_MAX_UPLOAD_BYTES + 1) if not content: raise HTTPException(status_code=400, detail="Uploaded file is empty.") if len(content) > _MAX_UPLOAD_BYTES: raise HTTPException(status_code=413, detail="Image is too large (maximum 5 MB).") try: with warnings.catch_warnings(): warnings.simplefilter("error", Image.DecompressionBombWarning) candidate = Image.open(BytesIO(content)) if candidate.format not in {"PNG", "JPEG", "WEBP"}: raise ValueError("Unsupported image format") if candidate.width * candidate.height > _MAX_IMAGE_PIXELS: raise Image.DecompressionBombError("Image pixel limit exceeded") candidate.verify() image = Image.open(BytesIO(content)) image.load() except (OSError, ValueError, Image.DecompressionBombError, Image.DecompressionBombWarning) as exc: raise HTTPException(status_code=400, detail="Image file could not be read.") from exc _ensure_branding_dir() image = _resize_image(image, 300) image.save(_LOGO_PATH, format="PNG") favicon = image.copy() favicon.thumbnail((64, 64)) try: favicon.save(_FAVICON_PATH, format="ICO", sizes=[(32, 32), (64, 64)]) except OSError: favicon.save(_FAVICON_PATH, format="ICO") return {"status": "ok", "width": image.width, "height": image.height}