[FEAT] create channel metadata and thumbnails. Closes #446

This commit is contained in:
arabcoders 2025-10-15 18:55:50 +03:00
parent 319f7d2f7a
commit e085e680cc
9 changed files with 474 additions and 26 deletions

View file

@ -87,6 +87,7 @@
"jmespath",
"jsonschema",
"kibibytes",
"Kodi",
"lastgroup",
"levelno",
"libcurl",

53
API.md
View file

@ -35,6 +35,7 @@ This document describes the available endpoints and their usage. All endpoints r
- [POST /api/tasks/inspect](#post-apitasksinspect)
- [POST /api/tasks/{id}/mark](#post-apitasksidmark)
- [DELETE /api/tasks/{id}/mark](#delete-apitasksidmark)
- [POST /api/tasks/{id}/metadata](#post-apitasksidmetadata)
- [GET /api/task\_definitions/](#get-apitask_definitions)
- [GET /api/task\_definitions/{identifier}](#get-apitask_definitionsidentifier)
- [POST /api/task\_definitions/](#post-apitask_definitions)
@ -658,6 +659,58 @@ or
---
### POST /api/tasks/{id}/metadata
**Purpose**: Generate metadata files for a scheduled task, including NFO files, thumbnails, and JSON metadata compatible with media centers.
**Path Parameter**:
- `id`: Task ID.
**Response**:
```json
{
"id": "channel_or_playlist_id",
"id_type": "youtube|twitch|...",
"title": "Channel/Playlist Title",
"description": "Description text...",
"uploader": "Uploader name",
"tags": ["tag1", "tag2"],
"year": 2024,
"json_file": "path/to/Title [id].info.json",
"nfo_file": "path/to/tvshow.nfo",
"thumbnails": {
"poster": "path/to/poster.jpg",
"fanart": "path/to/fanart.jpg",
"thumb": "path/to/thumb.jpg",
"banner": "path/to/banner.jpg",
"icon": "path/to/icon.jpg",
"landscape": "path/to/landscape.jpg"
}
}
```
or
```json
{ "error": "..." }
```
**Files Generated**:
- `tvshow.nfo` - NFO file for media center compatibility (Kodi, Jellyfin, Emby, etc.)
- `Title [id].info.json` - yt-dlp metadata file.
- `poster.jpg`, `fanart.jpg`, `thumb.jpg`, `banner.jpg`, `icon.jpg`, `landscape.jpg` - Thumbnail images (if available from the source)
**Notes**:
- This endpoint fetches metadata from the URL associated with the task
- Files are saved to the task's configured folder (or default download path)
- Existing files will be overwritten.
- Thumbnail images are only downloaded if available from the source.
- The NFO file follows the Kodi tvshow.nfo format
**Status Codes**:
- `200 OK` - Metadata generated successfully
- `400 Bad Request` - Missing task ID, invalid folder path, or failed to fetch metadata
- `404 Not Found` - Task does not exist
---
### GET /api/task_definitions/
**Purpose**: Retrieve all task definitions.

View file

@ -46,9 +46,27 @@ class Task:
return Encoder().encode(self.serialize())
def get(self, key: str, default: Any = None) -> Any:
"""
Get a value from the task by key.
Args:
key (str): The key to get.
default (Any): The default value if the key is not found.
Returns:
Any: The value of the key or the default value.
"""
return self.serialize().get(key, default)
def get_ytdlp_opts(self) -> YTDLPOpts:
"""
Get the yt-dlp options for the task.
Returns:
YTDLPOpts: The yt-dlp options.
"""
params: YTDLPOpts = YTDLPOpts.get_instance()
if self.preset:
@ -60,6 +78,13 @@ class Task:
return params
def mark(self) -> tuple[bool, str]:
"""
Mark the task's items as downloaded in the archive file.
Returns:
tuple[bool, str]: A tuple indicating success and a message.
"""
ret = self._mark_logic()
if isinstance(ret, tuple):
return ret
@ -73,6 +98,13 @@ class Task:
return (True, f"Task '{self.name}' items marked as downloaded.")
def unmark(self) -> tuple[bool, str]:
"""
Unmark the task's items from the archive file.
Returns:
tuple[bool, str]: A tuple indicating success and a message.
"""
ret: tuple[bool, str] | set[tuple[Path, set[str]]] = self._mark_logic()
if isinstance(ret, tuple):
return ret
@ -85,6 +117,33 @@ class Task:
return (True, f"Removed '{self.name}' items from archive file.")
def fetch_metadata(self, full: bool = False) -> tuple[dict[str, Any] | None, bool, str]:
"""
Fetch metadata for the task's URL.
Args:
full (bool): Whether to fetch full metadata including all entries for playlists.
Returns:
tuple[dict[str, Any]|None, bool, str]: A tuple containing the metadata (or None on failure), a boolean
indicating if the operation was successful, and a message.
"""
if not self.url:
return ({}, False, "No URL found in task parameters.")
params = self.get_ytdlp_opts()
if not full:
params.add_cli("-I0", from_user=False)
params = params.get_all()
ie_info: dict | None = extract_info(params, self.url, no_archive=True, follow_redirect=False, cache=True)
if not ie_info or not isinstance(ie_info, dict):
return ({}, False, "Failed to extract information from URL.")
return (ie_info, True, "")
def _mark_logic(self) -> tuple[bool, str] | set[tuple[Path, set[str]]]:
if not self.url:
return (False, "No URL found in task parameters.")

View file

@ -1572,3 +1572,60 @@ def archive_delete(file: str | Path, ids: list[str]) -> bool:
from app.library.Archiver import Archiver
return Archiver.get_instance().delete(file, ids)
def get_channel_images(thumbnails: list[dict]) -> dict:
"""
Extract channel images from a list of thumbnail dictionaries.
Args:
thumbnails (list[dict]): List of thumbnail dictionaries with keys 'url', 'width', 'height', and 'id'.
Returns:
dict: A dictionary with keys 'icon', 'landscape', 'poster', 'thumb', 'fanart', and 'banner' mapped to their respective URLs.
"""
artwork = {}
for t in thumbnails:
url = t.get("url")
if not url:
continue
width = t.get("width")
height = t.get("height")
tid = t.get("id", "")
if not width or not height:
if "avatar_uncropped" in tid:
artwork["icon"] = url
elif "banner_uncropped" in tid:
artwork["landscape"] = url
continue
ratio = width / height
if 0.55 <= ratio <= 0.75: # portrait → poster
artwork["poster"] = url
elif 0.9 <= ratio <= 1.1: # square → thumb
artwork.setdefault("thumb", url)
elif ratio >= 5: # very wide
if width >= 1920:
artwork["fanart"] = url
else:
artwork["banner"] = url
elif 1.6 <= ratio <= 1.8: # landscape
artwork["landscape"] = url
if "fanart" not in artwork and "banner" in artwork:
artwork["fanart"] = artwork["banner"]
if "banner" not in artwork and "fanart" in artwork:
artwork["banner"] = artwork["fanart"]
if "poster" not in artwork and "thumb" in artwork:
artwork["poster"] = artwork["thumb"] # optional fallback
return artwork

View file

@ -211,7 +211,7 @@ class NFOMakerPP(PostProcessor):
# collapse multiline descriptions
if "description" == resolved_key and isinstance(resolved_val, str):
resolved_val = self._clean_description(resolved_val)
resolved_val = NFOMakerPP._clean_description(resolved_val)
if resolved_val not in (None, ""):
data[nfo_name] = resolved_val
@ -321,15 +321,9 @@ class NFOMakerPP(PostProcessor):
repl (dict[str, Any]): Replacement dictionary.
"""
from xml.sax.saxutils import escape
# escape XML on a copy
safe_repl: dict[str, Any] = {}
for k, v in repl.items():
if isinstance(v, str):
safe_repl[k] = escape(v)
else:
safe_repl[k] = v
safe_repl[k] = NFOMakerPP._escape_text(v)
# replace placeholders
rendered = text
@ -357,7 +351,24 @@ class NFOMakerPP(PostProcessor):
except Exception as e:
self.to_screen(f"Error writing NFO file: {e}")
def _clean_description(self, text: str) -> str:
@staticmethod
def _escape_text(text: Any) -> Any:
"""
Escape text for XML.
Args:
text (str): Text to escape.
Returns:
Any: Escaped text if input is str, else original value.
"""
from xml.sax.saxutils import escape
return escape(text) if isinstance(text, str) else text
@staticmethod
def _clean_description(text: str) -> str:
"""
Strip links, chapters/timestamps, pure hashtags/mentions, and promo lines.
Return a compact single-line summary suitable for NFO <plot>.
@ -374,20 +385,20 @@ class NFOMakerPP(PostProcessor):
continue
# remove markdown links, keep labels
ln = self._MD_LINK.sub(r"\1", ln)
ln = NFOMakerPP._MD_LINK.sub(r"\1", ln)
# drop lines that are clearly noise
if self._TIME_LINE_PAT.match(ln):
if NFOMakerPP._TIME_LINE_PAT.match(ln):
continue
if self._HASHTAGS_LINE.match(ln):
if NFOMakerPP._HASHTAGS_LINE.match(ln):
continue
if self._MENTION_LINE.match(ln):
if NFOMakerPP._MENTION_LINE.match(ln):
continue
if self._PROMO_LINE_PAT.search(ln):
if NFOMakerPP._PROMO_LINE_PAT.search(ln):
continue
# strip raw/bare urls and domains
ln = self._URL_PAT.sub("", ln)
ln = NFOMakerPP._URL_PAT.sub("", ln)
# collapse leftover multiple spaces and stray separators
ln = re.sub(r"\s{2,}", " ", ln)
@ -402,7 +413,7 @@ class NFOMakerPP(PostProcessor):
# optional minimum signal: if too short, fall back to original first sentence without links
if 8 > len(summary.split()):
fallback = self._URL_PAT.sub("", text)
fallback = NFOMakerPP._URL_PAT.sub("", text)
fallback = re.sub(r"\s{2,}", " ", fallback).strip()
if 8 <= len(fallback.split()):
summary = fallback

View file

@ -1,14 +1,17 @@
import logging
import uuid
from typing import Any
from aiohttp import web
from aiohttp.web import Request, Response
from app.library.ag_utils import ag
from app.library.config import Config
from app.library.encoder import Encoder
from app.library.router import route
from app.library.Tasks import Task, TaskFailure, TaskResult, Tasks
from app.library.Utils import init_class, validate_url, validate_uuid
from app.library.Utils import get_channel_images, get_file, init_class, validate_url, validate_uuid
from app.postprocessors.nfo_maker import NFOMakerPP
LOG: logging.Logger = logging.getLogger(__name__)
@ -195,3 +198,144 @@ async def task_unmark(request: Request, encoder: Encoder) -> Response:
return web.json_response(data={"message": _message}, status=web.HTTPOk.status_code, dumps=encoder.encode)
except ValueError as e:
return web.json_response(data={"error": str(e)}, status=web.HTTPBadRequest.status_code)
@route("POST", "api/tasks/{id}/metadata", "tasks_metadata")
async def task_metadata(request: Request, config: Config, encoder: Encoder) -> Response:
"""
Generate metadata for the task.
Args:
request (Request): The request object.
config (Config): The config instance.
encoder (Encoder): The encoder instance.
Returns:
Response: The response object
"""
task_id: str = request.match_info.get("id", None)
if not task_id:
return web.json_response(data={"error": "No task id."}, status=web.HTTPBadRequest.status_code)
tasks: Tasks = Tasks.get_instance()
try:
task: Task | None = tasks.get(task_id)
if not task:
return web.json_response(
data={"error": f"Task '{task_id}' does not exist."}, status=web.HTTPNotFound.status_code
)
(save_path, _) = get_file(config.download_path, task.folder)
if not str(save_path or "").startswith(str(config.download_path)):
return web.json_response(data={"error": "Invalid task folder."}, status=web.HTTPBadRequest.status_code)
if not save_path.exists():
save_path.mkdir(parents=True, exist_ok=True)
metadata, status, message = task.fetch_metadata(full=False)
if not status:
return web.json_response(data={"error": message}, status=web.HTTPBadRequest.status_code)
info = {
"id": ag(metadata, ["id", "channel_id"]),
"id_type": metadata.get("extractor", "").split(":")[0].lower() if metadata.get("extractor") else None,
"title": ag(metadata, ["title", "fulltitle"]) or None,
"description": metadata.get("description", ""),
"uploader": metadata.get("uploader", ""),
"tags": metadata.get("tags", []),
"year": metadata.get("release_year"),
"thumbnails": get_channel_images(metadata.get("thumbnails", {})),
}
if not info.get("title"):
return web.json_response(
data={"error": "Failed to get title from metadata."}, status=web.HTTPBadRequest.status_code
)
from yt_dlp.utils import sanitize_filename
title = sanitize_filename(info.get("title"))
filename = save_path / f"{title} [{info.get('id')}].info.json"
filename.write_text(encoder.encode(metadata), encoding="utf-8")
info["json_file"] = str(filename.relative_to(config.download_path))
filename = save_path / "tvshow.nfo"
info["nfo_file"] = f"{save_path}/{filename}"
xml_content = "<tvshow>\n"
xml_content += f" <title>{NFOMakerPP._escape_text(info.get('title'))}</title>\n"
if info.get("description"):
xml_content += (
f" <plot>{NFOMakerPP._escape_text(NFOMakerPP._clean_description(info.get('description')))}</plot>\n"
)
if info.get("id"):
xml_content += f" <id>{NFOMakerPP._escape_text(info.get('id'))}</id>\n"
if info.get("id_type") and info.get("id"):
xml_content += f' <uniqueid type="{NFOMakerPP._escape_text(info.get("id_type"))}" default="true">{NFOMakerPP._escape_text(info.get("id"))}</uniqueid>\n'
if info.get("uploader"):
xml_content += f" <studio>{NFOMakerPP._escape_text(info.get('uploader'))}</studio>\n"
if info.get("tags", []):
for tag in info.get("tags", []):
xml_content += f" <tags>{NFOMakerPP._escape_text(tag)}</tags>\n"
if info.get("year"):
xml_content += f" <year>{info.get('year')}</year>\n"
xml_content += " <status>Continuing</status>\n"
xml_content += "</tvshow>\n"
filename.write_text(xml_content, encoding="utf-8")
try:
import httpx
from yt_dlp.utils.networking import random_user_agent
ytdlp_args: dict = task.get_ytdlp_opts().get_all()
opts: dict[str, Any] = {
"headers": {
"User-Agent": request.headers.get("User-Agent", ytdlp_args.get("user_agent", random_user_agent())),
},
}
if proxy := ytdlp_args.get("proxy"):
opts["proxy"] = proxy
try:
from httpx_curl_cffi import AsyncCurlTransport, CurlOpt
opts["transport"] = AsyncCurlTransport(
impersonate="chrome",
default_headers=True,
curl_options={CurlOpt.FRESH_CONNECT: True},
)
opts.pop("headers", None)
except Exception:
pass
async with httpx.AsyncClient(**opts) as client:
for key in info.get("thumbnails", {}):
try:
url = info["thumbnails"][key]
LOG.info(f"Fetching thumbnail '{key}' from '{url}'")
if not url:
continue
try:
validate_url(url, allow_internal=config.allow_internal_urls)
except ValueError:
LOG.warning(f"Invalid thumbnail url '{url}'")
continue
resp = await client.request(method="GET", url=url, follow_redirects=True)
filename = save_path / f"{key}.jpg"
filename.write_bytes(resp.content)
info["thumbnails"][key] = str(filename.relative_to(config.download_path))
except Exception as e:
LOG.warning(f"Failed to fetch thumbnail '{key}' from '{url}'. '{e!s}'")
continue
except Exception as e:
LOG.warning(f"Failed to fetch thumbnails. '{e!s}'")
return web.json_response(data=info, status=web.HTTPOk.status_code, dumps=encoder.encode)
except ValueError as e:
return web.json_response(data={"error": str(e)}, status=web.HTTPBadRequest.status_code)

View file

@ -200,6 +200,13 @@
<span>Run now</span>
</NuxtLink>
<NuxtLink class="dropdown-item" @click="generateMeta(item)">
<span class="icon"><i class="fa-solid fa-photo-film" /></span>
<span>Generate metadata</span>
</NuxtLink>
<hr class="dropdown-divider" />
<NuxtLink class="dropdown-item" @click="() => inspectTask = item">
<span class="icon"><i class="fa-solid fa-magnifying-glass" /></span>
<span>Inspect Handler</span>
@ -243,6 +250,9 @@
<NuxtLink target="_blank" :href="item.url">
{{ remove_tags(item.name) }}
</NuxtLink>
<span class="icon" v-if="item.in_progress">
<i class="fa-solid fa-spinner fa-spin has-text-info" />
</span>
</div>
<div class="card-header-icon">
<div class="field is-grouped">
@ -329,6 +339,13 @@
<span>Run now</span>
</NuxtLink>
<NuxtLink class="dropdown-item" @click="generateMeta(item)">
<span class="icon"><i class="fa-solid fa-photo-film" /></span>
<span>Generate metadata</span>
</NuxtLink>
<hr class="dropdown-divider" />
<NuxtLink class="dropdown-item" @click="() => inspectTask = item">
<span class="icon"><i class="fa-solid fa-magnifying-glass" /></span>
<span>Inspect Handler</span>
@ -857,4 +874,55 @@ const unarchiveAll = async (item: task_item) => {
item.in_progress = false
}
}
const generateMeta = async (item: task_item) => {
try {
let path = '/';
if (item.folder) {
path = `/${sTrim(item.folder, '/')}`
}
const { status } = await cDialog({
rawHTML: `
<p>
Generate '${item.name}' metadata in '<b class="has-text-danger">${path}</b>'? you will be notified when it is done.
</p>
<p>
<b>This action will generate:</b>
<ul>
<li><strong>tvshow.nfo</strong> - for media center compatibility</li>
<li><strong>title [id].info.json</strong> - yt-dlp metadata file</li>
<li>
<strong>Thumbnails</strong>: poster.jpg, fanart.jpg, thumb.jpg, banner.jpg, icon.jpg, landscape.jpg
<u>if they are available</u>.
</li>
</ul>
</p>
<p class="has-text-danger">
<span class="icon"><i class="fa-solid fa-triangle-exclamation"></i></span>
<span>Warning</span>: This will overwrite existing metadata files if they exist.
</p>`
})
if (true !== status) {
return;
}
item.in_progress = true
const response = await request(`/api/tasks/${item.id}/metadata`, { method: 'POST' })
const data = await response.json()
if (data?.error) {
toast.error(data.error)
return
}
toast.success('Metadata generation completed.')
} catch (e: any) {
toast.error(`Failed to generate metadata. ${e.message || 'Unknown error.'}`)
return
} finally {
item.in_progress = false
}
}
</script>

View file

@ -366,20 +366,16 @@ const iTrim = (str: string, delim: string, position: 'start' | 'end' | 'both' =
throw new Error('Delimiter is required')
}
if (']' === delim) {
delim = '\\]'
}
if ('\\' === delim) {
delim = '\\\\'
}
// Escape special regex characters for use in character class
// Characters that need escaping in character classes: \ ] ^ -
const escapedDelim = delim.replace(/[\\^\-\]]/g, '\\$&')
if (['both', 'start'].includes(position)) {
str = str.replace(new RegExp(`^[${delim}]+`, 'g'), '')
str = str.replace(new RegExp(`^[${escapedDelim}]+`, 'g'), '')
}
if (['both', 'end'].includes(position)) {
str = str.replace(new RegExp(`[${delim}]+$`, 'g'), '')
str = str.replace(new RegExp(`[${escapedDelim}]+$`, 'g'), '')
}
return str

View file

@ -265,6 +265,65 @@ describe('string manipulation helpers', () => {
expect(utils.iTrim('value::', ':', 'end')).toBe('value')
})
it('iTrim handles forward slash delimiter', () => {
expect(utils.iTrim('//value//', '/', 'both')).toBe('value')
expect(utils.iTrim('/value', '/', 'start')).toBe('value')
expect(utils.iTrim('value/', '/', 'end')).toBe('value')
expect(utils.iTrim('///multiple///', '/', 'both')).toBe('multiple')
})
it('iTrim handles backslash delimiter', () => {
expect(utils.iTrim('\\\\value\\\\', '\\', 'both')).toBe('value')
expect(utils.iTrim('\\value', '\\', 'start')).toBe('value')
expect(utils.iTrim('value\\', '\\', 'end')).toBe('value')
})
it('iTrim handles hyphen delimiter', () => {
expect(utils.iTrim('--value--', '-', 'both')).toBe('value')
expect(utils.iTrim('-value', '-', 'start')).toBe('value')
expect(utils.iTrim('value-', '-', 'end')).toBe('value')
expect(utils.iTrim('---multiple---', '-', 'both')).toBe('multiple')
})
it('iTrim handles caret delimiter', () => {
expect(utils.iTrim('^^value^^', '^', 'both')).toBe('value')
expect(utils.iTrim('^value', '^', 'start')).toBe('value')
expect(utils.iTrim('value^', '^', 'end')).toBe('value')
})
it('iTrim handles bracket delimiters', () => {
expect(utils.iTrim('[[value]]', '[', 'both')).toBe('value]]')
expect(utils.iTrim(']]value[[', ']', 'both')).toBe('value[[')
})
it('iTrim handles dot delimiter', () => {
expect(utils.iTrim('..value..', '.', 'both')).toBe('value')
expect(utils.iTrim('.value', '.', 'start')).toBe('value')
expect(utils.iTrim('value.', '.', 'end')).toBe('value')
})
it('iTrim handles special regex characters', () => {
expect(utils.iTrim('**value**', '*', 'both')).toBe('value')
expect(utils.iTrim('++value++', '+', 'both')).toBe('value')
expect(utils.iTrim('??value??', '?', 'both')).toBe('value')
expect(utils.iTrim('||value||', '|', 'both')).toBe('value')
expect(utils.iTrim('((value))', '(', 'both')).toBe('value))')
expect(utils.iTrim('((value))', ')', 'both')).toBe('((value')
})
it('iTrim handles empty string', () => {
expect(utils.iTrim('', '/', 'both')).toBe('')
})
it('iTrim throws error when delimiter is empty', () => {
expect(() => utils.iTrim('value', '', 'both')).toThrow('Delimiter is required')
})
it('iTrim preserves middle occurrences', () => {
expect(utils.iTrim('/path/to/file/', '/', 'both')).toBe('path/to/file')
expect(utils.iTrim('//path//to//file//', '/', 'both')).toBe('path//to//file')
})
it('eTrim and sTrim delegate to iTrim ends', () => {
expect(utils.eTrim('##name##', '#')).toBe('##name')
expect(utils.sTrim('##name##', '#')).toBe('name##')