diff --git a/AGENT.MD b/AGENT.MD index ef3ec40..bee2295 100644 --- a/AGENT.MD +++ b/AGENT.MD @@ -190,7 +190,10 @@ Remote image search: - `GET /api/artwork/providers` exposes available providers to the UI - `GET /api/artwork/search` calls the selected provider - `POST /api/artwork/import` downloads, validates, size-limits, and caches a selected remote image -- The first provider is Wikimedia Commons; add future providers by implementing `ImageSearchProvider` and registering it in `IMAGE_SEARCH_PROVIDERS` +- Current providers are Wikimedia Commons, optional TMDB, and optional Google Custom Search +- TMDB is enabled with `TMDB_BEARER_TOKEN` or `TMDB_API_KEY` +- Google Custom Search is enabled with `GOOGLE_CUSTOM_SEARCH_API_KEY` and `GOOGLE_CUSTOM_SEARCH_ENGINE_ID` +- Add future providers by implementing `ImageSearchProvider` and registering it in `IMAGE_SEARCH_PROVIDERS` ## Rendering Pipeline diff --git a/README.md b/README.md index 23d880b..5cd64fc 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,20 @@ A self-hosted web UI that generates landscape thumbnails from your Emby library 3. Give it a name (e.g. "Thumb Generator") 4. Copy the key +## Optional Artwork Providers + +The artwork editor always supports Emby images and Wikimedia Commons. You can enable additional search providers with environment variables: + +```bash +export TMDB_BEARER_TOKEN=your-tmdb-read-access-token +# or: export TMDB_API_KEY=your-tmdb-v3-api-key + +export GOOGLE_CUSTOM_SEARCH_API_KEY=your-google-api-key +export GOOGLE_CUSTOM_SEARCH_ENGINE_ID=your-google-search-engine-id +``` + +TMDB is the preferred external artwork source for posters and backdrops. Google Custom Search is optional and only uses Google's official Custom Search JSON API. + ## Notes - First generation will be slower as rembg downloads the U2-Net model (~170MB) diff --git a/app.py b/app.py index 20021a9..1980f9d 100644 --- a/app.py +++ b/app.py @@ -1,24 +1,30 @@ import asyncio import io +import logging import os import hashlib import base64 +import random import time -import urllib.parse +from contextlib import asynccontextmanager from datetime import datetime, timedelta, timezone -from dataclasses import dataclass from pathlib import Path +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%H:%M:%S", +) +logger = logging.getLogger("embytoolkit") + import httpx from fastapi import FastAPI, HTTPException, Query, Request from fastapi.responses import HTMLResponse, Response, StreamingResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates -from PIL import Image, ImageDraw, ImageFont, ImageFilter, ImageColor, ImageOps, UnidentifiedImageError +from PIL import Image, ImageChops, ImageDraw, ImageFont, ImageFilter, ImageColor, ImageOps, UnidentifiedImageError -app = FastAPI(title="Emby Thumbnail Generator") -app.mount("/static", StaticFiles(directory="static"), name="static") -templates = Jinja2Templates(directory="templates") +Image.MAX_IMAGE_PIXELS = None # Emby backdrops can exceed PIL's default bomb threshold; source is trusted EMBY_URL = os.environ.get("EMBY_URL", "http://10.0.0.2:8096") EMBY_API_KEY = os.environ.get("EMBY_API_KEY", "b9af54b630f6448289ab96422add567a") @@ -26,7 +32,12 @@ CACHE_DIR = Path("cache") CACHE_DIR.mkdir(exist_ok=True) IMPORT_CACHE_DIR = CACHE_DIR / "imports" IMPORT_CACHE_DIR.mkdir(exist_ok=True) -RENDER_VERSION = "series-banner-v10" +EMBY_IMAGE_CACHE_DIR = CACHE_DIR / "emby_images" +EMBY_IMAGE_CACHE_DIR.mkdir(exist_ok=True) +CLEAN_POSTER_CACHE_DIR = CACHE_DIR / "clean_posters" +CLEAN_POSTER_CACHE_DIR.mkdir(exist_ok=True) +RENDER_VERSION = "series-banner-v24-poster-logo" +COLLECTION_RENDER_VERSION = "collection-cover-v1" THUMB_WIDTH = 800 THUMB_HEIGHT = 450 @@ -34,13 +45,10 @@ PRIMARY_WIDTH = 1000 PRIMARY_HEIGHT = 1500 PRIMARY_MIN_ZOOM = 1.0 PRIMARY_MAX_ZOOM = 2.6 -REMOTE_IMAGE_MAX_BYTES = 18 * 1024 * 1024 -EDITOR_TARGETS = { - "poster": {"width": 1000, "height": 1500, "emby_type": "Primary", "label": "Poster"}, - "thumb": {"width": 1920, "height": 1080, "emby_type": "Thumb", "label": "Thumb"}, - "backdrop": {"width": 1920, "height": 1080, "emby_type": "Backdrop", "label": "Backdrop"}, -} +UPLOAD_MAX_BYTES = 40 * 1024 * 1024 HTTP_TIMEOUT = 30.0 +EMBY_SOURCE_CACHE_TTL = 300 +EMBY_SOURCE_MISS_TTL = 90 http_client: httpx.AsyncClient | None = None AIRING_LOOKUP_CACHE_TTL = 900 NEW_SEASON_MIN_AGE_DAYS = 7 @@ -51,21 +59,56 @@ airing_lookup_lock = asyncio.Lock() # ── Studio logo file map ───────────────────────────────────────────────────── STUDIOS_DIR = Path("static/studios") -# Maps studio key → filename (relative to STUDIOS_DIR) -STUDIO_FILES: dict[str, str] = { - "hulu": "hulu.png", - "hbo": "hbo.png", - "disney": "disney.png", + +# Maps typed search aliases → internal studio key +STUDIO_SEARCH_ALIASES: dict[str, str] = { + "netflix": "netflix", + "apple tv": "appletv", + "apple tv+": "appletv", + "appletv": "appletv", + "appletv+": "appletv", + "apple": "appletv", + "paramount": "paramountplus", + "paramount+": "paramountplus", + "cbs": "paramountplus", + "hbo": "hbo", + "hbo max": "hbo", + "max": "hbo", + "disney": "disney", + "disney+": "disney", + "hulu": "hulu", +} + +# Emby studio display names to search for each key (tried in order) +STUDIO_EMBY_NAMES: dict[str, list[str]] = { + "netflix": ["Netflix"], + "appletv": ["Apple TV+", "Apple Studios", "Apple"], + "paramountplus": ["Paramount+", "CBS Studios", "CBS", "Paramount Network"], + "hbo": ["HBO", "HBO Max", "Max"], + "disney": ["Disney+", "Disney"], + "hulu": ["Hulu"], +} +STUDIO_FILES: dict[str, Path] = { + "hulu": STUDIOS_DIR / "hulu.png", + "hbo": STUDIOS_DIR / "hbo.png", + "disney": STUDIOS_DIR / "disney.png", + "appletv": STUDIOS_DIR / "apple-tv.png", + "paramountplus": STUDIOS_DIR / "paramount-plus.png", + "netflix": STUDIOS_DIR / "netflix.png", } +def round_image_corners(img: Image.Image, radius: int) -> Image.Image: + img = img.convert("RGBA") + mask = Image.new("L", img.size, 0) + ImageDraw.Draw(mask).rounded_rectangle([(0, 0), (img.width - 1, img.height - 1)], radius=radius, fill=255) + img.putalpha(ImageChops.multiply(img.getchannel("A"), mask)) + return img + + def make_studio_logo(studio: str, max_height: int = 52) -> Image.Image | None: - """Load a studio logo from disk and scale it to max_height, preserving aspect ratio.""" - filename = STUDIO_FILES.get(studio) - if not filename: - return None - path = STUDIOS_DIR / filename - if not path.exists(): + path = STUDIO_FILES.get(studio) + if not path or not path.exists(): return None img = Image.open(path).convert("RGBA") if img.height > max_height: @@ -74,6 +117,80 @@ def make_studio_logo(studio: str, max_height: int = 52) -> Image.Image | None: return img +def infer_auto_studio_key(item: dict) -> str | None: + item_type = (item.get("Type") or "").strip().lower() + if item_type not in ("series", "movie"): + return None + image_tags = item.get("ImageTags") or {} + if "Logo" not in image_tags: + return None + + studios = item.get("Studios") or [] + names: list[str] = [] + if isinstance(studios, list): + for studio in studios: + if isinstance(studio, dict): + name = studio.get("Name") or "" + else: + name = str(studio or "") + if name: + names.append(name) + elif isinstance(studios, str) and studios.strip(): + names.append(studios) + + normalized = " ".join(names).lower() + logger.info(" Studio detection : type=%s studios=%s", item_type, names) + + if "apple" in normalized: + return "appletv" + if "netflix" in normalized: + return "netflix" + if "paramount" in normalized: + return "paramountplus" + return None + + +async def get_item_summary(item_id: str) -> dict | None: + data = await emby_get("/Items", { + "Ids": item_id, + "Recursive": "true", + "Fields": "ProductionYear,Studios", + "ImageTypeLimit": "1", + "EnableImageTypes": "Primary,Logo,Backdrop", + }) + items = data.get("Items") or [] + return items[0] if items else None + + +async def apply_auto_studio_defaults(options: dict) -> dict: + effective = dict(options) + requested_studio = (effective.get("studio") or "none").strip().lower() + if requested_studio == "appletv": + effective["studio_position"] = "top-left" + return effective + if requested_studio not in ("auto", ""): + return effective + + item = await get_item_summary(effective["item_id"]) + if not item: + effective["studio"] = "none" + return effective + auto_studio = infer_auto_studio_key(item) + if auto_studio == "appletv": + effective["studio"] = "appletv" + effective["studio_position"] = "top-left" + elif auto_studio == "netflix": + effective["studio"] = "netflix" + effective["studio_position"] = "top-left" + elif auto_studio == "paramountplus": + effective["studio"] = "paramountplus" + effective["studio_position"] = "top-left" + else: + effective["studio"] = "none" + logger.info(" Auto studio : %s → %s", item.get("Name"), effective["studio"]) + return effective + + def load_image_from_bytes(image_bytes: bytes, mode: str = "RGB") -> Image.Image: """Fully decode image bytes into an in-memory PIL image. @@ -106,152 +223,89 @@ def validate_image_bytes(image_bytes: bytes, *, max_pixels: int = 24_000_000) -> return width, height, fmt -def get_editor_target(target_mode: str) -> dict: - target = EDITOR_TARGETS.get(target_mode) - if target is None: - raise HTTPException(status_code=400, detail="Invalid artwork target mode.") - return target - - -def get_import_cache_path(import_id: str) -> Path: - if not import_id or any(ch not in "0123456789abcdef" for ch in import_id.lower()): - raise HTTPException(status_code=400, detail="Invalid imported image id.") - path = IMPORT_CACHE_DIR / f"{import_id}.img" +def get_upload_cache_path(upload_id: str) -> Path: + if not upload_id or any(ch not in "0123456789abcdef" for ch in upload_id.lower()): + raise HTTPException(status_code=400, detail="Invalid upload id.") + path = IMPORT_CACHE_DIR / f"{upload_id}.img" if not path.exists(): - raise HTTPException(status_code=404, detail="Imported image was not found in cache.") + raise HTTPException(status_code=404, detail="Uploaded background not found in cache.") return path -async def resolve_editor_source_bytes(body: dict) -> bytes: - source_kind = body.get("source_kind", "emby") - item_id = body.get("item_id") - if source_kind == "import": - return get_import_cache_path(body.get("import_id", "")).read_bytes() - if source_kind != "emby": - raise HTTPException(status_code=400, detail="Invalid source image kind.") - if not item_id: - raise HTTPException(status_code=400, detail="Missing item id.") - image_type = body.get("source_type", "Primary") - if image_type not in {"Primary", "Thumb", "Backdrop"}: - raise HTTPException(status_code=400, detail="Invalid Emby source image type.") - index_raw = body.get("source_index", None) - index = int(index_raw) if index_raw is not None else None - return await emby_get_image(item_id, image_type, index=index) +def get_emby_image_cache_key(item_id: str, image_type: str, index: int | None = None) -> str: + payload = f"{item_id}:{image_type}:{index if index is not None else 'default'}" + return hashlib.md5(payload.encode("utf-8")).hexdigest() -def image_to_data_url(image_bytes: bytes, media_type: str = "image/jpeg") -> str: - return f"data:{media_type};base64,{base64.b64encode(image_bytes).decode('ascii')}" +def get_emby_image_cache_path(item_id: str, image_type: str, index: int | None = None) -> Path: + return EMBY_IMAGE_CACHE_DIR / f"{get_emby_image_cache_key(item_id, image_type, index)}.img" -def fit_image_positioned( - img: Image.Image, - width: int, - height: int, +def get_emby_image_miss_path(item_id: str, image_type: str, index: int | None = None) -> Path: + return EMBY_IMAGE_CACHE_DIR / f"{get_emby_image_cache_key(item_id, image_type, index)}.missing" + + +def get_fresh_cached_bytes(path: Path, ttl_seconds: int) -> bytes | None: + if not path.exists(): + return None + age = time.time() - path.stat().st_mtime + if age > ttl_seconds: + return None + return path.read_bytes() + + +async def get_cached_emby_source_image( + item_id: str, + image_type: str, + index: int | None = None, *, - fit_mode: str = "cover", - zoom: float = 1.0, - pan_x: float = 0.0, - pan_y: float = 0.0, -) -> tuple[Image.Image, tuple[int, int]]: - fit_mode = fit_mode if fit_mode in {"cover", "contain"} else "cover" - pan_x = max(-1.0, min(1.0, float(pan_x))) - pan_y = max(-1.0, min(1.0, float(pan_y))) - zoom = max(0.25, min(5.0, float(zoom))) - scale_fn = max if fit_mode == "cover" else min - scale = scale_fn(width / max(1, img.width), height / max(1, img.height)) * zoom - fitted_w = max(1, int(round(img.width * scale))) - fitted_h = max(1, int(round(img.height * scale))) - fitted = img.resize((fitted_w, fitted_h), Image.LANCZOS) + optional: bool = False, +) -> bytes | None: + cache_path = get_emby_image_cache_path(item_id, image_type, index) + cached_bytes = get_fresh_cached_bytes(cache_path, EMBY_SOURCE_CACHE_TTL) + if cached_bytes is not None: + return cached_bytes - if fitted_w >= width: - x = -int(round((fitted_w - width) * ((pan_x + 1.0) / 2.0))) - else: - x = int(round((width - fitted_w) * ((pan_x + 1.0) / 2.0))) - if fitted_h >= height: - y = -int(round((fitted_h - height) * ((pan_y + 1.0) / 2.0))) - else: - y = int(round((height - fitted_h) * ((pan_y + 1.0) / 2.0))) - return fitted, (x, y) + miss_path = get_emby_image_miss_path(item_id, image_type, index) + if optional and miss_path.exists() and (time.time() - miss_path.stat().st_mtime) <= EMBY_SOURCE_MISS_TTL: + return None - -def build_editor_fill( - img: Image.Image, - width: int, - height: int, - *, - fill_mode: str, - fill_color: str | None, -) -> Image.Image: - if fill_mode == "solid": - try: - color = ImageColor.getrgb(fill_color or "#111318") - except ValueError: - color = (17, 19, 24) - return Image.new("RGB", (width, height), color) - - base = cover_crop(img, width, height) - if fill_mode == "mirror": - base = ImageOps.mirror(base) - overlay = Image.new("RGBA", (width, height), (0, 0, 0, 54)) - return Image.alpha_composite(base.convert("RGBA"), overlay).convert("RGB") - - base = base.filter(ImageFilter.GaussianBlur(radius=max(18, int(min(width, height) * 0.035)))) - overlay = Image.new("RGBA", (width, height), (0, 0, 0, 70)) - return Image.alpha_composite(base.convert("RGBA"), overlay).convert("RGB") - - -def render_artwork_editor_image( - source_bytes: bytes, - *, - target_mode: str = "poster", - fit_mode: str = "cover", - fill_mode: str = "blur", - fill_color: str | None = "#101318", - zoom: float = 1.0, - pan_x: float = 0.0, - pan_y: float = 0.0, - quality: int = 95, -) -> bytes: - target = get_editor_target(target_mode) - width = target["width"] - height = target["height"] - source = load_image_from_bytes(source_bytes, mode="RGB") - canvas = build_editor_fill(source, width, height, fill_mode=fill_mode, fill_color=fill_color) - fitted, (x, y) = fit_image_positioned( - source, - width, - height, - fit_mode=fit_mode, - zoom=zoom, - pan_x=pan_x, - pan_y=pan_y, + image_bytes = ( + await emby_get_image_optional(item_id, image_type, index=index) + if optional + else await emby_get_image(item_id, image_type, index=index) ) - canvas.paste(fitted, (x, y)) - return encode_jpeg_bytes(canvas, quality=max(80, min(98, int(quality)))) + if image_bytes is None: + if optional: + miss_path.touch() + return None + raise HTTPException(status_code=404, detail=f"Emby image {image_type} was not found.") - -def build_editor_cache_key(body: dict) -> str: - payload = { - "version": "artwork-editor-v1", - "item_id": body.get("item_id"), - "source_kind": body.get("source_kind", "emby"), - "source_type": body.get("source_type", "Primary"), - "source_index": body.get("source_index"), - "import_id": body.get("import_id"), - "target_mode": body.get("target_mode", "poster"), - "fit_mode": body.get("fit_mode", "cover"), - "fill_mode": body.get("fill_mode", "blur"), - "fill_color": body.get("fill_color", "#101318"), - "zoom": round(float(body.get("zoom", 1.0)), 4), - "pan_x": round(float(body.get("pan_x", 0.0)), 4), - "pan_y": round(float(body.get("pan_y", 0.0)), 4), - "quality": int(body.get("quality", 95)), - } - return hashlib.md5(repr(payload).encode("utf-8")).hexdigest() + cache_path.write_bytes(image_bytes) + if miss_path.exists(): + miss_path.unlink(missing_ok=True) + return image_bytes # --- Emby API helpers --- + +@asynccontextmanager +async def lifespan(app: FastAPI): + get_http_client() + try: + yield + finally: + global http_client + if http_client is not None: + await http_client.aclose() + http_client = None + + +app = FastAPI(title="EmbyToolkit", lifespan=lifespan) +app.mount("/static", StaticFiles(directory="static"), name="static") +templates = Jinja2Templates(directory="templates") + def get_http_client() -> httpx.AsyncClient: global http_client if http_client is None: @@ -259,26 +313,51 @@ def get_http_client() -> httpx.AsyncClient: return http_client -@app.on_event("startup") -async def startup_event(): - get_http_client() - - -@app.on_event("shutdown") -async def shutdown_event(): - global http_client - if http_client is not None: - await http_client.aclose() - http_client = None - -async def emby_get(path: str, params: dict = None) -> dict: +async def emby_request(method: str, path: str, *, params: dict | None = None, **kwargs) -> httpx.Response: client = get_http_client() url = f"{EMBY_URL}{path}" - p = params or {} + p = dict(params or {}) p["api_key"] = EMBY_API_KEY - r = await client.get(url, params=p) - r.raise_for_status() - return r.json() + try: + return await client.request(method, url, params=p, **kwargs) + except httpx.ReadTimeout as exc: + raise HTTPException( + status_code=504, + detail=f"Emby timed out while requesting {path}. Try again in a moment.", + ) from exc + except httpx.ConnectTimeout as exc: + raise HTTPException( + status_code=504, + detail=f"Timed out connecting to Emby at {EMBY_URL}.", + ) from exc + except httpx.ConnectError as exc: + raise HTTPException( + status_code=502, + detail=f"Could not connect to Emby at {EMBY_URL}.", + ) from exc + except httpx.RequestError as exc: + raise HTTPException( + status_code=502, + detail=f"Emby request failed while requesting {path}: {exc}", + ) from exc + + +def ensure_emby_success(response: httpx.Response, *, context: str) -> httpx.Response: + try: + response.raise_for_status() + except httpx.HTTPStatusError as exc: + detail = response.text.strip() or str(exc) + raise HTTPException( + status_code=502, + detail=f"{context} failed ({response.status_code}): {detail}", + ) from exc + return response + + +async def emby_get(path: str, params: dict = None) -> dict: + response = await emby_request("GET", path, params=params) + ensure_emby_success(response, context=f"Emby request to {path}") + return response.json() async def emby_get_image( @@ -290,8 +369,6 @@ async def emby_get_image( max_height: int | None = None, quality: int | None = None, ) -> bytes: - client = get_http_client() - url = f"{EMBY_URL}/Items/{item_id}/Images/{image_type}" params: dict[str, str | int] = {"api_key": EMBY_API_KEY} if index is not None: params["Index"] = index @@ -301,9 +378,9 @@ async def emby_get_image( params["maxHeight"] = max_height if quality is not None: params["quality"] = quality - r = await client.get(url, params=params) - r.raise_for_status() - return r.content + response = await emby_request("GET", f"/Items/{item_id}/Images/{image_type}", params=params) + ensure_emby_success(response, context=f"Emby image request for {image_type}") + return response.content async def emby_get_image_with_type( @@ -315,8 +392,6 @@ async def emby_get_image_with_type( max_height: int | None = None, quality: int | None = None, ) -> tuple[bytes, str]: - client = get_http_client() - url = f"{EMBY_URL}/Items/{item_id}/Images/{image_type}" params: dict[str, str | int] = {"api_key": EMBY_API_KEY} if index is not None: params["Index"] = index @@ -326,27 +401,25 @@ async def emby_get_image_with_type( params["maxHeight"] = max_height if quality is not None: params["quality"] = quality - r = await client.get(url, params=params) - r.raise_for_status() - content_type = (r.headers.get("content-type", "image/jpeg") or "image/jpeg").split(";")[0] - return r.content, content_type + response = await emby_request("GET", f"/Items/{item_id}/Images/{image_type}", params=params) + ensure_emby_success(response, context=f"Emby image request for {image_type}") + content_type = (response.headers.get("content-type", "image/jpeg") or "image/jpeg").split(";")[0] + return response.content, content_type async def emby_get_image_optional(item_id: str, image_type: str, index: int | None = None) -> bytes | None: """Returns image bytes or None if the image type doesn't exist for this item.""" - client = get_http_client() - url = f"{EMBY_URL}/Items/{item_id}/Images/{image_type}" params: dict[str, str | int] = {"api_key": EMBY_API_KEY} if index is not None: params["Index"] = index - r = await client.get(url, params=params) - if r.status_code in (404, 400): + response = await emby_request("GET", f"/Items/{item_id}/Images/{image_type}", params=params) + if response.status_code in (404, 400): return None - r.raise_for_status() - content_type = r.headers.get("content-type", "").lower() + ensure_emby_success(response, context=f"Optional Emby image request for {image_type}") + content_type = response.headers.get("content-type", "").lower() if content_type and not content_type.startswith("image/"): return None - return r.content + return response.content async def emby_upload_image(item_id: str, image_bytes: bytes, image_type: str = "Thumb"): @@ -355,10 +428,9 @@ async def emby_upload_image(item_id: str, image_bytes: bytes, image_type: str = upload_image.save(upload_buf, format="JPEG", quality=92, optimize=True) upload_b64 = base64.b64encode(upload_buf.getvalue()) - client = get_http_client() - url = f"{EMBY_URL}/Items/{item_id}/Images/{image_type}" - r = await client.post( - url, + response = await emby_request( + "POST", + f"/Items/{item_id}/Images/{image_type}", params={"api_key": EMBY_API_KEY}, content=upload_b64, headers={ @@ -366,84 +438,20 @@ async def emby_upload_image(item_id: str, image_bytes: bytes, image_type: str = "X-Emby-Token": EMBY_API_KEY, }, ) - try: - r.raise_for_status() - except httpx.HTTPStatusError as exc: - detail = r.text.strip() or str(exc) - raise HTTPException( - status_code=502, - detail=f"Emby upload failed ({r.status_code}): {detail}", - ) from exc - return r.status_code + ensure_emby_success(response, context=f"Emby upload for {image_type}") + return response.status_code -@dataclass -class ArtworkSearchResult: - id: str - title: str - thumbnail_url: str - image_url: str - source_url: str - provider: str - width: int | None = None - height: int | None = None +async def emby_delete_image(item_id: str, image_type: str) -> int: + response = await emby_request( + "DELETE", + f"/Items/{item_id}/Images/{image_type}", + params={"api_key": EMBY_API_KEY}, + headers={"X-Emby-Token": EMBY_API_KEY}, + ) + return response.status_code -class ImageSearchProvider: - key = "base" - label = "Base" - - async def search(self, query: str, limit: int = 12) -> list[ArtworkSearchResult]: - raise NotImplementedError - - -class WikimediaImageSearchProvider(ImageSearchProvider): - key = "wikimedia" - label = "Wikimedia Commons" - - async def search(self, query: str, limit: int = 12) -> list[ArtworkSearchResult]: - client = get_http_client() - params = { - "action": "query", - "generator": "search", - "gsrnamespace": "6", - "gsrsearch": query, - "gsrlimit": str(max(1, min(24, limit))), - "prop": "imageinfo", - "iiprop": "url|size|mime", - "iiurlwidth": "420", - "format": "json", - "origin": "*", - } - r = await client.get("https://commons.wikimedia.org/w/api.php", params=params) - r.raise_for_status() - pages = (r.json().get("query") or {}).get("pages") or {} - results: list[ArtworkSearchResult] = [] - for page in pages.values(): - info = (page.get("imageinfo") or [{}])[0] - mime = info.get("mime", "") - if not mime.startswith("image/"): - continue - image_url = info.get("url") - if not image_url: - continue - title = (page.get("title") or "Artwork").replace("File:", "", 1) - results.append(ArtworkSearchResult( - id=hashlib.md5(image_url.encode("utf-8")).hexdigest(), - title=title, - thumbnail_url=info.get("thumburl") or image_url, - image_url=image_url, - source_url=info.get("descriptionurl") or image_url, - provider=self.key, - width=info.get("width"), - height=info.get("height"), - )) - return results - - -IMAGE_SEARCH_PROVIDERS: dict[str, ImageSearchProvider] = { - WikimediaImageSearchProvider.key: WikimediaImageSearchProvider(), -} def parse_emby_datetime(value: str | None) -> datetime | None: @@ -462,6 +470,28 @@ def to_emby_iso(value: datetime) -> str: return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") +async def emby_refresh_backdrops(item_id: str, *, wait_seconds: float = 6.0) -> int: + """Ask Emby to fetch additional backdrop images for an item, then return the new backdrop count.""" + resp = await emby_request("POST", f"/Items/{item_id}/Refresh", params={ + "MetadataRefreshMode": "None", + "ImageRefreshMode": "FullRefresh", + "ReplaceAllImages": "false", + "ReplaceAllMetadata": "false", + }) + if resp.status_code not in (200, 204): + logger.warning("Emby refresh returned %d for item %s", resp.status_code, item_id) + return 0 + await asyncio.sleep(wait_seconds) + data = await emby_get("/Items", { + "Ids": item_id, + "Fields": "", + "ImageTypeLimit": "1", + "EnableImageTypes": "Backdrop", + }) + backdrop_tags = (data.get("Items") or [{}])[0].get("BackdropImageTags") or [] + return len(backdrop_tags) + + async def emby_get_all(path: str, params: dict | None = None, *, page_size: int = 200) -> list[dict]: items: list[dict] = [] start_index = 0 @@ -503,7 +533,7 @@ async def build_airing_snapshot(week_offset: int = 0) -> dict: "SortBy": "SortName", "SortOrder": "Ascending", "ImageTypeLimit": "1", - "EnableImageTypes": "Primary", + "EnableImageTypes": "Primary,Logo", }) recent_episodes_task = emby_get_all("/Items", { "IncludeItemTypes": "Episode", @@ -684,6 +714,7 @@ async def build_airing_snapshot(week_offset: int = 0) -> dict: "status": status or ("Continuing" if upcoming else "Unknown"), "air_days": air_days, "poster_url": f"/api/poster/{series_id}?w=180&h=270&q=84", + "has_logo": "Logo" in (series.get("ImageTags") or {}), "selected_week_air_at": to_emby_iso(selected_week_episode["premiere_at"]) if selected_week_episode else None, "selected_week_episode_label": ( f"S{selected_week_episode['season_number']}E{selected_week_episode['episode_number']} · {selected_week_episode['episode_name']}".strip(" ·") @@ -739,6 +770,235 @@ async def get_airing_snapshot(force_refresh: bool = False, week_offset: int = 0) return data +def can_apply_new_season_snapshot_item(item: dict) -> bool: + return bool(item.get("eligible_new_season")) or ( + (item.get("status") or "").strip().lower() == "continuing" + and (item.get("selected_week_air_at") or item.get("next_air_at")) + ) + + +async def apply_new_season_artwork_for_snapshot_item( + item: dict, + *, + title: str | None = None, + generate_primary: bool = True, +) -> dict: + item_id = item["id"] + if not can_apply_new_season_snapshot_item(item): + raise HTTPException(status_code=400, detail="Series is not eligible for New Season artwork.") + if not item.get("has_logo"): + raise HTTPException(status_code=400, detail="Series requires an Emby logo before New Season artwork can be applied.") + + title = title or item["name"] + item_summary = await get_item_summary(item_id) + auto_studio = infer_auto_studio_key(item_summary or {}) + effective_studio = auto_studio or "none" + effective_studio_position = "top-left" + + cache_key = build_cache_key( + item_id=item_id, + bg_mode="backdrop", + backdrop_index=0, + text_color="#FFFFFF", + logo_align="bottom-center", + logo_scale=1.3, + darkness=0.0, + studio=effective_studio, + studio_position=effective_studio_position, + new_episodes_tag=True, + season_finale_tag=False, + logo_index=0, + ) + thumb_cache_path = get_thumb_cache_path(cache_key) + primary_cache_path = get_primary_cache_path(cache_key) + + if not thumb_cache_path.exists() or (generate_primary and not primary_cache_path.exists()): + poster_bytes = await emby_get_image(item_id, "Primary") + logo_bytes = await emby_get_image_optional(item_id, "Logo", 0) + if logo_bytes is None: + raise HTTPException(status_code=400, detail="Series requires an Emby logo before New Season artwork can be applied.") + backdrop_bytes = await emby_get_image_optional(item_id, "Backdrop", 0) + thumb_bytes = generate_thumbnail( + poster_bytes, + title, + bg_mode="backdrop", + text_color="#FFFFFF", + logo_align="bottom-center", + logo_scale=1.3, + darkness=0.0, + studio=effective_studio, + studio_position=effective_studio_position, + new_episodes_tag=True, + season_finale_tag=False, + logo_index=0, + logo_bytes=logo_bytes, + backdrop_bytes=backdrop_bytes, + ) + thumb_cache_path.write_bytes(thumb_bytes) + if generate_primary: + primary_bytes = generate_primary_cover( + poster_bytes, + title, + bg_mode="backdrop", + text_color="#FFFFFF", + logo_align="bottom-center", + logo_scale=1.3, + darkness=0.0, + studio=effective_studio, + studio_position=effective_studio_position, + new_episodes_tag=True, + season_finale_tag=False, + logo_index=0, + logo_bytes=logo_bytes, + ) + primary_cache_path.write_bytes(primary_bytes) + + thumb_status = await emby_upload_image(item_id, thumb_cache_path.read_bytes(), "Thumb") + + primary_status = None + primary_error = None + if generate_primary: + try: + primary_status = await emby_upload_image(item_id, primary_cache_path.read_bytes(), "Primary") + except Exception as exc: + primary_error = str(exc) + + return { + "status": "applied", + "thumb_code": thumb_status, + "primary_code": primary_status, + "primary_attempted": generate_primary, + "primary_error": primary_error, + } + + +def can_bulk_assign_series_item(item: dict) -> bool: + return bool(item.get("has_logo")) and bool(item.get("has_backdrop")) + + +def build_bulk_assign_series_options(item_id: str, title: str) -> dict: + return { + "item_id": item_id, + "title": title, + "bg_mode": "backdrop", + "backdrop_index": 0, + "text_color": "#FFFFFF", + "logo_align": "bottom-center", + "logo_scale": 1.3, + "darkness": 0.0, + "studio": "auto", + "studio_position": "top-left", + "new_episodes_tag": False, + "season_finale_tag": False, + "generate_primary": False, + "logo_index": 0, + "primary_zoom": 1.0, + "primary_pan_x": 0.0, + "primary_pan_y": -0.16, + "thumb_zoom": 1.0, + "thumb_pan_x": 0.0, + "thumb_pan_y": 0.0, + "upload_bg_id": None, + } + + +async def stamp_studio_logo_on_poster(item_id: str, studio: str, studio_position: str = "top-left") -> int: + """Fetch the clean original poster, stamp the studio logo, and re-upload.""" + logger.info(" Stamping poster : studio=%s position=%s", studio, studio_position) + clean_path = CLEAN_POSTER_CACHE_DIR / f"{item_id}.img" + if clean_path.exists(): + poster_bytes = clean_path.read_bytes() + logger.info(" Poster source : clean cache hit for %s", item_id) + else: + # Primary may be temporarily absent if Emby is mid-refresh after a reset. + # Retry a few times with a short wait before giving up. + poster_bytes = None + for attempt in range(4): + poster_bytes = await emby_get_image_optional(item_id, "Primary") + if poster_bytes: + break + if attempt < 3: + logger.info(" Poster source : Primary not ready (attempt %d/4), waiting…", attempt + 1) + await asyncio.sleep(5) + if not poster_bytes: + logger.warning(" Poster stamp skipped: no Primary image found in Emby for item %s after retries", item_id) + return 0 + clean_path.write_bytes(poster_bytes) + logger.info(" Poster source : saved clean baseline for %s", item_id) + img = load_image_from_bytes(poster_bytes, mode="RGB") + w, h = img.size + logo_max_h = max(70, int(w * 0.09)) + slogo = make_studio_logo(studio, max_height=logo_max_h) + if not slogo: + logo_path = STUDIO_FILES.get(studio) + logger.warning(" Poster stamp skipped: logo file not found for studio=%s path=%s", studio, logo_path) + return 0 + if studio == "paramountplus": + slogo = round_image_corners(slogo, radius=max(6, slogo.height // 5)) + if studio == "appletv": + slogo = slogo.filter(ImageFilter.UnsharpMask(radius=1.0, percent=180, threshold=2)) + s_pad = max(20, int(w * 0.025)) + sw, sh = slogo.size + positions = { + "top-left": (s_pad, s_pad), + "top-right": (w - sw - s_pad, s_pad), + "bottom-left": (s_pad, h - sh - s_pad), + "bottom-right": (w - sw - s_pad, h - sh - s_pad), + } + sx, sy = positions.get(studio_position, positions["top-left"]) + if studio == "appletv": + glow_pad = 10 + shadow_crop = Image.new("RGBA", (sw + glow_pad * 2, sh + glow_pad * 2), (0, 0, 0, 0)) + black = Image.new("RGBA", slogo.size, (0, 0, 0, 160)) + alpha = slogo.getchannel("A") + shadow_badge = Image.merge("RGBA", (*black.split()[:3], alpha)) + shadow_crop.paste(shadow_badge, (glow_pad + 2, glow_pad + 2), shadow_badge) + shadow_crop = shadow_crop.filter(ImageFilter.GaussianBlur(6)) + img = img.convert("RGBA") + img.alpha_composite(shadow_crop, (sx - glow_pad, sy - glow_pad)) + img = img.convert("RGB") + img.paste(slogo, (sx, sy), slogo) + status = await emby_upload_image(item_id, encode_jpeg_bytes(img), "Primary") + logger.info(" Poster stamped : status=%d", status) + return status + + +async def apply_bulk_assign_series_thumb(item: dict, *, backdrop_index: int = 0) -> dict: + if not can_bulk_assign_series_item(item): + raise HTTPException(status_code=400, detail="Series requires Emby primary, logo, and backdrop images before bulk assign can apply.") + + options = await apply_auto_studio_defaults(build_bulk_assign_series_options(item["id"], item["name"])) + options["backdrop_index"] = backdrop_index + + logger.info(" Rendering thumb : %s (backdrop #%d)", item["name"], backdrop_index) + + # Thumb + cache_key = get_generator_cache_key(options) + thumb_cache_path = get_thumb_cache_path(cache_key) + if not thumb_cache_path.exists(): + thumb_bytes, _ = await render_item_artwork(options, fallback_logo_to_first=True) + thumb_cache_path.write_bytes(thumb_bytes) + thumb_status = await emby_upload_image(item["id"], thumb_cache_path.read_bytes(), "Thumb") + + # Poster — stamp logo on existing primary image + studio = options.get("studio") or "none" + poster_status = 0 + if studio and studio != "none": + try: + poster_status = await stamp_studio_logo_on_poster( + item["id"], studio, options.get("studio_position", "top-left") + ) + except Exception as exc: + logger.error(" Poster stamp failed for %s: %s", item["name"], exc) + + logger.info(" Done : %s (thumb=%d, poster=%d)", item["name"], thumb_status, poster_status) + return { + "status": "applied", + "thumb_code": thumb_status, + "poster_code": poster_status, + } + + # --- Image helpers --- def cover_crop(img: Image.Image, width: int, height: int) -> Image.Image: @@ -897,30 +1157,171 @@ def wrap_text(text: str, font: ImageFont.FreeTypeFont, max_width: int) -> list: return lines +def parse_generator_request(body: dict) -> dict: + logo_index_raw = body.get("logo_index", None) + logo_index = int(logo_index_raw) if logo_index_raw is not None else None + return { + "item_id": body["item_id"], + "title": body.get("title", ""), + "bg_mode": body.get("bg_mode", "backdrop"), + "backdrop_index": int(body.get("backdrop_index", 0)), + "text_color": body.get("text_color", "#FFFFFF"), + "logo_align": body.get("logo_align", "bottom-center"), + "logo_scale": float(body.get("logo_scale", 1.3)), + "darkness": float(body.get("darkness", 0.0)), + "studio": body["studio"] if "studio" in body else "auto", + "studio_position": body.get("studio_position", "top-left"), + "new_episodes_tag": bool(body.get("new_episodes_tag", False)), + "season_finale_tag": bool(body.get("season_finale_tag", False)), + "generate_primary": bool(body.get("generate_primary", False)), + "logo_index": logo_index, + "primary_zoom": float(body.get("primary_zoom", 1.0)), + "primary_pan_x": float(body.get("primary_pan_x", 0.0)), + "primary_pan_y": float(body.get("primary_pan_y", -0.16)), + "thumb_zoom": float(body.get("thumb_zoom", 1.0)), + "thumb_pan_x": float(body.get("thumb_pan_x", 0.0)), + "thumb_pan_y": float(body.get("thumb_pan_y", 0.0)), + "upload_bg_id": body.get("upload_bg_id") or None, + } + + +async def get_logo_bytes_for_item( + item_id: str, + preferred_index: int | None, + *, + fallback_to_first: bool = False, +) -> bytes | None: + if preferred_index is not None and preferred_index < 0: + return None + if preferred_index is not None: + logo_bytes = await get_cached_emby_source_image(item_id, "Logo", preferred_index, optional=True) + if logo_bytes or not fallback_to_first or preferred_index == 0: + return logo_bytes + fallback_index = 0 if fallback_to_first else preferred_index + if fallback_index is not None and fallback_index < 0: + return None + return await get_cached_emby_source_image(item_id, "Logo", fallback_index, optional=True) + + +async def render_item_artwork(options: dict, *, fallback_logo_to_first: bool = False) -> tuple[bytes, bytes | None]: + poster_bytes = await get_cached_emby_source_image(options["item_id"], "Primary", optional=True) + logo_bytes = await get_logo_bytes_for_item( + options["item_id"], + options["logo_index"], + fallback_to_first=fallback_logo_to_first, + ) + if options["bg_mode"] == "upload" and options["upload_bg_id"]: + upload_path = get_upload_cache_path(options["upload_bg_id"]) + backdrop_bytes = upload_path.read_bytes() + else: + backdrop_bytes = await get_cached_emby_source_image( + options["item_id"], + "Backdrop", + options["backdrop_index"], + optional=True, + ) + + effective_bg_mode = "backdrop" if options["bg_mode"] == "upload" else options["bg_mode"] + thumb_bytes = generate_thumbnail( + poster_bytes, + options["title"], + bg_mode=effective_bg_mode, + text_color=options["text_color"], + logo_align=options["logo_align"], + logo_scale=options["logo_scale"], + darkness=options["darkness"], + studio=options["studio"], + studio_position=options["studio_position"], + new_episodes_tag=options["new_episodes_tag"], + season_finale_tag=options["season_finale_tag"], + logo_index=options["logo_index"], + logo_bytes=logo_bytes, + backdrop_bytes=backdrop_bytes, + primary_zoom=options["primary_zoom"], + primary_pan_x=options["primary_pan_x"], + primary_pan_y=options["primary_pan_y"], + thumb_zoom=options["thumb_zoom"], + thumb_pan_x=options["thumb_pan_x"], + thumb_pan_y=options["thumb_pan_y"], + ) + + primary_bytes = None + if options["generate_primary"]: + try: + primary_bytes = generate_primary_cover( + poster_bytes, + options["title"], + bg_mode=effective_bg_mode, + text_color=options["text_color"], + logo_align=options["logo_align"], + logo_scale=options["logo_scale"], + darkness=options["darkness"], + studio=options["studio"], + studio_position=options["studio_position"], + new_episodes_tag=options["new_episodes_tag"], + season_finale_tag=options["season_finale_tag"], + logo_index=options["logo_index"], + logo_bytes=logo_bytes, + backdrop_bytes=backdrop_bytes, + primary_zoom=options["primary_zoom"], + primary_pan_x=options["primary_pan_x"], + primary_pan_y=options["primary_pan_y"], + ) + except Exception: + primary_bytes = None + return thumb_bytes, primary_bytes + + +def get_generator_cache_key(options: dict) -> str: + return build_cache_key( + item_id=options["item_id"], + bg_mode=options["bg_mode"], + backdrop_index=options["backdrop_index"], + text_color=options["text_color"], + logo_align=options["logo_align"], + logo_scale=options["logo_scale"], + darkness=options["darkness"], + studio=options["studio"], + studio_position=options["studio_position"], + new_episodes_tag=options["new_episodes_tag"], + season_finale_tag=options["season_finale_tag"], + logo_index=options["logo_index"], + primary_zoom=options["primary_zoom"], + primary_pan_x=options["primary_pan_x"], + primary_pan_y=options["primary_pan_y"], + thumb_zoom=options["thumb_zoom"], + thumb_pan_x=options["thumb_pan_x"], + thumb_pan_y=options["thumb_pan_y"], + upload_bg_id=options["upload_bg_id"], + ) + + def build_cache_key( item_id: str, bg_mode: str = "backdrop", backdrop_index: int = 0, - custom_bg: str | None = None, text_color: str = "#FFFFFF", logo_align: str = "bottom-center", logo_scale: float = 1.3, darkness: float = 0.0, studio: str = "none", - studio_position: str = "bottom-right", + studio_position: str = "top-left", new_episodes_tag: bool = False, season_finale_tag: bool = False, logo_index: int | None = None, primary_zoom: float = 1.0, primary_pan_x: float = 0.0, primary_pan_y: float = -0.16, + thumb_zoom: float = 1.0, + thumb_pan_x: float = 0.0, + thumb_pan_y: float = 0.0, + upload_bg_id: str | None = None, ) -> str: payload = { "render_version": RENDER_VERSION, "item_id": item_id, "bg_mode": bg_mode, "backdrop_index": int(backdrop_index), - "custom_bg": custom_bg, "text_color": text_color, "logo_align": logo_align, "logo_scale": round(float(logo_scale), 4), @@ -933,6 +1334,10 @@ def build_cache_key( "primary_zoom": round(float(primary_zoom), 4), "primary_pan_x": round(float(primary_pan_x), 4), "primary_pan_y": round(float(primary_pan_y), 4), + "thumb_zoom": round(float(thumb_zoom), 4), + "thumb_pan_x": round(float(thumb_pan_x), 4), + "thumb_pan_y": round(float(thumb_pan_y), 4), + "upload_bg_id": upload_bg_id, } return hashlib.md5(repr(payload).encode("utf-8")).hexdigest() @@ -949,13 +1354,12 @@ def generate_thumbnail( poster_bytes: bytes, title: str, bg_mode: str = "backdrop", - custom_bg: str = None, text_color: str = "#FFFFFF", logo_align: str = "bottom-center", logo_scale: float = 1.3, darkness: float = 0.0, studio: str = "none", - studio_position: str = "bottom-right", + studio_position: str = "top-left", new_episodes_tag: bool = False, season_finale_tag: bool = False, logo_index: int | None = None, @@ -966,6 +1370,9 @@ def generate_thumbnail( primary_zoom: float = 1.0, primary_pan_x: float = 0.0, primary_pan_y: float = -0.16, + thumb_zoom: float = 1.0, + thumb_pan_x: float = 0.0, + thumb_pan_y: float = 0.0, ) -> bytes: # darkness is 0.0 (no dimming) → 1.0 (very dark); scale bg and vignette together @@ -975,7 +1382,7 @@ def generate_thumbnail( is_tall_layout = height >= int(width * 1.3) # --- Build background --- - if backdrop_bytes and bg_mode != "solid": + if backdrop_bytes: backdrop = load_image_from_bytes(backdrop_bytes, mode="RGB") if is_tall_layout: bg = build_tall_backdrop_background( @@ -988,31 +1395,31 @@ def generate_thumbnail( pan_y=primary_pan_y, ) else: - bg = cover_crop(backdrop, width, height) + bg = cover_crop_positioned(backdrop, width, height, zoom=thumb_zoom, pan_x=thumb_pan_x, pan_y=thumb_pan_y) bg = bg.point(lambda p: int(p * bg_dim)) - elif bg_mode == "solid" and custom_bg: - try: - bg_color = ImageColor.getrgb(custom_bg) - except ValueError: - bg_color = (15, 15, 25) - bg = Image.new("RGB", (width, height), bg_color) else: - poster = load_image_from_bytes(poster_bytes, mode="RGB") - if is_tall_layout: - bg = build_tall_backdrop_background( - poster, - width, - height, - dim_factor=bg_dim, - zoom=primary_zoom, - pan_x=primary_pan_x, - pan_y=primary_pan_y, - ) + if not poster_bytes: + bg = Image.new("RGB", (width, height), (20, 20, 28)) + bg = bg.point(lambda p: int(p * bg_dim)) + poster = None else: - # Blurred poster fallback for landscape thumbs. - bg = cover_crop(poster, width, height) - bg = bg.filter(ImageFilter.GaussianBlur(radius=20)) - bg = bg.point(lambda p: int(p * bg_dim * 0.85)) + poster = load_image_from_bytes(poster_bytes, mode="RGB") + if poster is not None: + if is_tall_layout: + bg = build_tall_backdrop_background( + poster, + width, + height, + dim_factor=bg_dim, + zoom=primary_zoom, + pan_x=primary_pan_x, + pan_y=primary_pan_y, + ) + else: + # Blurred poster fallback for landscape thumbs. + bg = cover_crop_positioned(poster, width, height, zoom=thumb_zoom, pan_x=thumb_pan_x, pan_y=thumb_pan_y) + bg = bg.filter(ImageFilter.GaussianBlur(radius=20)) + bg = bg.point(lambda p: int(p * bg_dim * 0.85)) # --- Vignette overlay tuned per logo position --- overlay = Image.new("RGBA", (width, height), (0, 0, 0, 0)) @@ -1127,15 +1534,15 @@ def generate_thumbnail( logo = logo.resize((lw, lh), Image.LANCZOS) logo_x, logo_y = logo_position(lw, lh) - # Drop shadow — blur only a tight crop around the logo, not the whole frame, - # to avoid a visible dark haze spreading across the backdrop. - blur_r = 5 + # Drop shadow — diffuse glow with no offset for a soft, professional look. + # Blur only a tight crop around the logo to avoid dark haze on the backdrop. + blur_r = 11 pad = blur_r * 3 _, _, _, a = logo.split() black = Image.new("RGB", (lw, lh), (0, 0, 0)) shadow_logo = Image.merge("RGBA", (*black.split()[:3], a)) shadow_crop = Image.new("RGBA", (lw + pad * 2, lh + pad * 2), (0, 0, 0, 0)) - shadow_crop.paste(shadow_logo, (pad + 4, pad + 4), shadow_logo) + shadow_crop.paste(shadow_logo, (pad, pad), shadow_logo) shadow_crop = shadow_crop.filter(ImageFilter.GaussianBlur(radius=blur_r)) bg = bg.convert("RGBA") bg.alpha_composite(shadow_crop, (logo_x - pad, logo_y - pad)) @@ -1160,9 +1567,16 @@ def generate_thumbnail( # --- Studio logo overlay --- if studio and studio != "none": - slogo = make_studio_logo(studio, max_height=46) + effective_studio_position = studio_position + studio_max_height = 63 if studio == "appletv" else 75 if studio == "paramountplus" else 60 if studio == "netflix" else 65 + slogo = make_studio_logo(studio, max_height=studio_max_height) if slogo: - s_pad = 22 + if studio == "paramountplus": + corner_r = max(6, slogo.height // 5) + slogo = round_image_corners(slogo, radius=corner_r) + if studio == "appletv": + slogo = slogo.filter(ImageFilter.UnsharpMask(radius=1.0, percent=180, threshold=2)) + s_pad = 18 if studio == "appletv" else 22 sw, sh = slogo.size top_y = s_pad if not banner else banner["y"] + banner["h"] + 10 positions = { @@ -1171,7 +1585,19 @@ def generate_thumbnail( "bottom-left": (s_pad, height - sh - s_pad), "bottom-right": (width - sw - s_pad, height - sh - s_pad), } - sx, sy = positions.get(studio_position, positions["bottom-right"]) + sx, sy = positions.get(effective_studio_position, positions["bottom-right"]) + if studio == "appletv": + # Soft drop-shadow so the wordmark reads cleanly on any backdrop. + glow_pad = 8 + shadow_crop = Image.new("RGBA", (sw + glow_pad * 2, sh + glow_pad * 2), (0, 0, 0, 0)) + black = Image.new("RGBA", slogo.size, (0, 0, 0, 160)) + alpha = slogo.getchannel("A") + shadow_badge = Image.merge("RGBA", (*black.split()[:3], alpha)) + shadow_crop.paste(shadow_badge, (glow_pad + 2, glow_pad + 2), shadow_badge) + shadow_crop = shadow_crop.filter(ImageFilter.GaussianBlur(5)) + bg = bg.convert("RGBA") + bg.alpha_composite(shadow_crop, (sx - glow_pad, sy - glow_pad)) + bg = bg.convert("RGB") bg.paste(slogo, (sx, sy), slogo) studio_box = (sx, sy, sw, sh) @@ -1193,13 +1619,12 @@ def generate_primary_cover( poster_bytes: bytes, title: str, bg_mode: str = "backdrop", - custom_bg: str | None = None, text_color: str = "#FFFFFF", logo_align: str = "bottom-center", logo_scale: float = 1.3, darkness: float = 0.0, studio: str = "none", - studio_position: str = "bottom-right", + studio_position: str = "top-left", new_episodes_tag: bool = False, season_finale_tag: bool = False, logo_index: int | None = None, @@ -1209,11 +1634,12 @@ def generate_primary_cover( primary_pan_x: float = 0.0, primary_pan_y: float = -0.16, ) -> bytes: + # Use the same backdrop_bytes as the thumb (backdrop image or uploaded photo). + # When none is provided, falls back to using poster_bytes as background. return generate_thumbnail( poster_bytes, title, bg_mode=bg_mode, - custom_bg=custom_bg, text_color=text_color, logo_align=logo_align, logo_scale=logo_scale, @@ -1233,6 +1659,328 @@ def generate_primary_cover( ) +def normalize_collection_target_type(value: str | None) -> str: + normalized = (value or "Thumb").strip().lower() + if normalized == "primary": + return "Primary" + if normalized == "thumb": + return "Thumb" + raise HTTPException(status_code=400, detail="Invalid target type.") + + +def normalize_collection_choice(value: str | None, allowed: set[str], default: str) -> str: + normalized = (value or default).strip().lower() + if normalized not in allowed: + return default + return normalized + + +def normalize_collection_text(value: str | None) -> str: + text = " ".join((value or "").split()) + if not text: + raise HTTPException(status_code=400, detail="Text is required.") + return text[:140] + + +def clamp_float(value: float, minimum: float, maximum: float) -> float: + return max(minimum, min(maximum, float(value))) + + +def get_collection_canvas_size(target_type: str) -> tuple[int, int]: + return (PRIMARY_WIDTH, PRIMARY_HEIGHT) if target_type == "Primary" else (THUMB_WIDTH, THUMB_HEIGHT) + + +def build_collection_cache_key( + *, + item_id: str, + target_type: str, + text: str, + text_color: str, + text_align: str, + text_position: str, + text_scale: float, + darkness: float, + zoom: float, + pan_x: float, + pan_y: float, + upload_bg_id: str | None, +) -> str: + payload = { + "render_version": COLLECTION_RENDER_VERSION, + "item_id": item_id, + "target_type": target_type, + "text": text, + "text_color": text_color, + "text_align": text_align, + "text_position": text_position, + "text_scale": round(float(text_scale), 4), + "darkness": round(float(darkness), 4), + "zoom": round(float(zoom), 4), + "pan_x": round(float(pan_x), 4), + "pan_y": round(float(pan_y), 4), + "upload_bg_id": upload_bg_id, + } + return hashlib.md5(repr(payload).encode("utf-8")).hexdigest() + + +def get_collection_cache_path(cache_key: str) -> Path: + return CACHE_DIR / f"{cache_key}.collection.png" + + +def build_collection_fallback_background(width: int, height: int) -> Image.Image: + bg = Image.new("RGB", (width, height), "#121723") + draw = ImageDraw.Draw(bg) + for y in range(height): + t = y / max(1, height - 1) + r = int(18 + (28 * t)) + g = int(24 + (18 * t)) + b = int(35 + (48 * t)) + draw.line([(0, y), (width, y)], fill=(r, g, b)) + accent_radius = max(width, height) // 2 + accent = Image.new("RGBA", (width, height), (0, 0, 0, 0)) + accent_draw = ImageDraw.Draw(accent) + accent_draw.ellipse( + ( + width - accent_radius, + -accent_radius // 3, + width + accent_radius // 2, + accent_radius + accent_radius // 6, + ), + fill=(74, 111, 255, 50), + ) + accent_draw.ellipse( + ( + -accent_radius // 2, + height - accent_radius, + accent_radius, + height + accent_radius // 3, + ), + fill=(52, 211, 153, 34), + ) + return Image.alpha_composite(bg.convert("RGBA"), accent).convert("RGB") + + +def fit_collection_text_block( + text: str, + *, + width: int, + height: int, + text_scale: float, +) -> tuple[ImageFont.FreeTypeFont, list[str], list[tuple[int, int, int, int]], int]: + is_tall = height > width + max_width = width - int(width * 0.12) + max_height = int(height * (0.54 if is_tall else 0.42)) + max_lines = 5 if is_tall else 3 + base_size = int(height * (0.07 if is_tall else 0.12) * text_scale) + min_size = 34 if is_tall else 24 + + size = max(base_size, min_size) + while size >= min_size: + font = get_arial_bold_font(size) + lines = wrap_text(text, font, max_width) + boxes = [font.getbbox(line) for line in lines] if lines else [] + heights = [(box[3] - box[1]) for box in boxes] + line_gap = max(8, int(size * 0.16)) + total_height = sum(heights) + line_gap * max(0, len(lines) - 1) + if lines and len(lines) <= max_lines and total_height <= max_height: + return font, lines, boxes, line_gap + size -= 4 + + font = get_arial_bold_font(min_size) + lines = wrap_text(text, font, max_width) + boxes = [font.getbbox(line) for line in lines] if lines else [] + line_gap = max(8, int(min_size * 0.16)) + return font, lines or [text], boxes or [font.getbbox(text)], line_gap + + +def generate_collection_art( + *, + source_bytes: bytes | None, + target_type: str, + text: str, + text_color: str, + text_align: str, + text_position: str, + text_scale: float, + darkness: float, + zoom: float, + pan_x: float, + pan_y: float, +) -> bytes: + width, height = get_collection_canvas_size(target_type) + darkness = clamp_float(darkness, 0.0, 1.0) + zoom = clamp_float(zoom, PRIMARY_MIN_ZOOM, PRIMARY_MAX_ZOOM) + pan_x = clamp_float(pan_x, -1.0, 1.0) + pan_y = clamp_float(pan_y, -1.0, 1.0) + + if source_bytes: + source = load_image_from_bytes(source_bytes, mode="RGB") + bg = cover_crop_positioned(source, width, height, zoom=zoom, pan_x=pan_x, pan_y=pan_y) + else: + bg = build_collection_fallback_background(width, height) + + if darkness > 0: + bg = bg.point(lambda p: int(p * (1.0 - (darkness * 0.42)))) + + font, lines, boxes, line_gap = fit_collection_text_block( + text, + width=width, + height=height, + text_scale=clamp_float(text_scale, 0.65, 1.8), + ) + try: + text_fill = ImageColor.getrgb(text_color) + except ValueError: + text_fill = (255, 255, 255) + padding_x = max(34, int(width * 0.06)) + padding_y = max(28, int(height * 0.05)) + line_heights = [(box[3] - box[1]) for box in boxes] + line_widths = [(box[2] - box[0]) for box in boxes] + text_block_w = max(line_widths) if line_widths else 0 + text_block_h = sum(line_heights) + line_gap * max(0, len(lines) - 1) + + if text_align == "left": + text_x = padding_x + elif text_align == "right": + text_x = width - padding_x - text_block_w + else: + text_x = (width - text_block_w) // 2 + + if text_position == "top": + text_y = padding_y + elif text_position == "center": + text_y = max(padding_y, (height - text_block_h) // 2) + else: + text_y = height - padding_y - text_block_h + + panel_pad_x = max(22, int(font.size * 0.45)) + panel_pad_y = max(18, int(font.size * 0.34)) + panel_left = max(14, text_x - panel_pad_x) + panel_top = max(14, text_y - panel_pad_y) + panel_right = min(width - 14, text_x + text_block_w + panel_pad_x) + panel_bottom = min(height - 14, text_y + text_block_h + panel_pad_y) + + overlay = Image.new("RGBA", (width, height), (0, 0, 0, 0)) + overlay_draw = ImageDraw.Draw(overlay) + overlay_draw.rounded_rectangle( + [(panel_left, panel_top), (panel_right, panel_bottom)], + radius=max(18, int(min(width, height) * 0.024)), + fill=(0, 0, 0, 92 + int(darkness * 80)), + outline=(255, 255, 255, 18), + width=1, + ) + + bg = Image.alpha_composite(bg.convert("RGBA"), overlay).convert("RGB") + draw = ImageDraw.Draw(bg) + stroke_width = max(2, int(font.size * 0.08)) + current_y = text_y + for line, box, line_height, line_width in zip(lines, boxes, line_heights, line_widths): + if text_align == "left": + line_x = padding_x - box[0] + elif text_align == "right": + line_x = width - padding_x - line_width - box[0] + else: + line_x = ((width - line_width) // 2) - box[0] + draw.text( + (line_x, current_y - box[1]), + line, + font=font, + fill=text_fill, + stroke_width=stroke_width, + stroke_fill=(0, 0, 0), + ) + current_y += line_height + line_gap + + buf = io.BytesIO() + bg.save(buf, format="PNG") + buf.seek(0) + return buf.getvalue() + + +def parse_collection_render_request(body: dict) -> dict: + try: + item_id = str(body["item_id"]) + except KeyError as exc: + raise HTTPException(status_code=400, detail="Missing item id.") from exc + + return { + "item_id": item_id, + "target_type": normalize_collection_target_type(body.get("target_type", "Thumb")), + "text": normalize_collection_text(body.get("text") or body.get("title")), + "text_color": body.get("text_color", "#FFFFFF"), + "text_align": normalize_collection_choice(body.get("text_align"), {"left", "center", "right"}, "center"), + "text_position": normalize_collection_choice(body.get("text_position"), {"top", "center", "bottom"}, "bottom"), + "text_scale": clamp_float(body.get("text_scale", 1.0), 0.65, 1.8), + "darkness": clamp_float(body.get("darkness", 0.18), 0.0, 0.85), + "zoom": clamp_float(body.get("zoom", 1.0), PRIMARY_MIN_ZOOM, PRIMARY_MAX_ZOOM), + "pan_x": clamp_float(body.get("pan_x", 0.0), -1.0, 1.0), + "pan_y": clamp_float(body.get("pan_y", 0.0), -1.0, 1.0), + "upload_bg_id": body.get("upload_bg_id") or None, + } + + +async def get_collection_source_bytes(item_id: str, target_type: str, upload_bg_id: str | None) -> bytes | None: + if upload_bg_id: + return get_upload_cache_path(upload_bg_id).read_bytes() + + fallback_types = [target_type] + if target_type == "Thumb": + fallback_types.append("Primary") + else: + fallback_types.append("Thumb") + + seen: set[str] = set() + for image_type in fallback_types: + if image_type in seen: + continue + seen.add(image_type) + image_bytes = await emby_get_image_optional(item_id, image_type) + if image_bytes: + return image_bytes + return None + + +async def render_collection_art_preview(options: dict) -> tuple[str, bytes]: + cache_key = build_collection_cache_key( + item_id=options["item_id"], + target_type=options["target_type"], + text=options["text"], + text_color=options["text_color"], + text_align=options["text_align"], + text_position=options["text_position"], + text_scale=options["text_scale"], + darkness=options["darkness"], + zoom=options["zoom"], + pan_x=options["pan_x"], + pan_y=options["pan_y"], + upload_bg_id=options["upload_bg_id"], + ) + cache_path = get_collection_cache_path(cache_key) + if cache_path.exists(): + return cache_key, cache_path.read_bytes() + + source_bytes = await get_collection_source_bytes( + options["item_id"], + options["target_type"], + options["upload_bg_id"], + ) + preview_bytes = generate_collection_art( + source_bytes=source_bytes, + target_type=options["target_type"], + text=options["text"], + text_color=options["text_color"], + text_align=options["text_align"], + text_position=options["text_position"], + text_scale=options["text_scale"], + darkness=options["darkness"], + zoom=options["zoom"], + pan_x=options["pan_x"], + pan_y=options["pan_y"], + ) + cache_path.write_bytes(preview_bytes) + return cache_key, preview_bytes + + # --- API Routes --- @app.get("/", response_class=HTMLResponse) @@ -1240,11 +1988,121 @@ async def index(request: Request): return templates.TemplateResponse(request, "index.html") +@app.get("/collections", response_class=HTMLResponse) +async def collections_page(request: Request): + return templates.TemplateResponse(request, "collections.html") + + @app.get("/airing", response_class=HTMLResponse) async def airing_page(request: Request): return templates.TemplateResponse(request, "airing.html") +@app.get("/bulk-assign", response_class=HTMLResponse) +async def bulk_assign_page(request: Request): + return templates.TemplateResponse(request, "bulk_assign.html") + + +@app.get("/api/bulk-assign/series") +async def get_bulk_assign_series( + q: str = Query(""), + start: int = Query(0, ge=0), + limit: int = Query(24, ge=1, le=60), +): + params = { + "IncludeItemTypes": "Series", + "Recursive": "true", + "SortBy": "SortName", + "SortOrder": "Ascending", + "Fields": "ProductionYear", + "StartIndex": str(start), + "Limit": str(limit), + "ImageTypeLimit": "1", + "EnableImageTypes": "Primary,Logo,Backdrop", + } + if q.strip(): + params["SearchTerm"] = q.strip() + + data = await emby_get("/Items", params) + items: list[dict] = [] + for item in data.get("Items", []): + image_tags = item.get("ImageTags") or {} + backdrop_tags = item.get("BackdropImageTags") or [] + entry = { + "id": item["Id"], + "name": item.get("Name", ""), + "year": item.get("ProductionYear"), + "type": item.get("Type", ""), + "poster_url": f"/api/poster/{item['Id']}?w=180&h=270&q=84" if "Primary" in image_tags else None, + "has_primary": "Primary" in image_tags, + "has_logo": "Logo" in image_tags, + "has_backdrop": bool(backdrop_tags), + "backdrop_count": len(backdrop_tags), + } + entry["can_bulk_assign"] = can_bulk_assign_series_item(entry) + items.append(entry) + + total = int(data.get("TotalRecordCount", start + len(items))) + return { + "items": items, + "start": start, + "limit": limit, + "total": total, + "has_more": start + len(items) < total, + "query": q.strip(), + } + + +@app.get("/api/bulk-assign/movies") +async def get_bulk_assign_movies( + q: str = Query(""), + start: int = Query(0, ge=0), + limit: int = Query(24, ge=1, le=60), +): + params = { + "IncludeItemTypes": "Movie", + "Recursive": "true", + "SortBy": "SortName", + "SortOrder": "Ascending", + "Fields": "ProductionYear", + "StartIndex": str(start), + "Limit": str(limit), + "ImageTypeLimit": "1", + "EnableImageTypes": "Primary,Logo,Backdrop", + } + if q.strip(): + params["SearchTerm"] = q.strip() + + data = await emby_get("/Items", params) + items: list[dict] = [] + for item in data.get("Items", []): + image_tags = item.get("ImageTags") or {} + backdrop_tags = item.get("BackdropImageTags") or [] + entry = { + "id": item["Id"], + "name": item.get("Name", ""), + "year": item.get("ProductionYear"), + "type": item.get("Type", ""), + "poster_url": f"/api/poster/{item['Id']}?w=180&h=270&q=84" if "Primary" in image_tags else None, + "has_primary": "Primary" in image_tags, + "has_logo": "Logo" in image_tags, + "has_backdrop": bool(backdrop_tags), + "backdrop_count": len(backdrop_tags), + } + entry["can_bulk_assign"] = can_bulk_assign_series_item(entry) + items.append(entry) + + total = int(data.get("TotalRecordCount", start + len(items))) + return { + "items": items, + "start": start, + "limit": limit, + "total": total, + "has_more": start + len(items) < total, + "query": q.strip(), + } + + @app.get("/api/airing") async def get_airing_titles( page: int = Query(1, ge=1), @@ -1285,11 +2143,7 @@ async def apply_new_season_banner(request: Request): item = next((entry for entry in snapshot["items"] if entry["id"] == item_id), None) if item is None: raise HTTPException(status_code=404, detail="Series was not found in the current airing snapshot.") - can_apply_new_season = bool(item["eligible_new_season"]) or ( - (item.get("status") or "").strip().lower() == "continuing" - and (item.get("selected_week_air_at") or item.get("next_air_at")) - ) - if not can_apply_new_season: + if not can_apply_new_season_snapshot_item(item): raise HTTPException( status_code=400, detail=( @@ -1298,102 +2152,339 @@ async def apply_new_season_banner(request: Request): f"from the season premiere." ), ) - - title = body.get("title") or item["name"] - cache_key = build_cache_key( - item_id=item_id, - bg_mode="backdrop", - backdrop_index=0, - custom_bg=None, - text_color="#FFFFFF", - logo_align="bottom-center", - logo_scale=1.3, - darkness=0.0, - studio="none", - studio_position="bottom-right", - new_episodes_tag=True, - season_finale_tag=False, - logo_index=0, + return await apply_new_season_artwork_for_snapshot_item( + item, + title=body.get("title") or item["name"], + generate_primary=generate_primary, ) - thumb_cache_path = get_thumb_cache_path(cache_key) - primary_cache_path = get_primary_cache_path(cache_key) - if not thumb_cache_path.exists() or (generate_primary and not primary_cache_path.exists()): - poster_bytes = await emby_get_image(item_id, "Primary") - logo_bytes = await emby_get_image_optional(item_id, "Logo", 0) - backdrop_bytes = await emby_get_image_optional(item_id, "Backdrop", 0) - thumb_bytes = generate_thumbnail( - poster_bytes, - title, - bg_mode="backdrop", - text_color="#FFFFFF", - logo_align="bottom-center", - logo_scale=1.3, - darkness=0.0, - studio="none", - studio_position="bottom-right", - new_episodes_tag=True, - season_finale_tag=False, - logo_index=0, - logo_bytes=logo_bytes, - backdrop_bytes=backdrop_bytes, - ) - thumb_cache_path.write_bytes(thumb_bytes) - if generate_primary: - primary_bytes = generate_primary_cover( - poster_bytes, - title, - bg_mode="backdrop", - text_color="#FFFFFF", - logo_align="bottom-center", - logo_scale=1.3, - darkness=0.0, - studio="none", - studio_position="bottom-right", - new_episodes_tag=True, - season_finale_tag=False, - logo_index=0, - logo_bytes=logo_bytes, - backdrop_bytes=backdrop_bytes, - ) - primary_cache_path.write_bytes(primary_bytes) - thumb_status = await emby_upload_image(item_id, thumb_cache_path.read_bytes(), "Thumb") +@app.post("/api/bulk-assign/apply") +@app.post("/api/airing/apply-new-season/bulk") +async def bulk_apply_new_season_banner(request: Request): + body = await request.json() + item_ids_raw = body.get("item_ids") or [] + if not isinstance(item_ids_raw, list): + raise HTTPException(status_code=400, detail="item_ids must be a list.") - primary_status = None - primary_error = None + item_ids: list[str] = [] + seen_item_ids: set[str] = set() + for raw_item_id in item_ids_raw: + item_id = str(raw_item_id or "").strip() + if not item_id or item_id in seen_item_ids: + continue + seen_item_ids.add(item_id) + item_ids.append(item_id) + if not item_ids: + raise HTTPException(status_code=400, detail="At least one series must be selected.") + + generate_primary = bool(body.get("generate_primary", False)) if generate_primary: + raise HTTPException(status_code=400, detail="Bulk assign only applies thumb artwork.") + + items = await emby_get_all("/Items", { + "Ids": ",".join(item_ids), + "IncludeItemTypes": "Series,Movie", + "Recursive": "true", + "Fields": "ProductionYear", + "ImageTypeLimit": "1", + "EnableImageTypes": "Primary,Logo,Backdrop", + }) + items_by_id = {} + for entry in items: + image_tags = entry.get("ImageTags") or {} + backdrop_tags = entry.get("BackdropImageTags") or [] + item = { + "id": entry.get("Id"), + "name": entry.get("Name", ""), + "has_primary": "Primary" in image_tags, + "has_logo": "Logo" in image_tags, + "has_backdrop": bool(backdrop_tags), + } + if item["id"]: + items_by_id[item["id"]] = item + + applied: list[str] = [] + skipped_not_found: list[str] = [] + skipped_missing_assets: list[str] = [] + failed: list[dict] = [] + + logger.info("=== Bulk Apply Selected: %d %s item(s) ===", len(item_ids), emby_type) + for i, item_id in enumerate(item_ids): + item = items_by_id.get(item_id) + if item is None: + logger.warning("[%d/%d] ID %s — not found in Emby", i + 1, len(item_ids), item_id) + skipped_not_found.append(item_id) + continue + if not can_bulk_assign_series_item(item): + logger.info("[%d/%d] %s — skipped (missing logo or backdrop)", i + 1, len(item_ids), item["name"]) + skipped_missing_assets.append(item.get("name") or item_id) + continue + logger.info("[%d/%d] %s", i + 1, len(item_ids), item["name"]) try: - primary_status = await emby_upload_image(item_id, primary_cache_path.read_bytes(), "Primary") + await apply_bulk_assign_series_thumb(item) + applied.append(item.get("name") or item_id) except Exception as exc: - primary_error = str(exc) + err_msg = exc.detail if isinstance(exc, HTTPException) else str(exc) + logger.error(" FAILED: %s — %s", item["name"], err_msg) + failed.append({"name": item.get("name") or item_id, "error": err_msg}) + + logger.info( + "=== Bulk Apply Selected complete: %d applied, %d skipped, %d not found, %d failed ===", + len(applied), len(skipped_missing_assets), len(skipped_not_found), len(failed), + ) + return { + "status": "completed", + "requested_count": len(item_ids), + "applied_count": len(applied), + "skipped_missing_assets_count": len(skipped_missing_assets), + "skipped_not_found_count": len(skipped_not_found), + "failed_count": len(failed), + "applied": applied[:12], + "failed": failed[:12], + } + + +@app.post("/api/bulk-assign/apply-all") +async def bulk_apply_all_eligible(request: Request): + body = await request.json() + item_type = str(body.get("item_type") or "series").strip().lower() + if item_type not in ("series", "movie"): + raise HTTPException(status_code=400, detail="item_type must be 'series' or 'movie'.") + emby_type = "Series" if item_type == "series" else "Movie" + + logger.info("=== Bulk Apply All: fetching all %s items from Emby ===", emby_type) + all_entries = await emby_get_all("/Items", { + "IncludeItemTypes": emby_type, + "Recursive": "true", + "Fields": "ProductionYear", + "ImageTypeLimit": "1", + "EnableImageTypes": "Primary,Logo,Backdrop", + }) + logger.info("Fetched %d total %s entries from Emby", len(all_entries), emby_type) + + eligible = [] + skipped_missing = 0 + for entry in all_entries: + image_tags = entry.get("ImageTags") or {} + backdrop_tags = entry.get("BackdropImageTags") or [] + item = { + "id": entry.get("Id"), + "name": entry.get("Name", ""), + "has_primary": "Primary" in image_tags, + "has_logo": "Logo" in image_tags, + "has_backdrop": bool(backdrop_tags), + "backdrop_count": len(backdrop_tags), + } + if item["id"] and can_bulk_assign_series_item(item): + eligible.append(item) + else: + skipped_missing += 1 + + logger.info( + "%d eligible (logo+backdrop present), %d skipped (missing assets)", + len(eligible), skipped_missing, + ) + + applied: list[str] = [] + failed: list[dict] = [] + + for i, item in enumerate(eligible): + position = i + 1 + # Every 5th item: try to use a second backdrop for variety + if position % 5 == 0: + if item["backdrop_count"] < 2: + logger.info( + "[%d/%d] %s — every-5th variant, only 1 backdrop; requesting more from Emby…", + position, len(eligible), item["name"], + ) + new_count = await emby_refresh_backdrops(item["id"]) + item["backdrop_count"] = new_count + logger.info( + " Emby returned %d backdrop(s) after refresh", new_count, + ) + if item["backdrop_count"] >= 2: + backdrop_index = random.randint(0, 1) + logger.info( + "[%d/%d] %s — every-5th variant, using random backdrop #%d", + position, len(eligible), item["name"], backdrop_index, + ) + else: + backdrop_index = 0 + logger.info( + "[%d/%d] %s — every-5th variant, only 1 backdrop available, using #0", + position, len(eligible), item["name"], + ) + else: + backdrop_index = 0 + logger.info("[%d/%d] %s", position, len(eligible), item["name"]) + + try: + await apply_bulk_assign_series_thumb(item, backdrop_index=backdrop_index) + applied.append(item.get("name") or item["id"]) + except Exception as exc: + err_msg = exc.detail if isinstance(exc, HTTPException) else str(exc) + logger.error(" FAILED: %s — %s", item["name"], err_msg) + failed.append({"name": item.get("name") or item["id"], "error": err_msg}) + + logger.info( + "=== Bulk Apply All complete: %d applied, %d failed ===", + len(applied), len(failed), + ) + if failed: + for f in failed: + logger.warning(" Failed: %s — %s", f["name"], f["error"]) return { - "status": "applied", - "thumb_code": thumb_status, - "primary_code": primary_status, - "primary_attempted": generate_primary, - "primary_error": primary_error, + "status": "completed", + "eligible_count": len(eligible), + "applied_count": len(applied), + "failed_count": len(failed), + "applied": applied[:20], + "failed": failed[:20], } +@app.post("/api/bulk-reset/studio") +async def bulk_reset_studio(request: Request): + body = await request.json() + studio_key = (body.get("studio_key") or "").strip().lower() + if studio_key not in STUDIO_EMBY_NAMES: + raise HTTPException(status_code=400, detail=f"Unknown studio key: {studio_key}") + + studio_ids = await find_studio_ids(studio_key) + if not studio_ids: + return {"reset": 0, "skipped": 0, "message": "No matching studio found in Emby."} + + items = await emby_get_all("/Items", { + "StudioIds": ",".join(studio_ids), + "IncludeItemTypes": "Series,Movie", + "Recursive": "true", + "Fields": "", + "ImageTypeLimit": "0", + }) + + reset = 0 + skipped = 0 + for entry in items: + item_id = entry.get("Id") + if not item_id: + continue + try: + await emby_delete_image(item_id, "Thumb") + await emby_delete_image(item_id, "Primary") + # Invalidate local image caches so the next fetch gets the fresh Emby version + for image_type in ("Primary", "Thumb"): + cache_path = get_emby_image_cache_path(item_id, image_type) + cache_path.unlink(missing_ok=True) + get_emby_image_miss_path(item_id, "Primary").unlink(missing_ok=True) + (CLEAN_POSTER_CACHE_DIR / f"{item_id}.img").unlink(missing_ok=True) + # Ask Emby to fill in the now-missing Primary from its image providers (e.g. TMDB). + # ReplaceAllImages=false is intentional: Emby will only download images that are + # absent, so once we upload our stamped poster it won't be overwritten by this job. + resp = await emby_request( + "POST", + f"/Items/{item_id}/Refresh", + params={ + "MetadataRefreshMode": "None", + "ImageRefreshMode": "FullRefresh", + "ReplaceAllImages": "false", + "ReplaceAllMetadata": "false", + }, + ) + if resp.status_code not in (200, 204): + logger.warning(" Emby refresh returned %d for item %s", resp.status_code, item_id) + reset += 1 + except Exception as exc: + logger.error(" Reset failed for item %s: %s", item_id, exc) + skipped += 1 + + if reset: + # Give Emby a head start re-downloading images before the client re-applies. + await asyncio.sleep(5) + + return {"reset": reset, "skipped": skipped} + + +@app.post("/api/clean-poster-cache/clear") +async def clear_clean_poster_cache(request: Request): + """Delete clean poster baselines for given item IDs (or all if none supplied). + + Call this after restoring original posters in Emby so the next bulk assign + picks up the fresh clean image rather than a previously-stamped one. + """ + body = await request.json() + item_ids = body.get("item_ids") or [] + if item_ids: + cleared = 0 + for item_id in item_ids: + path = CLEAN_POSTER_CACHE_DIR / f"{str(item_id).strip()}.img" + if path.exists(): + path.unlink() + cleared += 1 + return {"cleared": cleared, "scope": "selected"} + else: + files = list(CLEAN_POSTER_CACHE_DIR.glob("*.img")) + for f in files: + f.unlink(missing_ok=True) + return {"cleared": len(files), "scope": "all"} + + +async def find_studio_ids(studio_key: str) -> list[str]: + """Return Emby studio IDs for a given studio key by searching known display names.""" + ids: list[str] = [] + seen: set[str] = set() + for name in STUDIO_EMBY_NAMES.get(studio_key, []): + data = await emby_get("/Studios", {"SearchTerm": name, "Limit": "10"}) + for item in data.get("Items", []): + sid = item.get("Id") + sname = (item.get("Name") or "").lower() + if sid and sid not in seen and (name.lower() in sname or sname in name.lower()): + ids.append(sid) + seen.add(sid) + return ids + + @app.get("/api/search") async def search_items( q: str = Query(..., min_length=1), type: str = "Movie,Series", start: int = Query(0, ge=0), - limit: int = Query(12, ge=1, le=24), + limit: int = Query(12, ge=1, le=500), ): - data = await emby_get("/Items", { - "SearchTerm": q, - "IncludeItemTypes": type, - "Recursive": "true", - "Fields": "ProductionYear", - "StartIndex": str(start), - "Limit": str(limit), - "ImageTypeLimit": "1", - "EnableImageTypes": "Primary,Logo,Backdrop", - }) + studio_key = STUDIO_SEARCH_ALIASES.get(q.strip().lower()) + if studio_key: + studio_ids = await find_studio_ids(studio_key) + if studio_ids: + base_params = { + "StudioIds": ",".join(studio_ids), + "IncludeItemTypes": type, + "Recursive": "true", + "Fields": "ProductionYear,Studios", + "SortBy": "SortName", + "SortOrder": "Ascending", + "ImageTypeLimit": "1", + "EnableImageTypes": "Primary,Logo,Backdrop", + } + if limit >= 100: + # Bulk fetch — paginate through everything Emby has + all_items = await emby_get_all("/Items", base_params) + data = {"Items": all_items, "TotalRecordCount": len(all_items)} + else: + data = await emby_get("/Items", {**base_params, "StartIndex": str(start), "Limit": str(limit)}) + else: + data = {"Items": [], "TotalRecordCount": 0} + else: + data = await emby_get("/Items", { + "SearchTerm": q, + "IncludeItemTypes": type, + "Recursive": "true", + "Fields": "ProductionYear,Studios", + "StartIndex": str(start), + "Limit": str(limit), + "ImageTypeLimit": "1", + "EnableImageTypes": "Primary,Logo,Backdrop", + }) items = [] for item in data.get("Items", []): image_tags = item.get("ImageTags", {}) @@ -1404,6 +2495,7 @@ async def search_items( "year": item.get("ProductionYear", ""), "type": item.get("Type", ""), "has_logo": "Logo" in image_tags, + "auto_studio": infer_auto_studio_key(item), "backdrop_count": len(backdrop_count), "poster_url": f"/api/poster/{item['Id']}?w=72&h=108&q=72", }) @@ -1414,6 +2506,88 @@ async def search_items( "limit": limit, "total": total, "has_more": start + len(items) < total, + "studio_key": studio_key, + } + + +@app.get("/api/categories") +async def get_categories(): + data = await emby_get("/Genres", { + "Recursive": "true", + "SortBy": "SortName", + "SortOrder": "Ascending", + }) + items = [] + for item in data.get("Items", []): + items.append({ + "id": item.get("Id"), + "name": item.get("Name", ""), + "item_count": item.get("ItemCount") or item.get("ChildCount") or 0, + }) + items.sort(key=lambda entry: entry["name"].lower()) + return {"items": items} + + +@app.get("/api/collections") +async def get_collections( + q: str = Query(""), + start: int = Query(0, ge=0), + limit: int = Query(24, ge=1, le=60), +): + params = { + "IncludeItemTypes": "BoxSet", + "Recursive": "true", + "SortBy": "SortName", + "SortOrder": "Ascending", + "Fields": "ChildCount", + "StartIndex": str(start), + "Limit": str(limit), + "ImageTypeLimit": "1", + "EnableImageTypes": "Primary,Thumb", + } + if q.strip(): + params["SearchTerm"] = q.strip() + + data = await emby_get("/Items", params) + items = [] + for item in data.get("Items", []): + image_tags = item.get("ImageTags") or {} + items.append({ + "id": item["Id"], + "name": item.get("Name", ""), + "type": item.get("Type", ""), + "child_count": item.get("ChildCount") or 0, + "poster_url": f"/api/poster/{item['Id']}?w=72&h=108&q=72" if "Primary" in image_tags else None, + }) + + total = int(data.get("TotalRecordCount", start + len(items))) + return { + "items": items, + "start": start, + "limit": limit, + "total": total, + "has_more": start + len(items) < total, + } + + +@app.get("/api/collections/{item_id}/artwork") +async def get_collection_artwork(item_id: str): + images = await emby_get(f"/Items/{item_id}/Images") + artwork = { + "Primary": {"exists": False, "url": None}, + "Thumb": {"exists": False, "url": None}, + } + for image in images: + image_type = image.get("ImageType") + if image_type not in artwork or artwork[image_type]["exists"]: + continue + artwork[image_type] = { + "exists": True, + "url": f"/api/item-image/{item_id}?type={image_type}&w={300 if image_type == 'Primary' else 420}&h={450 if image_type == 'Primary' else 236}&q=88", + } + return { + "primary": artwork["Primary"], + "thumb": artwork["Thumb"], } @@ -1513,273 +2687,89 @@ async def get_cached_primary_preview(cache_key: str): ) -@app.get("/api/artwork/providers") -async def list_artwork_search_providers(): - return { - "providers": [ - {"key": provider.key, "label": provider.label} - for provider in IMAGE_SEARCH_PROVIDERS.values() - ], - "targets": [ - { - "key": key, - "label": value["label"], - "width": value["width"], - "height": value["height"], - "emby_type": value["emby_type"], - } - for key, value in EDITOR_TARGETS.items() - ], - } - - -@app.get("/api/artwork/search") -async def search_external_artwork( - q: str = Query(..., min_length=2), - provider: str = Query("wikimedia"), - limit: int = Query(12, ge=1, le=24), -): - search_provider = IMAGE_SEARCH_PROVIDERS.get(provider) - if search_provider is None: - raise HTTPException(status_code=400, detail="Unknown image search provider.") - try: - results = await search_provider.search(q, limit=limit) - except httpx.HTTPError as exc: - raise HTTPException(status_code=502, detail=f"Image search failed: {exc}") from exc - return { - "provider": provider, - "items": [result.__dict__ for result in results], - } - - -@app.post("/api/artwork/import") -async def import_remote_artwork(request: Request): - body = await request.json() - image_url = body.get("image_url", "") - if not image_url: - raise HTTPException(status_code=400, detail="Missing image URL.") - parsed = urllib.parse.urlparse(image_url) - if parsed.scheme not in {"http", "https"}: - raise HTTPException(status_code=400, detail="Only http and https image URLs are supported.") - - client = get_http_client() - try: - r = await client.get(image_url, follow_redirects=True) - r.raise_for_status() - except httpx.HTTPError as exc: - raise HTTPException(status_code=502, detail=f"Could not import remote image: {exc}") from exc - - content_type = r.headers.get("content-type", "").split(";")[0].lower() - if content_type and not content_type.startswith("image/"): - raise HTTPException(status_code=400, detail="Remote URL did not return an image.") - image_bytes = r.content - if len(image_bytes) > REMOTE_IMAGE_MAX_BYTES: - raise HTTPException(status_code=400, detail="Remote image is too large.") - width, height, fmt = validate_image_bytes(image_bytes) - import_id = hashlib.sha256(image_bytes).hexdigest() - import_path = IMPORT_CACHE_DIR / f"{import_id}.img" - if not import_path.exists(): - import_path.write_bytes(image_bytes) - return { - "id": import_id, - "width": width, - "height": height, - "format": fmt, - "preview_url": f"/api/artwork/import/{import_id}", - } - - -@app.get("/api/artwork/import/{import_id}") -async def get_imported_artwork(import_id: str): - image_bytes = get_import_cache_path(import_id).read_bytes() - return Response( - content=image_bytes, - media_type="image/jpeg", - headers={"Cache-Control": "public, max-age=86400"}, - ) - - -@app.post("/api/artwork/export") -async def export_artwork(request: Request): - body = await request.json() - source_bytes = await resolve_editor_source_bytes(body) - cache_key = build_editor_cache_key(body) - cache_path = CACHE_DIR / f"{cache_key}.editor.jpg" - if not cache_path.exists(): - cache_path.write_bytes(render_artwork_editor_image( - source_bytes, - target_mode=body.get("target_mode", "poster"), - fit_mode=body.get("fit_mode", "cover"), - fill_mode=body.get("fill_mode", "blur"), - fill_color=body.get("fill_color", "#101318"), - zoom=float(body.get("zoom", 1.0)), - pan_x=float(body.get("pan_x", 0.0)), - pan_y=float(body.get("pan_y", 0.0)), - quality=int(body.get("quality", 95)), - )) - target = get_editor_target(body.get("target_mode", "poster")) - return Response( - content=cache_path.read_bytes(), - media_type="image/jpeg", - headers={ - "Cache-Control": "no-store", - "X-Editor-Cache-Key": cache_key, - "X-Editor-Width": str(target["width"]), - "X-Editor-Height": str(target["height"]), - "X-Editor-Emby-Type": target["emby_type"], - }, - ) - - -@app.get("/api/artwork/export/{cache_key}") -async def get_exported_artwork(cache_key: str): +@app.get("/api/cache/{cache_key}/collection") +async def get_cached_collection_preview(cache_key: str): if not cache_key or any(ch not in "0123456789abcdef" for ch in cache_key.lower()): - raise HTTPException(status_code=400, detail="Invalid export cache key.") - cache_path = CACHE_DIR / f"{cache_key}.editor.jpg" + raise HTTPException(status_code=400, detail="Invalid cache key.") + cache_path = get_collection_cache_path(cache_key) if not cache_path.exists(): - raise HTTPException(status_code=404, detail="Export was not found in cache.") + raise HTTPException(status_code=404, detail="Collection preview not found.") return Response( content=cache_path.read_bytes(), - media_type="image/jpeg", + media_type="image/png", headers={"Cache-Control": "no-store"}, ) -@app.post("/api/artwork/apply") -async def apply_artwork_export(request: Request): +@app.post("/api/upload-background") +async def upload_background(request: Request): + form = await request.form() + file = form.get("file") + if file is None: + raise HTTPException(status_code=400, detail="Missing file field.") + image_bytes = await file.read() + if len(image_bytes) > UPLOAD_MAX_BYTES: + raise HTTPException(status_code=400, detail="Uploaded file is too large (max 40 MB).") + width, height, fmt = validate_image_bytes(image_bytes) + upload_id = hashlib.sha256(image_bytes).hexdigest() + upload_path = IMPORT_CACHE_DIR / f"{upload_id}.img" + if not upload_path.exists(): + upload_path.write_bytes(image_bytes) + return {"upload_id": upload_id, "width": width, "height": height, "format": fmt} + + +@app.post("/api/collections/generate") +async def generate_collection_preview(request: Request): body = await request.json() - item_id = body.get("item_id") - if not item_id: - raise HTTPException(status_code=400, detail="Missing item id.") - target = get_editor_target(body.get("target_mode", "poster")) - cache_key = body.get("cache_key") or build_editor_cache_key(body) - cache_path = CACHE_DIR / f"{cache_key}.editor.jpg" - if cache_path.exists(): - image_bytes = cache_path.read_bytes() - else: - source_bytes = await resolve_editor_source_bytes(body) - image_bytes = render_artwork_editor_image( - source_bytes, - target_mode=body.get("target_mode", "poster"), - fit_mode=body.get("fit_mode", "cover"), - fill_mode=body.get("fill_mode", "blur"), - fill_color=body.get("fill_color", "#101318"), - zoom=float(body.get("zoom", 1.0)), - pan_x=float(body.get("pan_x", 0.0)), - pan_y=float(body.get("pan_y", 0.0)), - quality=int(body.get("quality", 95)), - ) - cache_path.write_bytes(image_bytes) - status = await emby_upload_image(item_id, image_bytes, target["emby_type"]) + options = parse_collection_render_request(body) + cache_key, preview_bytes = await render_collection_art_preview(options) + return StreamingResponse( + io.BytesIO(preview_bytes), + media_type="image/png", + headers={"X-Cache-Key": cache_key}, + ) + + +@app.post("/api/collections/apply") +async def apply_collection_art(request: Request): + body = await request.json() + options = parse_collection_render_request(body) + _, preview_bytes = await render_collection_art_preview(options) + status_code = await emby_upload_image(options["item_id"], preview_bytes, options["target_type"]) return { "status": "applied", - "emby_type": target["emby_type"], - "code": status, - "cache_key": cache_key, + "target_type": options["target_type"], + "code": status_code, } @app.post("/api/generate") async def generate(request: Request): - body = await request.json() - item_id = body["item_id"] - title = body["title"] - bg_mode = body.get("bg_mode", "backdrop") - custom_bg = body.get("custom_bg", None) - text_color = body.get("text_color", "#FFFFFF") - logo_align = body.get("logo_align", "bottom-center") - logo_scale = float(body.get("logo_scale", 1.3)) - darkness = float(body.get("darkness", 0.0)) - studio = body.get("studio", "none") - studio_position = body.get("studio_position", "bottom-right") - new_episodes_tag = bool(body.get("new_episodes_tag", False)) - season_finale_tag = bool(body.get("season_finale_tag", False)) - generate_primary = bool(body.get("generate_primary", False)) - logo_index_raw = body.get("logo_index", None) - logo_index = int(logo_index_raw) if logo_index_raw is not None else None - backdrop_index = int(body.get("backdrop_index", 0)) - primary_zoom = float(body.get("primary_zoom", 1.0)) - primary_pan_x = float(body.get("primary_pan_x", 0.0)) - primary_pan_y = float(body.get("primary_pan_y", -0.16)) - - poster_bytes = await emby_get_image(item_id, "Primary") - logo_bytes = await emby_get_image_optional(item_id, "Logo", logo_index) - backdrop_bytes = await emby_get_image_optional(item_id, "Backdrop", backdrop_index) - - thumb_bytes = generate_thumbnail( - poster_bytes, title, - bg_mode=bg_mode, - custom_bg=custom_bg, - text_color=text_color, - logo_align=logo_align, - logo_scale=logo_scale, - darkness=darkness, - studio=studio, - studio_position=studio_position, - new_episodes_tag=new_episodes_tag, - season_finale_tag=season_finale_tag, - logo_index=logo_index, - logo_bytes=logo_bytes, - backdrop_bytes=backdrop_bytes, - primary_zoom=primary_zoom, - primary_pan_x=primary_pan_x, - primary_pan_y=primary_pan_y, - ) - - cache_key = build_cache_key( - item_id=item_id, - bg_mode=bg_mode, - backdrop_index=backdrop_index, - custom_bg=custom_bg, - text_color=text_color, - logo_align=logo_align, - logo_scale=logo_scale, - darkness=darkness, - studio=studio, - studio_position=studio_position, - new_episodes_tag=new_episodes_tag, - season_finale_tag=season_finale_tag, - logo_index=logo_index, - primary_zoom=primary_zoom, - primary_pan_x=primary_pan_x, - primary_pan_y=primary_pan_y, - ) + options = await apply_auto_studio_defaults(parse_generator_request(await request.json())) + render_options = dict(options) + render_options["generate_primary"] = False + cache_key = get_generator_cache_key(options) thumb_cache_path = get_thumb_cache_path(cache_key) - thumb_cache_path.write_bytes(thumb_bytes) - primary_generated = False - if generate_primary: - try: - primary_bytes = generate_primary_cover( - poster_bytes, - title, - bg_mode=bg_mode, - custom_bg=custom_bg, - text_color=text_color, - logo_align=logo_align, - logo_scale=logo_scale, - darkness=darkness, - studio=studio, - studio_position=studio_position, - new_episodes_tag=new_episodes_tag, - season_finale_tag=season_finale_tag, - logo_index=logo_index, - logo_bytes=logo_bytes, - backdrop_bytes=backdrop_bytes, - primary_zoom=primary_zoom, - primary_pan_x=primary_pan_x, - primary_pan_y=primary_pan_y, - ) - get_primary_cache_path(cache_key).write_bytes(primary_bytes) - primary_generated = True - except Exception: - primary_generated = False + thumb_cached = thumb_cache_path.exists() + if thumb_cached: + return StreamingResponse( + io.BytesIO(thumb_cache_path.read_bytes()), + media_type="image/png", + headers={ + "X-Primary-Generated": "0", + "X-Cache-Key": cache_key, + }, + ) + + thumb_bytes, _ = await render_item_artwork(render_options) + thumb_cache_path.write_bytes(thumb_bytes) return StreamingResponse( io.BytesIO(thumb_bytes), media_type="image/png", headers={ - "X-Primary-Generated": "1" if primary_generated else "0", + "X-Primary-Generated": "0", "X-Cache-Key": cache_key, }, ) @@ -1787,135 +2777,32 @@ async def generate(request: Request): @app.post("/api/apply") async def apply_to_emby(request: Request): - body = await request.json() - item_id = body["item_id"] - title = body.get("title", "") - bg_mode = body.get("bg_mode", "backdrop") - backdrop_index = int(body.get("backdrop_index", 0)) - custom_bg = body.get("custom_bg", None) - text_color = body.get("text_color", "#FFFFFF") - logo_align = body.get("logo_align", "bottom-center") - logo_scale = float(body.get("logo_scale", 1.3)) - darkness = float(body.get("darkness", 0.0)) - studio = body.get("studio", "none") - studio_position = body.get("studio_position", "bottom-right") - new_episodes_tag = bool(body.get("new_episodes_tag", False)) - season_finale_tag = bool(body.get("season_finale_tag", False)) - generate_primary = bool(body.get("generate_primary", False)) - logo_index_raw = body.get("logo_index", None) - logo_index = int(logo_index_raw) if logo_index_raw is not None else None - primary_zoom = float(body.get("primary_zoom", 1.0)) - primary_pan_x = float(body.get("primary_pan_x", 0.0)) - primary_pan_y = float(body.get("primary_pan_y", -0.16)) - - cache_key = build_cache_key( - item_id=item_id, - bg_mode=bg_mode, - backdrop_index=backdrop_index, - custom_bg=custom_bg, - text_color=text_color, - logo_align=logo_align, - logo_scale=logo_scale, - darkness=darkness, - studio=studio, - studio_position=studio_position, - new_episodes_tag=new_episodes_tag, - season_finale_tag=season_finale_tag, - logo_index=logo_index, - primary_zoom=primary_zoom, - primary_pan_x=primary_pan_x, - primary_pan_y=primary_pan_y, - ) + options = await apply_auto_studio_defaults(parse_generator_request(await request.json())) + render_options = dict(options) + render_options["generate_primary"] = False + cache_key = get_generator_cache_key(options) thumb_cache_path = get_thumb_cache_path(cache_key) - primary_cache_path = get_primary_cache_path(cache_key) if not thumb_cache_path.exists(): - poster_bytes = await emby_get_image(item_id, "Primary") - logo_bytes = await emby_get_image_optional(item_id, "Logo", logo_index) - backdrop_bytes = await emby_get_image_optional(item_id, "Backdrop", backdrop_index) - thumb_bytes = generate_thumbnail( - poster_bytes, - title or "", - bg_mode=bg_mode, - custom_bg=custom_bg, - text_color=text_color, - logo_align=logo_align, - logo_scale=logo_scale, - darkness=darkness, - studio=studio, - studio_position=studio_position, - new_episodes_tag=new_episodes_tag, - season_finale_tag=season_finale_tag, - logo_index=logo_index, - logo_bytes=logo_bytes, - backdrop_bytes=backdrop_bytes, - primary_zoom=primary_zoom, - primary_pan_x=primary_pan_x, - primary_pan_y=primary_pan_y, - ) + thumb_bytes, _ = await render_item_artwork(render_options) thumb_cache_path.write_bytes(thumb_bytes) - if generate_primary and not primary_cache_path.exists(): - try: - primary_cache_path.write_bytes( - generate_primary_cover( - poster_bytes, - title or "", - bg_mode=bg_mode, - custom_bg=custom_bg, - text_color=text_color, - logo_align=logo_align, - logo_scale=logo_scale, - darkness=darkness, - studio=studio, - studio_position=studio_position, - new_episodes_tag=new_episodes_tag, - season_finale_tag=season_finale_tag, - logo_index=logo_index, - logo_bytes=logo_bytes, - backdrop_bytes=backdrop_bytes, - primary_zoom=primary_zoom, - primary_pan_x=primary_pan_x, - primary_pan_y=primary_pan_y, - ) - ) - except Exception: - pass thumb_bytes = thumb_cache_path.read_bytes() - thumb_status = await emby_upload_image(item_id, thumb_bytes, "Thumb") + thumb_status = await emby_upload_image(options["item_id"], thumb_bytes, "Thumb") primary_status = None primary_error = None - if generate_primary: + if options["generate_primary"]: try: - if primary_cache_path.exists(): - primary_bytes = primary_cache_path.read_bytes() + studio = (options.get("studio") or "none").strip().lower() + if studio == "none": + primary_error = "Primary poster stamping requires a studio logo selection." else: - poster_bytes = await emby_get_image(item_id, "Primary") - logo_bytes = await emby_get_image_optional(item_id, "Logo", logo_index) - backdrop_bytes = await emby_get_image_optional(item_id, "Backdrop", backdrop_index) - primary_bytes = generate_primary_cover( - poster_bytes, - title or "", - bg_mode=bg_mode, - custom_bg=custom_bg, - text_color=text_color, - logo_align=logo_align, - logo_scale=logo_scale, - darkness=darkness, - studio=studio, - studio_position=studio_position, - new_episodes_tag=new_episodes_tag, - season_finale_tag=season_finale_tag, - logo_index=logo_index, - logo_bytes=logo_bytes, - backdrop_bytes=backdrop_bytes, - primary_zoom=primary_zoom, - primary_pan_x=primary_pan_x, - primary_pan_y=primary_pan_y, + primary_status = await stamp_studio_logo_on_poster( + options["item_id"], + studio, + options.get("studio_position", "top-left"), ) - primary_cache_path.write_bytes(primary_bytes) - primary_status = await emby_upload_image(item_id, primary_bytes, "Primary") except Exception as exc: primary_error = str(exc) @@ -1923,11 +2810,85 @@ async def apply_to_emby(request: Request): "status": "applied", "thumb_code": thumb_status, "primary_code": primary_status, - "primary_attempted": generate_primary, + "primary_attempted": options["generate_primary"], "primary_error": primary_error, } +@app.post("/api/bulk-apply/category") +async def bulk_apply_category(request: Request): + body = await request.json() + category_id = (body.get("category_id") or "").strip() + category_name = (body.get("category_name") or "").strip() + if not category_id: + raise HTTPException(status_code=400, detail="Category is required.") + + base_options = await apply_auto_studio_defaults(parse_generator_request(body)) + items = await emby_get_all("/Items", { + "GenreIds": category_id, + "IncludeItemTypes": "Movie,Series", + "Recursive": "true", + "SortBy": "SortName", + "SortOrder": "Ascending", + "Fields": "ProductionYear", + "ImageTypeLimit": "1", + "EnableImageTypes": "Primary,Logo,Backdrop", + }) + eligible_items: list[dict] = [] + skipped_without_logo: list[str] = [] + skipped_without_primary: list[str] = [] + for item in items: + image_tags = item.get("ImageTags") or {} + name = item.get("Name") or "Unknown" + if "Primary" not in image_tags: + skipped_without_primary.append(name) + continue + if "Logo" not in image_tags: + skipped_without_logo.append(name) + continue + eligible_items.append(item) + + applied: list[str] = [] + failed: list[dict] = [] + for item in eligible_items: + item_options = dict(base_options) + item_options["item_id"] = item["Id"] + item_options["title"] = item.get("Name", "") + if item.get("Type") != "Series": + item_options["new_episodes_tag"] = False + item_options["season_finale_tag"] = False + item_options = await apply_auto_studio_defaults(item_options) + cache_key = get_generator_cache_key(item_options) + thumb_cache_path = get_thumb_cache_path(cache_key) + primary_cache_path = get_primary_cache_path(cache_key) + try: + thumb_bytes, primary_bytes = await render_item_artwork(item_options, fallback_logo_to_first=True) + thumb_cache_path.write_bytes(thumb_bytes) + await emby_upload_image(item["Id"], thumb_bytes, "Thumb") + if item_options["generate_primary"] and primary_bytes is not None: + primary_cache_path.write_bytes(primary_bytes) + await emby_upload_image(item["Id"], primary_bytes, "Primary") + applied.append(item.get("Name", "")) + except Exception as exc: + failed.append({ + "name": item.get("Name", ""), + "error": str(exc), + }) + return { + "status": "completed", + "category_id": category_id, + "category_name": category_name, + "matched_count": len(items), + "eligible_count": len(eligible_items), + "applied_count": len(applied), + "skipped_without_logo_count": len(skipped_without_logo), + "skipped_without_primary_count": len(skipped_without_primary), + "failed_count": len(failed), + "failed": failed[:12], + "applied": applied[:12], + } + + @app.get("/api/config") async def get_config(): return { diff --git a/atv.png b/atv.png new file mode 100644 index 0000000..4e52aeb Binary files /dev/null and b/atv.png differ diff --git a/docker-compose.yml b/docker-compose.yml index 1739a14..da34e5f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -7,6 +7,11 @@ services: environment: - EMBY_URL=http://your-emby-server:8096 - EMBY_API_KEY=your-api-key-here + # Optional external artwork providers: + # - TMDB_BEARER_TOKEN=your-tmdb-read-access-token + # - TMDB_API_KEY=your-tmdb-v3-api-key + # - GOOGLE_CUSTOM_SEARCH_API_KEY=your-google-api-key + # - GOOGLE_CUSTOM_SEARCH_ENGINE_ID=your-google-search-engine-id volumes: - ./output:/app/output - ./cache:/app/cache diff --git a/emby-cover.code-workspace b/emby-cover.code-workspace new file mode 100644 index 0000000..362d7c2 --- /dev/null +++ b/emby-cover.code-workspace @@ -0,0 +1,7 @@ +{ + "folders": [ + { + "path": "." + } + ] +} \ No newline at end of file diff --git a/output/appletv-test-2.png b/output/appletv-test-2.png new file mode 100644 index 0000000..dddd862 Binary files /dev/null and b/output/appletv-test-2.png differ diff --git a/output/appletv-test-local.png b/output/appletv-test-local.png new file mode 100644 index 0000000..c13b996 Binary files /dev/null and b/output/appletv-test-local.png differ diff --git a/output/appletv-test.png b/output/appletv-test.png new file mode 100644 index 0000000..dddd862 Binary files /dev/null and b/output/appletv-test.png differ diff --git a/paramount.png b/paramount.png new file mode 100644 index 0000000..1aeccd0 Binary files /dev/null and b/paramount.png differ diff --git a/static/studios/apple-tv.png b/static/studios/apple-tv.png new file mode 100644 index 0000000..4e52aeb Binary files /dev/null and b/static/studios/apple-tv.png differ diff --git a/static/studios/netflix.png b/static/studios/netflix.png new file mode 100644 index 0000000..3ce65a3 Binary files /dev/null and b/static/studios/netflix.png differ diff --git a/static/studios/paramount-plus.png b/static/studios/paramount-plus.png new file mode 100644 index 0000000..1aeccd0 Binary files /dev/null and b/static/studios/paramount-plus.png differ diff --git a/templates/airing.html b/templates/airing.html index d5ad7bc..ef416a6 100644 --- a/templates/airing.html +++ b/templates/airing.html @@ -3,7 +3,7 @@ -Current Airing Shows +EmbyToolkit + + + + + +
+
+
+

