admin/embycovers
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

updates

Browse Source
This commit is contained in:
ponzischeme89
2026-04-15 21:25:03 +12:00
parent 914e7454d0
commit 1805551dae
4 changed files with 1176 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+425 -65
View File
@@ -4,7 +4,9 @@ import os
import hashlib
import base64
import time
import urllib.parse
from datetime import datetime, timedelta, timezone
from dataclasses import dataclass
from pathlib import Path
import httpx
@@ -22,14 +24,22 @@ EMBY_URL = os.environ.get("EMBY_URL", "http://10.0.0.2:8096")
EMBY_API_KEY = os.environ.get("EMBY_API_KEY", "b9af54b630f6448289ab96422add567a")
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"
THUMB_WIDTH = 800
THUMB_HEIGHT = 450
PRIMARY_WIDTH = 1000
PRIMARY_HEIGHT = 1500
PRIMARY_MIN_ZOOM = 0.6
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"},
}
HTTP_TIMEOUT = 30.0
http_client: httpx.AsyncClient | None = None
AIRING_LOOKUP_CACHE_TTL = 900
@@ -77,6 +87,169 @@ def load_image_from_bytes(image_bytes: bytes, mode: str = "RGB") -> Image.Image:
return img.convert(mode).copy()
def encode_jpeg_bytes(img: Image.Image, *, quality: int = 95) -> bytes:
buf = io.BytesIO()
img.convert("RGB").save(buf, format="JPEG", quality=quality, optimize=True, progressive=True)
return buf.getvalue()
def validate_image_bytes(image_bytes: bytes, *, max_pixels: int = 24_000_000) -> tuple[int, int, str]:
try:
with Image.open(io.BytesIO(image_bytes)) as img:
img.verify()
width, height = img.size
fmt = img.format or "image"
except (UnidentifiedImageError, OSError, ValueError) as exc:
raise HTTPException(status_code=400, detail="The downloaded file is not a valid image.") from exc
if width * height > max_pixels:
raise HTTPException(status_code=400, detail="Image is too large to process safely.")
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"
if not path.exists():
raise HTTPException(status_code=404, detail="Imported image was 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 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 fit_image_positioned(
img: Image.Image,
width: int,
height: int,
*,
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)
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)
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,
)
canvas.paste(fitted, (x, y))
return encode_jpeg_bytes(canvas, quality=max(80, min(98, int(quality))))
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()
# --- Emby API helpers ---
def get_http_client() -> httpx.AsyncClient:
@@ -204,6 +377,75 @@ async def emby_upload_image(item_id: str, image_bytes: bytes, image_type: str =
return r.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
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:
if not value:
return None
@@ -527,65 +769,6 @@ def cover_crop_positioned(
pan_y = max(-1.0, min(1.0, float(pan_y)))
zoom = max(PRIMARY_MIN_ZOOM, min(PRIMARY_MAX_ZOOM, float(zoom)))
if zoom < 1.0:
contain_scale = min(width / max(1, img.width), height / max(1, img.height))
cover_scale = max(width / max(1, img.width), height / max(1, img.height))
zoom_t = (zoom - PRIMARY_MIN_ZOOM) / max(0.0001, 1.0 - PRIMARY_MIN_ZOOM)
zoom_t = max(0.0, min(1.0, zoom_t))
scale = contain_scale + (cover_scale - contain_scale) * zoom_t
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)
if fitted_w >= width:
paste_x = -int(round((fitted_w - width) * ((pan_x + 1.0) / 2.0)))
else:
paste_x = int(round((width - fitted_w) * ((pan_x + 1.0) / 2.0)))
if fitted_h >= height:
paste_y = -int(round((fitted_h - height) * ((pan_y + 1.0) / 2.0)))
else:
paste_y = int(round((height - fitted_h) * ((pan_y + 1.0) / 2.0)))
canvas = Image.new("RGB", (width, height))
left_gap = max(0, paste_x)
top_gap = max(0, paste_y)
right_gap = max(0, width - (paste_x + fitted_w))
bottom_gap = max(0, height - (paste_y + fitted_h))
strip = max(1, min(12, fitted_w, fitted_h))
if top_gap:
top_strip = fitted.crop((0, 0, fitted_w, strip)).resize((fitted_w, top_gap), Image.LANCZOS)
canvas.paste(top_strip, (paste_x, 0))
if bottom_gap:
bottom_strip = fitted.crop((0, fitted_h - strip, fitted_w, fitted_h)).resize((fitted_w, bottom_gap), Image.LANCZOS)
canvas.paste(bottom_strip, (paste_x, paste_y + fitted_h))
if left_gap:
left_strip = fitted.crop((0, 0, strip, fitted_h)).resize((left_gap, fitted_h), Image.LANCZOS)
canvas.paste(left_strip, (0, paste_y))
if right_gap:
right_strip = fitted.crop((fitted_w - strip, 0, fitted_w, fitted_h)).resize((right_gap, fitted_h), Image.LANCZOS)
canvas.paste(right_strip, (paste_x + fitted_w, paste_y))
if top_gap and left_gap:
canvas.paste(fitted.crop((0, 0, strip, strip)).resize((left_gap, top_gap), Image.LANCZOS), (0, 0))
if top_gap and right_gap:
canvas.paste(
fitted.crop((fitted_w - strip, 0, fitted_w, strip)).resize((right_gap, top_gap), Image.LANCZOS),
(paste_x + fitted_w, 0),
)
if bottom_gap and left_gap:
canvas.paste(
fitted.crop((0, fitted_h - strip, strip, fitted_h)).resize((left_gap, bottom_gap), Image.LANCZOS),
(0, paste_y + fitted_h),
)
if bottom_gap and right_gap:
canvas.paste(
fitted.crop((fitted_w - strip, fitted_h - strip, fitted_w, fitted_h)).resize((right_gap, bottom_gap), Image.LANCZOS),
(paste_x + fitted_w, paste_y + fitted_h),
)
canvas.paste(fitted, (paste_x, paste_y))
return canvas
target_ratio = width / max(1, height)
source_ratio = img.width / max(1, img.height)
@@ -814,11 +997,22 @@ def generate_thumbnail(
bg_color = (15, 15, 25)
bg = Image.new("RGB", (width, height), bg_color)
else:
# Blurred poster fallback
bg = load_image_from_bytes(poster_bytes, mode="RGB")
bg = cover_crop(bg, 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 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(poster, width, height)
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))
@@ -1319,6 +1513,172 @@ 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):
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"
if not cache_path.exists():
raise HTTPException(status_code=404, detail="Export was not found in cache.")
return Response(
content=cache_path.read_bytes(),
media_type="image/jpeg",
headers={"Cache-Control": "no-store"},
)
@app.post("/api/artwork/apply")
async def apply_artwork_export(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"])
return {
"status": "applied",
"emby_type": target["emby_type"],
"code": status,
"cache_key": cache_key,
}
@app.post("/api/generate")
async def generate(request: Request):
body = await request.json()