Bulk Assign

+

+ Bulk assign applies generated Thumb artwork only, + and is enabled when Emby already has a Logo and Backdrop. +

+
+
+ +
+ + +
+ +
+
+
0 series
+
Search All
+
+
+ + + + + + +
0 selected
+
+
+ +
+
Loading series…
+
+ +
+
No results
+
+ + +
+
+
+ +
+ + + + + diff --git a/templates/collections.html b/templates/collections.html new file mode 100644 index 0000000..405ed82 --- /dev/null +++ b/templates/collections.html @@ -0,0 +1,1799 @@ + + + + + +EmbyToolkit + + + + + + + + +
+ + +
+
+
+

Collection Artwork

+

Select an Emby collection, switch between `Thumb` and `Primary`, then use either the current artwork or a dropped image as the base. The editor lets you crop, darken, and overlay custom text before pushing the result back to Emby.

+
+
Waiting for collection
+
+ +
+
+
+

Collection

+ Editing Thumb +
+
+
+
Pick a collection
+
+ BoxSet + No selection yet +
+
+ +
+ + + +
+ +
+
+
Upload or drop a base image
+

Drop a JPG, PNG, or WEBP here to use it as the background for the current target. If you clear it, the editor falls back to the collection’s existing artwork.

+
+
Using current artwork when available.
+
+ + +
+ +
+
+
+ +
+
+
+
+ Preview + Select a collection to start editing. +
+
100% · X 0 · Y 0
+
+ +
+
+ +
Select a collection from the left to build custom artwork.
+
+ +
+ +
+ +
+
+ + +
+
Drag to position • Wheel to zoom
+
+
+
+
+ +
+
+
+ + +
+ +
+ +
+ + +
+
+ +
+ +
+ + + +
+
+ +
+ +
+ + + +
+
+ +
+ +
+ + +
+
+ +
+ +
+ + + + 100% +
+
+ +
+ +
+ + 18% +
+
+ +
+ +
+ + + + 100% +
+
+ +
+ +
+ + 0 +
+
+ +
+ +
+ + 0 +
+
+
+ +
+ + + +
+
+
+
+
+
+ +
+ + + + diff --git a/templates/index.html b/templates/index.html index d690511..872cd66 100644 --- a/templates/index.html +++ b/templates/index.html @@ -3,7 +3,7 @@ -Emby Thumbnail Generator +EmbyToolkit