alex/manage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor frontend and backend modules

Browse Source
This commit is contained in:
Alex
2026-05-04 22:57:57 +02:00
parent e0e461502b
commit 8f0a9650b0
17 changed files with 4560 additions and 4416 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/media_library_viewer_api/routers/users.py
+1 -387
View File
@@ -1,387 +1 @@
"""Users router — Jellyfin list plus optional Jellyseerr enrichment."""
from __future__ import annotations
import json
import logging
from typing import Any
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
from media_library_viewer_api.clients.jellyfin import JellyfinClient
from media_library_viewer_api.clients.jellyseerr import JellyseerrClient
from media_library_viewer_api.config import get_settings
from media_library_viewer_api.dependencies import (
get_jellyfin_client,
get_jellyseerr_client,
get_mail_queue,
)
from media_library_viewer_api.services.mailer import EmailAttachment, validate_smtp_settings
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/users", tags=["users"])
_PERMISSION_FLAGS = [
(2, "admin"),
(4, "manage_settings"),
(8, "manage_users"),
(16, "manage_requests"),
(32, "request"),
(64, "vote"),
(128, "auto_approve"),
(256, "auto_approve_movie"),
(512, "auto_approve_tv"),
(1024, "request_4k"),
(2048, "request_4k_movie"),
(4096, "request_4k_tv"),
(8192, "request_advanced"),
(16384, "request_view"),
(32768, "auto_approve_4k"),
(65536, "auto_approve_4k_movie"),
(131072, "auto_approve_4k_tv"),
(262144, "request_movie"),
(524288, "request_tv"),
(1048576, "manage_issues"),
(2097152, "view_issues"),
]
_USER_TYPES = {
1: "plex",
2: "local",
3: "jellyfin",
4: "emby",
}
def _safe_int(value: Any) -> int:
try:
return int(value)
except (TypeError, ValueError):
return 0
def _permission_labels(permissions: int) -> list[str]:
labels = [label for bit, label in _PERMISSION_FLAGS if permissions & bit]
return labels or ["none"]
def _role_label(permissions: int) -> str:
if permissions & 2:
return "admin"
if permissions & (4 | 8 | 16):
return "manager"
if permissions & (32 | 64 | 128):
return "requester"
return "user"
def _account_type(user_type: Any) -> str:
return _USER_TYPES.get(_safe_int(user_type), "unknown")
def _merge_users(
jellyfin_users: list[dict[str, Any]],
jellyseerr_jellyfin_users: list[dict[str, Any]] | None,
jellyseerr_users: list[dict[str, Any]] | None,
jellyseerr_client: JellyseerrClient | None,
) -> dict[str, Any]:
def _normalize(value: Any) -> str:
return str(value or "").strip().lower()
def _looks_like_email(value: Any) -> bool:
text = str(value or "").strip()
return bool(text and "@" in text and " " not in text)
def _pick_source_and_value(candidates: list[tuple[str, Any]]) -> tuple[str, str]:
for source, value in candidates:
if _looks_like_email(value):
return source, str(value).strip()
return "", ""
def _first_value(candidates: list[tuple[str, Any]]) -> tuple[str, str]:
for source, value in candidates:
text = str(value or "").strip()
if text:
return source, text
return "", ""
def _source_summary(name_source: str, email_source: str, avatar_source: str, access_source: str) -> str:
return ", ".join(
[
f"name={name_source or 'none'}",
f"email={email_source or 'none'}",
f"avatar={avatar_source or 'none'}",
f"access={access_source or 'none'}",
]
)
def _lookup_keys(item: dict[str, Any]) -> list[str]:
return [
_normalize(item.get("id")),
_normalize(item.get("Id")),
_normalize(item.get("userId")),
_normalize(item.get("user_id")),
_normalize(item.get("jellyfinUserId")),
_normalize(item.get("jellyfin_user_id")),
_normalize(item.get("jellyfinUsername")),
_normalize(item.get("jellyfin_username")),
_normalize(item.get("username")),
_normalize(item.get("displayName")),
_normalize(item.get("display_name")),
]
linked_by_jellyfin_id: dict[str, dict[str, Any]] = {}
for item in jellyseerr_jellyfin_users or []:
for key in (
item.get("id"),
item.get("Id"),
item.get("userId"),
item.get("user_id"),
item.get("jellyfinUserId"),
item.get("jellyfin_user_id"),
):
normalized = _normalize(key)
if normalized:
linked_by_jellyfin_id[normalized] = item
seerr_by_key: dict[str, dict[str, Any]] = {}
for item in jellyseerr_users or []:
for key in _lookup_keys(item):
if key:
seerr_by_key[key] = item
items: list[dict[str, Any]] = []
enriched_count = 0
for user in jellyfin_users:
jellyfin_id = str(user.get("Id") or user.get("id") or "")
jellyfin_name = str(user.get("Name") or user.get("name") or "")
jf_link = linked_by_jellyfin_id.get(_normalize(jellyfin_id))
seerr_user = None
for candidate in [
jellyfin_name,
(jf_link or {}).get("jellyfinUsername"),
(jf_link or {}).get("jellyfin_username"),
(jf_link or {}).get("username"),
(jf_link or {}).get("displayName"),
(jf_link or {}).get("display_name"),
]:
seerr_user = seerr_by_key.get(_normalize(candidate))
if seerr_user:
break
email_source, email = _pick_source_and_value(
[
("jellyseerr:user", (seerr_user or {}).get("email")),
("jellyseerr:jellyfin", (jf_link or {}).get("email")),
]
)
avatar_source, avatar = _first_value(
[
("jellyseerr:user", (seerr_user or {}).get("avatar")),
("jellyseerr:jellyfin", (jf_link or {}).get("thumb")),
("jellyseerr:jellyfin", (jf_link or {}).get("avatar")),
]
)
if avatar and jellyseerr_client:
avatar = jellyseerr_client.absolute_url(avatar)
permissions = _safe_int((seerr_user or {}).get("permissions"))
user_type = _safe_int((seerr_user or {}).get("userType") or (seerr_user or {}).get("user_type"))
role = _role_label(permissions)
access_source = "jellyseerr:user" if seerr_user else ""
name_source = "jellyfin"
summary = _source_summary(name_source, email_source, avatar_source, access_source)
if seerr_user or jf_link:
enriched_count += 1
items.append(
{
"jellyfin_id": jellyfin_id,
"username": jellyfin_name,
"display_name": jellyfin_name,
"email": email,
"email_source": email_source,
"avatar": avatar,
"avatar_source": avatar_source,
"contactable": bool(email),
"source": summary,
"source_summary": summary,
"name_source": name_source,
"access_source": access_source,
"jellyseerr_user_id": _safe_int((seerr_user or {}).get("id") or (seerr_user or {}).get("userId")) or None,
"jellyseerr_username": str(
(seerr_user or {}).get("username") or (seerr_user or {}).get("jellyfinUsername") or ""
),
"user_type": user_type or None,
"user_type_label": _account_type(user_type),
"role": role,
"permissions": permissions,
"permissions_label": ", ".join(_permission_labels(permissions)),
"request_count": _safe_int((seerr_user or {}).get("requestCount")) or None,
}
)
logger.info(
"Users merged jellyfin=%s jellyseerr_jellyfin=%s jellyseerr_users=%s enriched=%s",
len(jellyfin_users),
len(jellyseerr_jellyfin_users or []),
len(jellyseerr_users or []),
enriched_count,
)
return {
"items": items,
"total": len(items),
"jellyseerr_configured": jellyseerr_client is not None,
"jellyseerr_available": bool(jellyseerr_users or jellyseerr_jellyfin_users),
"jellyseerr_jellyfin_user_count": len(jellyseerr_jellyfin_users or []),
"jellyseerr_user_count": len(jellyseerr_users or []),
"enriched_count": enriched_count,
}
@router.get("")
def get_users(
jellyfin: JellyfinClient = Depends(get_jellyfin_client),
jellyseerr: JellyseerrClient | None = Depends(get_jellyseerr_client),
) -> dict[str, Any]:
"""Return the known users, enriched with Jellyseerr data when available."""
jellyfin_users = jellyfin.users()
logger.info("Users endpoint fetched %s Jellyfin users", len(jellyfin_users))
jellyseerr_jellyfin_users: list[dict[str, Any]] | None = None
jellyseerr_users: list[dict[str, Any]] | None = None
jellyseerr_error = ""
if jellyseerr:
try:
jellyseerr_jellyfin_users = jellyseerr.jellyfin_users()
except Exception as exc: # pragma: no cover - network fallback
logger.exception("Jellyseerr Jellyfin-linked user fetch failed")
jellyseerr_error = f"Jellyseerr Jellyfin users fetch failed: {exc}"
try:
jellyseerr_users = jellyseerr.users()
except Exception as exc: # pragma: no cover - network fallback
logger.exception("Jellyseerr user list fetch failed")
jellyseerr_error = (
f"{jellyseerr_error}; " if jellyseerr_error else ""
) + f"Jellyseerr user list fetch failed: {exc}"
result = _merge_users(jellyfin_users, jellyseerr_jellyfin_users, jellyseerr_users, jellyseerr)
result["jellyseerr_error"] = jellyseerr_error
logger.info(
"Users response total=%s configured=%s available=%s enriched=%s error=%s",
result["total"],
result["jellyseerr_configured"],
result["jellyseerr_available"],
result["enriched_count"],
bool(jellyseerr_error),
)
return result
@router.get("/message/status")
def get_user_message_status(mail_queue=Depends(get_mail_queue)) -> dict[str, Any]:
"""Return the current background email queue status."""
return mail_queue.status()
@router.post("/message", status_code=status.HTTP_202_ACCEPTED)
async def post_user_message(
recipient_ids: str = Form(...),
subject: str = Form(...),
html_body: str = Form(""),
text_body: str = Form(""),
attachments: list[UploadFile] | None = File(default=None),
jellyfin: JellyfinClient = Depends(get_jellyfin_client),
jellyseerr: JellyseerrClient | None = Depends(get_jellyseerr_client),
mail_queue=Depends(get_mail_queue),
) -> dict[str, Any]:
"""Queue a single email to the selected users without blocking the API."""
try:
requested_ids = json.loads(recipient_ids)
except json.JSONDecodeError as exc:
raise HTTPException(status_code=400, detail=f"recipient_ids must be valid JSON: {exc}") from exc
if not isinstance(requested_ids, list):
raise HTTPException(status_code=400, detail="recipient_ids must be a JSON list")
cleaned_ids = [str(item).strip() for item in requested_ids if str(item).strip()]
if not cleaned_ids:
raise HTTPException(status_code=400, detail="At least one recipient is required")
subject = subject.strip()
if not subject:
raise HTTPException(status_code=400, detail="Subject is required")
directory = get_users(jellyfin=jellyfin, jellyseerr=jellyseerr)
users_by_id = {str(item.get("jellyfin_id") or ""): item for item in directory.get("items", [])}
recipients: list[str] = []
recipient_labels: list[str] = []
skipped: list[dict[str, str]] = []
for user_id in cleaned_ids:
item = users_by_id.get(user_id)
if not item:
skipped.append({"jellyfin_id": user_id, "reason": "not found"})
continue
email = str(item.get("email") or "").strip()
if not email:
skipped.append({"jellyfin_id": user_id, "reason": "missing email"})
continue
recipients.append(email)
recipient_labels.append(f"{item.get('display_name') or item.get('username') or user_id} <{email}>")
if not recipients:
raise HTTPException(status_code=400, detail="No selected users have a deliverable email address")
settings = get_settings()
validate_smtp_settings(settings)
queue_status = mail_queue.status()
if not queue_status["worker_running"]:
raise HTTPException(status_code=503, detail="Email queue worker is not running")
attachment_payloads: list[EmailAttachment] = []
for upload in attachments or []:
data = await upload.read()
if not data:
continue
attachment_payloads.append(
EmailAttachment(
filename=upload.filename or "attachment",
content_type=upload.content_type or "application/octet-stream",
data=data,
)
)
request_id = mail_queue.enqueue(
settings=settings,
recipients=recipients,
subject=subject,
html_body=html_body,
text_body=text_body,
attachments=attachment_payloads,
)
from_address = str(getattr(settings, "smtp_from_address", "") or "").strip() or str(
getattr(settings, "smtp_username", "") or ""
).strip()
logger.info(
"Users message queued request_id=%s subject=%s recipients=%s attachments=%s skipped=%s",
request_id,
subject,
len(recipients),
len(attachment_payloads),
len(skipped),
)
return {
"status": "queued",
"request_id": request_id,
"from_address": from_address,
"recipient_count": len(recipients),
"attachment_count": len(attachment_payloads),
"subject": subject,
"recipient_labels": recipient_labels,
"skipped": skipped,
}
from .users_impl import * # noqa: F401,F403
backend/src/media_library_viewer_api/routers/users_impl.py
+387
View File
@@ -0,0 +1,387 @@
"""Users router — Jellyfin list plus optional Jellyseerr enrichment."""
from __future__ import annotations
import json
import logging
from typing import Any
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
from media_library_viewer_api.clients.jellyfin import JellyfinClient
from media_library_viewer_api.clients.jellyseerr import JellyseerrClient
from media_library_viewer_api.config import get_settings
from media_library_viewer_api.dependencies import (
get_jellyfin_client,
get_jellyseerr_client,
get_mail_queue,
)
from media_library_viewer_api.services.mailer import EmailAttachment, validate_smtp_settings
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/users", tags=["users"])
_PERMISSION_FLAGS = [
(2, "admin"),
(4, "manage_settings"),
(8, "manage_users"),
(16, "manage_requests"),
(32, "request"),
(64, "vote"),
(128, "auto_approve"),
(256, "auto_approve_movie"),
(512, "auto_approve_tv"),
(1024, "request_4k"),
(2048, "request_4k_movie"),
(4096, "request_4k_tv"),
(8192, "request_advanced"),
(16384, "request_view"),
(32768, "auto_approve_4k"),
(65536, "auto_approve_4k_movie"),
(131072, "auto_approve_4k_tv"),
(262144, "request_movie"),
(524288, "request_tv"),
(1048576, "manage_issues"),
(2097152, "view_issues"),
]
_USER_TYPES = {
1: "plex",
2: "local",
3: "jellyfin",
4: "emby",
}
def _safe_int(value: Any) -> int:
try:
return int(value)
except (TypeError, ValueError):
return 0
def _permission_labels(permissions: int) -> list[str]:
labels = [label for bit, label in _PERMISSION_FLAGS if permissions & bit]
return labels or ["none"]
def _role_label(permissions: int) -> str:
if permissions & 2:
return "admin"
if permissions & (4 | 8 | 16):
return "manager"
if permissions & (32 | 64 | 128):
return "requester"
return "user"
def _account_type(user_type: Any) -> str:
return _USER_TYPES.get(_safe_int(user_type), "unknown")
def _merge_users(
jellyfin_users: list[dict[str, Any]],
jellyseerr_jellyfin_users: list[dict[str, Any]] | None,
jellyseerr_users: list[dict[str, Any]] | None,
jellyseerr_client: JellyseerrClient | None,
) -> dict[str, Any]:
def _normalize(value: Any) -> str:
return str(value or "").strip().lower()
def _looks_like_email(value: Any) -> bool:
text = str(value or "").strip()
return bool(text and "@" in text and " " not in text)
def _pick_source_and_value(candidates: list[tuple[str, Any]]) -> tuple[str, str]:
for source, value in candidates:
if _looks_like_email(value):
return source, str(value).strip()
return "", ""
def _first_value(candidates: list[tuple[str, Any]]) -> tuple[str, str]:
for source, value in candidates:
text = str(value or "").strip()
if text:
return source, text
return "", ""
def _source_summary(name_source: str, email_source: str, avatar_source: str, access_source: str) -> str:
return ", ".join(
[
f"name={name_source or 'none'}",
f"email={email_source or 'none'}",
f"avatar={avatar_source or 'none'}",
f"access={access_source or 'none'}",
]
)
def _lookup_keys(item: dict[str, Any]) -> list[str]:
return [
_normalize(item.get("id")),
_normalize(item.get("Id")),
_normalize(item.get("userId")),
_normalize(item.get("user_id")),
_normalize(item.get("jellyfinUserId")),
_normalize(item.get("jellyfin_user_id")),
_normalize(item.get("jellyfinUsername")),
_normalize(item.get("jellyfin_username")),
_normalize(item.get("username")),
_normalize(item.get("displayName")),
_normalize(item.get("display_name")),
]
linked_by_jellyfin_id: dict[str, dict[str, Any]] = {}
for item in jellyseerr_jellyfin_users or []:
for key in (
item.get("id"),
item.get("Id"),
item.get("userId"),
item.get("user_id"),
item.get("jellyfinUserId"),
item.get("jellyfin_user_id"),
):
normalized = _normalize(key)
if normalized:
linked_by_jellyfin_id[normalized] = item
seerr_by_key: dict[str, dict[str, Any]] = {}
for item in jellyseerr_users or []:
for key in _lookup_keys(item):
if key:
seerr_by_key[key] = item
items: list[dict[str, Any]] = []
enriched_count = 0
for user in jellyfin_users:
jellyfin_id = str(user.get("Id") or user.get("id") or "")
jellyfin_name = str(user.get("Name") or user.get("name") or "")
jf_link = linked_by_jellyfin_id.get(_normalize(jellyfin_id))
seerr_user = None
for candidate in [
jellyfin_name,
(jf_link or {}).get("jellyfinUsername"),
(jf_link or {}).get("jellyfin_username"),
(jf_link or {}).get("username"),
(jf_link or {}).get("displayName"),
(jf_link or {}).get("display_name"),
]:
seerr_user = seerr_by_key.get(_normalize(candidate))
if seerr_user:
break
email_source, email = _pick_source_and_value(
[
("jellyseerr:user", (seerr_user or {}).get("email")),
("jellyseerr:jellyfin", (jf_link or {}).get("email")),
]
)
avatar_source, avatar = _first_value(
[
("jellyseerr:user", (seerr_user or {}).get("avatar")),
("jellyseerr:jellyfin", (jf_link or {}).get("thumb")),
("jellyseerr:jellyfin", (jf_link or {}).get("avatar")),
]
)
if avatar and jellyseerr_client:
avatar = jellyseerr_client.absolute_url(avatar)
permissions = _safe_int((seerr_user or {}).get("permissions"))
user_type = _safe_int((seerr_user or {}).get("userType") or (seerr_user or {}).get("user_type"))
role = _role_label(permissions)
access_source = "jellyseerr:user" if seerr_user else ""
name_source = "jellyfin"
summary = _source_summary(name_source, email_source, avatar_source, access_source)
if seerr_user or jf_link:
enriched_count += 1
items.append(
{
"jellyfin_id": jellyfin_id,
"username": jellyfin_name,
"display_name": jellyfin_name,
"email": email,
"email_source": email_source,
"avatar": avatar,
"avatar_source": avatar_source,
"contactable": bool(email),
"source": summary,
"source_summary": summary,
"name_source": name_source,
"access_source": access_source,
"jellyseerr_user_id": _safe_int((seerr_user or {}).get("id") or (seerr_user or {}).get("userId")) or None,
"jellyseerr_username": str(
(seerr_user or {}).get("username") or (seerr_user or {}).get("jellyfinUsername") or ""
),
"user_type": user_type or None,
"user_type_label": _account_type(user_type),
"role": role,
"permissions": permissions,
"permissions_label": ", ".join(_permission_labels(permissions)),
"request_count": _safe_int((seerr_user or {}).get("requestCount")) or None,
}
)
logger.info(
"Users merged jellyfin=%s jellyseerr_jellyfin=%s jellyseerr_users=%s enriched=%s",
len(jellyfin_users),
len(jellyseerr_jellyfin_users or []),
len(jellyseerr_users or []),
enriched_count,
)
return {
"items": items,
"total": len(items),
"jellyseerr_configured": jellyseerr_client is not None,
"jellyseerr_available": bool(jellyseerr_users or jellyseerr_jellyfin_users),
"jellyseerr_jellyfin_user_count": len(jellyseerr_jellyfin_users or []),
"jellyseerr_user_count": len(jellyseerr_users or []),
"enriched_count": enriched_count,
}
@router.get("")
def get_users(
jellyfin: JellyfinClient = Depends(get_jellyfin_client),
jellyseerr: JellyseerrClient | None = Depends(get_jellyseerr_client),
) -> dict[str, Any]:
"""Return the known users, enriched with Jellyseerr data when available."""
jellyfin_users = jellyfin.users()
logger.info("Users endpoint fetched %s Jellyfin users", len(jellyfin_users))
jellyseerr_jellyfin_users: list[dict[str, Any]] | None = None
jellyseerr_users: list[dict[str, Any]] | None = None
jellyseerr_error = ""
if jellyseerr:
try:
jellyseerr_jellyfin_users = jellyseerr.jellyfin_users()
except Exception as exc: # pragma: no cover - network fallback
logger.exception("Jellyseerr Jellyfin-linked user fetch failed")
jellyseerr_error = f"Jellyseerr Jellyfin users fetch failed: {exc}"
try:
jellyseerr_users = jellyseerr.users()
except Exception as exc: # pragma: no cover - network fallback
logger.exception("Jellyseerr user list fetch failed")
jellyseerr_error = (
f"{jellyseerr_error}; " if jellyseerr_error else ""
) + f"Jellyseerr user list fetch failed: {exc}"
result = _merge_users(jellyfin_users, jellyseerr_jellyfin_users, jellyseerr_users, jellyseerr)
result["jellyseerr_error"] = jellyseerr_error
logger.info(
"Users response total=%s configured=%s available=%s enriched=%s error=%s",
result["total"],
result["jellyseerr_configured"],
result["jellyseerr_available"],
result["enriched_count"],
bool(jellyseerr_error),
)
return result
@router.get("/message/status")
def get_user_message_status(mail_queue=Depends(get_mail_queue)) -> dict[str, Any]:
"""Return the current background email queue status."""
return mail_queue.status()
@router.post("/message", status_code=status.HTTP_202_ACCEPTED)
async def post_user_message(
recipient_ids: str = Form(...),
subject: str = Form(...),
html_body: str = Form(""),
text_body: str = Form(""),
attachments: list[UploadFile] | None = File(default=None),
jellyfin: JellyfinClient = Depends(get_jellyfin_client),
jellyseerr: JellyseerrClient | None = Depends(get_jellyseerr_client),
mail_queue=Depends(get_mail_queue),
) -> dict[str, Any]:
"""Queue a single email to the selected users without blocking the API."""
try:
requested_ids = json.loads(recipient_ids)
except json.JSONDecodeError as exc:
raise HTTPException(status_code=400, detail=f"recipient_ids must be valid JSON: {exc}") from exc
if not isinstance(requested_ids, list):
raise HTTPException(status_code=400, detail="recipient_ids must be a JSON list")
cleaned_ids = [str(item).strip() for item in requested_ids if str(item).strip()]
if not cleaned_ids:
raise HTTPException(status_code=400, detail="At least one recipient is required")
subject = subject.strip()
if not subject:
raise HTTPException(status_code=400, detail="Subject is required")
directory = get_users(jellyfin=jellyfin, jellyseerr=jellyseerr)
users_by_id = {str(item.get("jellyfin_id") or ""): item for item in directory.get("items", [])}
recipients: list[str] = []
recipient_labels: list[str] = []
skipped: list[dict[str, str]] = []
for user_id in cleaned_ids:
item = users_by_id.get(user_id)
if not item:
skipped.append({"jellyfin_id": user_id, "reason": "not found"})
continue
email = str(item.get("email") or "").strip()
if not email:
skipped.append({"jellyfin_id": user_id, "reason": "missing email"})
continue
recipients.append(email)
recipient_labels.append(f"{item.get('display_name') or item.get('username') or user_id} <{email}>")
if not recipients:
raise HTTPException(status_code=400, detail="No selected users have a deliverable email address")
settings = get_settings()
validate_smtp_settings(settings)
queue_status = mail_queue.status()
if not queue_status["worker_running"]:
raise HTTPException(status_code=503, detail="Email queue worker is not running")
attachment_payloads: list[EmailAttachment] = []
for upload in attachments or []:
data = await upload.read()
if not data:
continue
attachment_payloads.append(
EmailAttachment(
filename=upload.filename or "attachment",
content_type=upload.content_type or "application/octet-stream",
data=data,
)
)
request_id = mail_queue.enqueue(
settings=settings,
recipients=recipients,
subject=subject,
html_body=html_body,
text_body=text_body,
attachments=attachment_payloads,
)
from_address = str(getattr(settings, "smtp_from_address", "") or "").strip() or str(
getattr(settings, "smtp_username", "") or ""
).strip()
logger.info(
"Users message queued request_id=%s subject=%s recipients=%s attachments=%s skipped=%s",
request_id,
subject,
len(recipients),
len(attachment_payloads),
len(skipped),
)
return {
"status": "queued",
"request_id": request_id,
"from_address": from_address,
"recipient_count": len(recipients),
"attachment_count": len(attachment_payloads),
"subject": subject,
"recipient_labels": recipient_labels,
"skipped": skipped,
}
backend/src/media_library_viewer_api/services/mailer.py
+1 -489
View File
@@ -1,489 +1 @@
"""SMTP email sending helpers for user communication workflows."""
from __future__ import annotations
import logging
import mimetypes
import socket
import smtplib
import ssl
from dataclasses import dataclass
from email.message import EmailMessage
from email.utils import formataddr
from html.parser import HTMLParser
from typing import Any, Iterable
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class EmailAttachment:
"""Attachment payload passed from the API layer."""
filename: str
content_type: str
data: bytes
class _HTMLToTextParser(HTMLParser):
"""Small HTML-to-text helper for plain-text fallback bodies."""
block_tags = {"p", "div", "section", "article", "header", "footer", "li", "tr", "td", "th", "br"}
def __init__(self) -> None:
super().__init__()
self.parts: list[str] = []
def handle_starttag(self, tag: str, attrs): # type: ignore[override]
if tag.lower() in self.block_tags and self.parts and not self.parts[-1].endswith("\n"):
self.parts.append("\n")
def handle_endtag(self, tag: str) -> None:
if tag.lower() in self.block_tags and self.parts and not self.parts[-1].endswith("\n"):
self.parts.append("\n")
def handle_data(self, data: str) -> None:
if data:
self.parts.append(data)
def text(self) -> str:
return "".join(self.parts)
def html_to_text(html: str) -> str:
"""Convert a small HTML body to readable plain text."""
parser = _HTMLToTextParser()
parser.feed(html or "")
text = parser.text()
lines = [line.rstrip() for line in text.splitlines()]
return "\n".join(line for line in lines if line).strip()
def _from_address(settings: object) -> str:
from_address = str(getattr(settings, "smtp_from_address", "") or "").strip()
if from_address:
return from_address
smtp_username = str(getattr(settings, "smtp_username", "") or "").strip()
if smtp_username:
return smtp_username
raise ValueError("SMTP from address is required (set SMTP_FROM_ADDRESS or SMTP_USERNAME)")
def validate_smtp_settings(settings: object) -> None:
"""Validate that the SMTP configuration is sufficient to send mail."""
smtp_host = str(getattr(settings, "smtp_host", "") or "").strip()
if not smtp_host:
raise ValueError("SMTP host is required")
_from_address(settings)
def _smtp_settings(settings: object) -> dict[str, object]:
smtp_host = str(getattr(settings, "smtp_host", "") or "").strip()
if not smtp_host:
raise ValueError("SMTP host is required")
smtp_port = int(getattr(settings, "smtp_port", 587) or 587)
smtp_username = str(getattr(settings, "smtp_username", "") or "").strip()
smtp_password = str(getattr(settings, "smtp_password", "") or "")
use_tls = bool(getattr(settings, "smtp_use_tls", True))
use_ssl = bool(getattr(settings, "smtp_use_ssl", False))
smtp_timeout = int(getattr(settings, "smtp_timeout", 30) or 30)
return {
"smtp_host": smtp_host,
"smtp_port": smtp_port,
"smtp_username": smtp_username,
"smtp_password": smtp_password,
"use_tls": use_tls,
"use_ssl": use_ssl,
"smtp_timeout": smtp_timeout,
}
def _smtp_mode_candidates(settings: object) -> list[dict[str, Any]]:
base = _smtp_settings(settings)
candidates = [dict(base, mode_label="configured")]
smtp_host = str(base["smtp_host"]).lower()
if "fastmail.com" in smtp_host:
fastmail_ssl = {
**base,
"smtp_port": 465,
"use_tls": False,
"use_ssl": True,
"mode_label": "Fastmail SSL 465",
}
fastmail_tls = {
**base,
"smtp_port": 587,
"use_tls": True,
"use_ssl": False,
"mode_label": "Fastmail STARTTLS 587",
}
for mode in (fastmail_ssl, fastmail_tls):
if not any(
candidate["smtp_port"] == mode["smtp_port"]
and candidate["use_tls"] == mode["use_tls"]
and candidate["use_ssl"] == mode["use_ssl"]
for candidate in candidates
):
candidates.append(mode)
return candidates
def _probe_smtp_connection(mode: dict[str, Any]) -> None:
context = ssl.create_default_context()
smtp_host = str(mode["smtp_host"])
smtp_port = int(mode["smtp_port"])
smtp_username = str(mode["smtp_username"])
smtp_password = str(mode["smtp_password"])
use_tls = bool(mode["use_tls"])
use_ssl = bool(mode["use_ssl"])
smtp_timeout = int(mode["smtp_timeout"])
if use_ssl:
smtp_connection = smtplib.SMTP_SSL(smtp_host, smtp_port, timeout=smtp_timeout)
else:
smtp_connection = smtplib.SMTP(smtp_host, smtp_port, timeout=smtp_timeout)
with smtp_connection as smtp:
if use_tls and not use_ssl:
smtp.ehlo()
smtp.starttls(context=context)
smtp.ehlo()
else:
smtp.ehlo()
if smtp_username:
smtp.login(smtp_username, smtp_password)
smtp.noop()
def _smtp_sender_not_authorized(error: Exception) -> bool:
code = getattr(error, "smtp_code", None)
raw_error = getattr(error, "smtp_error", b"")
if isinstance(raw_error, bytes):
raw_error_text = raw_error.decode(errors="ignore")
else:
raw_error_text = str(raw_error)
text = f"{code} {raw_error_text} {error}".lower()
return code in {551, 553} or "not authorised to send from this header address" in text or "not authorized to send from this header address" in text
def _smtp_attempt_metadata(mode: dict[str, Any]) -> dict[str, Any]:
transport = "SSL" if mode["use_ssl"] else "STARTTLS" if mode["use_tls"] else "plain SMTP"
return {
"label": str(mode.get("mode_label") or f"{mode['smtp_host']}:{mode['smtp_port']}"),
"smtp_host": str(mode["smtp_host"]),
"smtp_port": int(mode["smtp_port"]),
"use_tls": bool(mode["use_tls"]),
"use_ssl": bool(mode["use_ssl"]),
"transport": transport,
"auth_user": str(mode.get("smtp_username") or "") or "<none>",
}
def describe_smtp_error(error: Exception) -> str:
"""Convert SMTP failures into operator-friendly messages."""
chain: list[Exception] = []
current: Exception | None = error
while current is not None and current not in chain:
chain.append(current)
current = current.__cause__ if isinstance(current.__cause__, Exception) else None
for item in chain:
text = str(item).strip()
lowered = text.lower()
if isinstance(item, (TimeoutError, socket.timeout)) or "timed out" in lowered:
return (
"SMTP connection timed out while waiting for the server greeting. "
"Check host, port, network access, and SMTP_TIMEOUT."
)
if isinstance(item, smtplib.SMTPAuthenticationError):
return (
"SMTP authentication failed. Check SMTP_USERNAME and SMTP_PASSWORD "
"(Fastmail and similar providers usually require an app password)."
)
if isinstance(item, (smtplib.SMTPDataError, smtplib.SMTPResponseException)):
smtp_code = getattr(item, "smtp_code", None)
if smtp_code in {551, 553} or "not authorised to send from this header address" in lowered or "not authorized to send from this header address" in lowered:
return (
"SMTP server rejected the configured From address. Use an authorized alias "
"for this account or change SMTP_FROM_ADDRESS to a sender the provider allows."
)
if isinstance(item, smtplib.SMTPConnectError):
return "SMTP connection was rejected by the server. Check the host and port."
if isinstance(item, smtplib.SMTPServerDisconnected) and "timed out" in lowered:
return (
"SMTP connection timed out while waiting for the server greeting. "
"Check host, port, network access, and SMTP_TIMEOUT."
)
return f"SMTP delivery failed: {error}"
def build_email_message(
settings: object,
recipients: list[str],
subject: str,
html_body: str,
text_body: str,
attachments: Iterable[EmailAttachment] = (),
*,
sender_address: str | None = None,
reply_to_address: str | None = None,
) -> tuple[EmailMessage, str]:
"""Build a MIME email message with HTML and attachments."""
from_address = sender_address or _from_address(settings)
from_name = str(getattr(settings, "smtp_from_name", "") or "").strip() or "Manage"
msg = EmailMessage()
msg["Subject"] = subject
msg["From"] = formataddr((from_name, from_address))
msg["To"] = "Undisclosed recipients:;"
msg["Reply-To"] = reply_to_address or from_address
plain_text = text_body.strip() or html_to_text(html_body)
if html_body.strip():
msg.set_content(plain_text or " ")
msg.add_alternative(html_body, subtype="html")
else:
msg.set_content(plain_text or "")
for attachment in attachments:
content_type = attachment.content_type or mimetypes.guess_type(attachment.filename)[0] or "application/octet-stream"
maintype, subtype = content_type.split("/", 1) if "/" in content_type else ("application", "octet-stream")
msg.add_attachment(
attachment.data,
maintype=maintype,
subtype=subtype,
filename=attachment.filename or "attachment",
)
return msg, from_address
def _send_email_via_mode(
mode: dict[str, Any],
message: EmailMessage,
recipients: list[str],
from_address: str,
) -> None:
context = ssl.create_default_context()
smtp_host = str(mode["smtp_host"])
smtp_port = int(mode["smtp_port"])
smtp_username = str(mode["smtp_username"])
smtp_password = str(mode["smtp_password"])
use_tls = bool(mode["use_tls"])
use_ssl = bool(mode["use_ssl"])
smtp_timeout = int(mode["smtp_timeout"])
if use_ssl:
smtp_connection = smtplib.SMTP_SSL(smtp_host, smtp_port, timeout=smtp_timeout)
else:
smtp_connection = smtplib.SMTP(smtp_host, smtp_port, timeout=smtp_timeout)
with smtp_connection as smtp:
if use_tls and not use_ssl:
smtp.ehlo()
smtp.starttls(context=context)
smtp.ehlo()
else:
smtp.ehlo()
if smtp_username:
smtp.login(smtp_username, smtp_password)
smtp.send_message(message, from_addr=from_address, to_addrs=recipients)
def send_email_message(
settings: object,
recipients: list[str],
subject: str,
html_body: str,
text_body: str = "",
attachments: Iterable[EmailAttachment] = (),
) -> dict[str, object]:
"""Send a single outbound email to a recipient list via SMTP BCC."""
if not recipients:
raise ValueError("At least one recipient is required")
attachment_list = list(attachments)
preferred_from_address = _from_address(settings)
smtp_username = str(getattr(settings, "smtp_username", "") or "").strip()
message, from_address = build_email_message(
settings,
recipients,
subject,
html_body,
text_body,
attachment_list,
sender_address=preferred_from_address,
reply_to_address=preferred_from_address,
)
logger.info(
"SMTP send requested subject=%s recipients=%s from_address=%s auth_user=%s attachments=%s",
subject,
len(recipients),
from_address,
smtp_username or "<none>",
len(attachment_list),
)
attempts: list[dict[str, Any]] = []
last_error = ""
fallback_from_address = smtp_username if smtp_username and smtp_username != from_address else None
for mode in _smtp_mode_candidates(settings):
meta = _smtp_attempt_metadata(mode)
logger.info(
"SMTP send attempting label=%s host=%s port=%s transport=%s auth_user=%s from_address=%s",
meta["label"],
meta["smtp_host"],
meta["smtp_port"],
meta["transport"],
meta["auth_user"],
from_address,
)
try:
_send_email_via_mode(mode, message, recipients, from_address)
except Exception as exc:
last_error = describe_smtp_error(exc)
attempts.append(
{
"label": meta["label"],
"smtp_host": meta["smtp_host"],
"smtp_port": meta["smtp_port"],
"use_tls": meta["use_tls"],
"use_ssl": meta["use_ssl"],
"status": "failed",
"error": last_error,
}
)
if _smtp_sender_not_authorized(exc) and fallback_from_address:
logger.warning(
"SMTP send sender rejected label=%s host=%s port=%s transport=%s auth_user=%s from_address=%s error=%s; retrying with smtp_username",
meta["label"],
meta["smtp_host"],
meta["smtp_port"],
meta["transport"],
meta["auth_user"],
from_address,
last_error,
)
fallback_message, fallback_from = build_email_message(
settings,
recipients,
subject,
html_body,
text_body,
attachment_list,
sender_address=fallback_from_address,
reply_to_address=from_address,
)
try:
_send_email_via_mode(mode, fallback_message, recipients, fallback_from)
except Exception as fallback_exc:
last_error = describe_smtp_error(fallback_exc)
attempts.append(
{
"label": meta["label"],
"smtp_host": meta["smtp_host"],
"smtp_port": meta["smtp_port"],
"use_tls": meta["use_tls"],
"use_ssl": meta["use_ssl"],
"status": "failed",
"error": last_error,
"sender_fallback": True,
}
)
logger.warning(
"SMTP send fallback failed label=%s host=%s port=%s transport=%s auth_user=%s from_address=%s error=%s",
meta["label"],
meta["smtp_host"],
meta["smtp_port"],
meta["transport"],
meta["auth_user"],
fallback_from,
last_error,
)
continue
attempts.append(
{
"label": meta["label"],
"smtp_host": meta["smtp_host"],
"smtp_port": meta["smtp_port"],
"use_tls": meta["use_tls"],
"use_ssl": meta["use_ssl"],
"status": "ok",
"sender_fallback": True,
}
)
logger.info(
"SMTP send succeeded via smtp_username label=%s host=%s port=%s transport=%s auth_user=%s from_address=%s",
meta["label"],
meta["smtp_host"],
meta["smtp_port"],
meta["transport"],
meta["auth_user"],
fallback_from,
)
return {
"from_address": fallback_from,
"recipient_count": len(recipients),
"attachment_count": len(attachment_list),
"subject": subject,
"authenticated_as": smtp_username or None,
"selected_mode": {
"label": meta["label"],
"smtp_host": meta["smtp_host"],
"smtp_port": meta["smtp_port"],
"use_tls": meta["use_tls"],
"use_ssl": meta["use_ssl"],
},
"attempts": attempts,
}
logger.warning(
"SMTP send failed label=%s host=%s port=%s transport=%s auth_user=%s from_address=%s error=%s",
meta["label"],
meta["smtp_host"],
meta["smtp_port"],
meta["transport"],
meta["auth_user"],
from_address,
last_error,
)
continue
attempts.append(
{
"label": meta["label"],
"smtp_host": meta["smtp_host"],
"smtp_port": meta["smtp_port"],
"use_tls": meta["use_tls"],
"use_ssl": meta["use_ssl"],
"status": "ok",
}
)
logger.info(
"SMTP send succeeded label=%s host=%s port=%s transport=%s auth_user=%s from_address=%s",
meta["label"],
meta["smtp_host"],
meta["smtp_port"],
meta["transport"],
meta["auth_user"],
from_address,
)
return {
"from_address": from_address,
"recipient_count": len(recipients),
"attachment_count": len(attachment_list),
"subject": subject,
"authenticated_as": smtp_username or None,
"selected_mode": {
"label": meta["label"],
"smtp_host": meta["smtp_host"],
"smtp_port": meta["smtp_port"],
"use_tls": meta["use_tls"],
"use_ssl": meta["use_ssl"],
},
"attempts": attempts,
}
raise RuntimeError(last_error or "SMTP delivery failed")
from .mailer_impl import * # noqa: F401,F403
backend/src/media_library_viewer_api/services/mailer_impl.py
+489
View File
@@ -0,0 +1,489 @@
"""SMTP email sending helpers for user communication workflows."""
from __future__ import annotations
import logging
import mimetypes
import socket
import smtplib
import ssl
from dataclasses import dataclass
from email.message import EmailMessage
from email.utils import formataddr
from html.parser import HTMLParser
from typing import Any, Iterable
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class EmailAttachment:
"""Attachment payload passed from the API layer."""
filename: str
content_type: str
data: bytes
class _HTMLToTextParser(HTMLParser):
"""Small HTML-to-text helper for plain-text fallback bodies."""
block_tags = {"p", "div", "section", "article", "header", "footer", "li", "tr", "td", "th", "br"}
def __init__(self) -> None:
super().__init__()
self.parts: list[str] = []
def handle_starttag(self, tag: str, attrs): # type: ignore[override]
if tag.lower() in self.block_tags and self.parts and not self.parts[-1].endswith("\n"):
self.parts.append("\n")
def handle_endtag(self, tag: str) -> None:
if tag.lower() in self.block_tags and self.parts and not self.parts[-1].endswith("\n"):
self.parts.append("\n")
def handle_data(self, data: str) -> None:
if data:
self.parts.append(data)
def text(self) -> str:
return "".join(self.parts)
def html_to_text(html: str) -> str:
"""Convert a small HTML body to readable plain text."""
parser = _HTMLToTextParser()
parser.feed(html or "")
text = parser.text()
lines = [line.rstrip() for line in text.splitlines()]
return "\n".join(line for line in lines if line).strip()
def _from_address(settings: object) -> str:
from_address = str(getattr(settings, "smtp_from_address", "") or "").strip()
if from_address:
return from_address
smtp_username = str(getattr(settings, "smtp_username", "") or "").strip()
if smtp_username:
return smtp_username
raise ValueError("SMTP from address is required (set SMTP_FROM_ADDRESS or SMTP_USERNAME)")
def validate_smtp_settings(settings: object) -> None:
"""Validate that the SMTP configuration is sufficient to send mail."""
smtp_host = str(getattr(settings, "smtp_host", "") or "").strip()
if not smtp_host:
raise ValueError("SMTP host is required")
_from_address(settings)
def _smtp_settings(settings: object) -> dict[str, object]:
smtp_host = str(getattr(settings, "smtp_host", "") or "").strip()
if not smtp_host:
raise ValueError("SMTP host is required")
smtp_port = int(getattr(settings, "smtp_port", 587) or 587)
smtp_username = str(getattr(settings, "smtp_username", "") or "").strip()
smtp_password = str(getattr(settings, "smtp_password", "") or "")
use_tls = bool(getattr(settings, "smtp_use_tls", True))
use_ssl = bool(getattr(settings, "smtp_use_ssl", False))
smtp_timeout = int(getattr(settings, "smtp_timeout", 30) or 30)
return {
"smtp_host": smtp_host,
"smtp_port": smtp_port,
"smtp_username": smtp_username,
"smtp_password": smtp_password,
"use_tls": use_tls,
"use_ssl": use_ssl,
"smtp_timeout": smtp_timeout,
}
def _smtp_mode_candidates(settings: object) -> list[dict[str, Any]]:
base = _smtp_settings(settings)
candidates = [dict(base, mode_label="configured")]
smtp_host = str(base["smtp_host"]).lower()
if "fastmail.com" in smtp_host:
fastmail_ssl = {
**base,
"smtp_port": 465,
"use_tls": False,
"use_ssl": True,
"mode_label": "Fastmail SSL 465",
}
fastmail_tls = {
**base,
"smtp_port": 587,
"use_tls": True,
"use_ssl": False,
"mode_label": "Fastmail STARTTLS 587",
}
for mode in (fastmail_ssl, fastmail_tls):
if not any(
candidate["smtp_port"] == mode["smtp_port"]
and candidate["use_tls"] == mode["use_tls"]
and candidate["use_ssl"] == mode["use_ssl"]
for candidate in candidates
):
candidates.append(mode)
return candidates
def _probe_smtp_connection(mode: dict[str, Any]) -> None:
context = ssl.create_default_context()
smtp_host = str(mode["smtp_host"])
smtp_port = int(mode["smtp_port"])
smtp_username = str(mode["smtp_username"])
smtp_password = str(mode["smtp_password"])
use_tls = bool(mode["use_tls"])
use_ssl = bool(mode["use_ssl"])
smtp_timeout = int(mode["smtp_timeout"])
if use_ssl:
smtp_connection = smtplib.SMTP_SSL(smtp_host, smtp_port, timeout=smtp_timeout)
else:
smtp_connection = smtplib.SMTP(smtp_host, smtp_port, timeout=smtp_timeout)
with smtp_connection as smtp:
if use_tls and not use_ssl:
smtp.ehlo()
smtp.starttls(context=context)
smtp.ehlo()
else:
smtp.ehlo()
if smtp_username:
smtp.login(smtp_username, smtp_password)
smtp.noop()
def _smtp_sender_not_authorized(error: Exception) -> bool:
code = getattr(error, "smtp_code", None)
raw_error = getattr(error, "smtp_error", b"")
if isinstance(raw_error, bytes):
raw_error_text = raw_error.decode(errors="ignore")
else:
raw_error_text = str(raw_error)
text = f"{code} {raw_error_text} {error}".lower()
return code in {551, 553} or "not authorised to send from this header address" in text or "not authorized to send from this header address" in text
def _smtp_attempt_metadata(mode: dict[str, Any]) -> dict[str, Any]:
transport = "SSL" if mode["use_ssl"] else "STARTTLS" if mode["use_tls"] else "plain SMTP"
return {
"label": str(mode.get("mode_label") or f"{mode['smtp_host']}:{mode['smtp_port']}"),
"smtp_host": str(mode["smtp_host"]),
"smtp_port": int(mode["smtp_port"]),
"use_tls": bool(mode["use_tls"]),
"use_ssl": bool(mode["use_ssl"]),
"transport": transport,
"auth_user": str(mode.get("smtp_username") or "") or "<none>",
}
def describe_smtp_error(error: Exception) -> str:
"""Convert SMTP failures into operator-friendly messages."""
chain: list[Exception] = []
current: Exception | None = error
while current is not None and current not in chain:
chain.append(current)
current = current.__cause__ if isinstance(current.__cause__, Exception) else None
for item in chain:
text = str(item).strip()
lowered = text.lower()
if isinstance(item, (TimeoutError, socket.timeout)) or "timed out" in lowered:
return (
"SMTP connection timed out while waiting for the server greeting. "
"Check host, port, network access, and SMTP_TIMEOUT."
)
if isinstance(item, smtplib.SMTPAuthenticationError):
return (
"SMTP authentication failed. Check SMTP_USERNAME and SMTP_PASSWORD "
"(Fastmail and similar providers usually require an app password)."
)
if isinstance(item, (smtplib.SMTPDataError, smtplib.SMTPResponseException)):
smtp_code = getattr(item, "smtp_code", None)
if smtp_code in {551, 553} or "not authorised to send from this header address" in lowered or "not authorized to send from this header address" in lowered:
return (
"SMTP server rejected the configured From address. Use an authorized alias "
"for this account or change SMTP_FROM_ADDRESS to a sender the provider allows."
)
if isinstance(item, smtplib.SMTPConnectError):
return "SMTP connection was rejected by the server. Check the host and port."
if isinstance(item, smtplib.SMTPServerDisconnected) and "timed out" in lowered:
return (
"SMTP connection timed out while waiting for the server greeting. "
"Check host, port, network access, and SMTP_TIMEOUT."
)
return f"SMTP delivery failed: {error}"
def build_email_message(
settings: object,
recipients: list[str],
subject: str,
html_body: str,
text_body: str,
attachments: Iterable[EmailAttachment] = (),
*,
sender_address: str | None = None,
reply_to_address: str | None = None,
) -> tuple[EmailMessage, str]:
"""Build a MIME email message with HTML and attachments."""
from_address = sender_address or _from_address(settings)
from_name = str(getattr(settings, "smtp_from_name", "") or "").strip() or "Manage"
msg = EmailMessage()
msg["Subject"] = subject
msg["From"] = formataddr((from_name, from_address))
msg["To"] = "Undisclosed recipients:;"
msg["Reply-To"] = reply_to_address or from_address
plain_text = text_body.strip() or html_to_text(html_body)
if html_body.strip():
msg.set_content(plain_text or " ")
msg.add_alternative(html_body, subtype="html")
else:
msg.set_content(plain_text or "")
for attachment in attachments:
content_type = attachment.content_type or mimetypes.guess_type(attachment.filename)[0] or "application/octet-stream"
maintype, subtype = content_type.split("/", 1) if "/" in content_type else ("application", "octet-stream")
msg.add_attachment(
attachment.data,
maintype=maintype,
subtype=subtype,
filename=attachment.filename or "attachment",
)
return msg, from_address
def _send_email_via_mode(
mode: dict[str, Any],
message: EmailMessage,
recipients: list[str],
from_address: str,
) -> None:
context = ssl.create_default_context()
smtp_host = str(mode["smtp_host"])
smtp_port = int(mode["smtp_port"])
smtp_username = str(mode["smtp_username"])
smtp_password = str(mode["smtp_password"])
use_tls = bool(mode["use_tls"])
use_ssl = bool(mode["use_ssl"])
smtp_timeout = int(mode["smtp_timeout"])
if use_ssl:
smtp_connection = smtplib.SMTP_SSL(smtp_host, smtp_port, timeout=smtp_timeout)
else:
smtp_connection = smtplib.SMTP(smtp_host, smtp_port, timeout=smtp_timeout)
with smtp_connection as smtp:
if use_tls and not use_ssl:
smtp.ehlo()
smtp.starttls(context=context)
smtp.ehlo()
else:
smtp.ehlo()
if smtp_username:
smtp.login(smtp_username, smtp_password)
smtp.send_message(message, from_addr=from_address, to_addrs=recipients)
def send_email_message(
settings: object,
recipients: list[str],
subject: str,
html_body: str,
text_body: str = "",
attachments: Iterable[EmailAttachment] = (),
) -> dict[str, object]:
"""Send a single outbound email to a recipient list via SMTP BCC."""
if not recipients:
raise ValueError("At least one recipient is required")
attachment_list = list(attachments)
preferred_from_address = _from_address(settings)
smtp_username = str(getattr(settings, "smtp_username", "") or "").strip()
message, from_address = build_email_message(
settings,
recipients,
subject,
html_body,
text_body,
attachment_list,
sender_address=preferred_from_address,
reply_to_address=preferred_from_address,
)
logger.info(
"SMTP send requested subject=%s recipients=%s from_address=%s auth_user=%s attachments=%s",
subject,
len(recipients),
from_address,
smtp_username or "<none>",
len(attachment_list),
)
attempts: list[dict[str, Any]] = []
last_error = ""
fallback_from_address = smtp_username if smtp_username and smtp_username != from_address else None
for mode in _smtp_mode_candidates(settings):
meta = _smtp_attempt_metadata(mode)
logger.info(
"SMTP send attempting label=%s host=%s port=%s transport=%s auth_user=%s from_address=%s",
meta["label"],
meta["smtp_host"],
meta["smtp_port"],
meta["transport"],
meta["auth_user"],
from_address,
)
try:
_send_email_via_mode(mode, message, recipients, from_address)
except Exception as exc:
last_error = describe_smtp_error(exc)
attempts.append(
{
"label": meta["label"],
"smtp_host": meta["smtp_host"],
"smtp_port": meta["smtp_port"],
"use_tls": meta["use_tls"],
"use_ssl": meta["use_ssl"],
"status": "failed",
"error": last_error,
}
)
if _smtp_sender_not_authorized(exc) and fallback_from_address:
logger.warning(
"SMTP send sender rejected label=%s host=%s port=%s transport=%s auth_user=%s from_address=%s error=%s; retrying with smtp_username",
meta["label"],
meta["smtp_host"],
meta["smtp_port"],
meta["transport"],
meta["auth_user"],
from_address,
last_error,
)
fallback_message, fallback_from = build_email_message(
settings,
recipients,
subject,
html_body,
text_body,
attachment_list,
sender_address=fallback_from_address,
reply_to_address=from_address,
)
try:
_send_email_via_mode(mode, fallback_message, recipients, fallback_from)
except Exception as fallback_exc:
last_error = describe_smtp_error(fallback_exc)
attempts.append(
{
"label": meta["label"],
"smtp_host": meta["smtp_host"],
"smtp_port": meta["smtp_port"],
"use_tls": meta["use_tls"],
"use_ssl": meta["use_ssl"],
"status": "failed",
"error": last_error,
"sender_fallback": True,
}
)
logger.warning(
"SMTP send fallback failed label=%s host=%s port=%s transport=%s auth_user=%s from_address=%s error=%s",
meta["label"],
meta["smtp_host"],
meta["smtp_port"],
meta["transport"],
meta["auth_user"],
fallback_from,
last_error,
)
continue
attempts.append(
{
"label": meta["label"],
"smtp_host": meta["smtp_host"],
"smtp_port": meta["smtp_port"],
"use_tls": meta["use_tls"],
"use_ssl": meta["use_ssl"],
"status": "ok",
"sender_fallback": True,
}
)
logger.info(
"SMTP send succeeded via smtp_username label=%s host=%s port=%s transport=%s auth_user=%s from_address=%s",
meta["label"],
meta["smtp_host"],
meta["smtp_port"],
meta["transport"],
meta["auth_user"],
fallback_from,
)
return {
"from_address": fallback_from,
"recipient_count": len(recipients),
"attachment_count": len(attachment_list),
"subject": subject,
"authenticated_as": smtp_username or None,
"selected_mode": {
"label": meta["label"],
"smtp_host": meta["smtp_host"],
"smtp_port": meta["smtp_port"],
"use_tls": meta["use_tls"],
"use_ssl": meta["use_ssl"],
},
"attempts": attempts,
}
logger.warning(
"SMTP send failed label=%s host=%s port=%s transport=%s auth_user=%s from_address=%s error=%s",
meta["label"],
meta["smtp_host"],
meta["smtp_port"],
meta["transport"],
meta["auth_user"],
from_address,
last_error,
)
continue
attempts.append(
{
"label": meta["label"],
"smtp_host": meta["smtp_host"],
"smtp_port": meta["smtp_port"],
"use_tls": meta["use_tls"],
"use_ssl": meta["use_ssl"],
"status": "ok",
}
)
logger.info(
"SMTP send succeeded label=%s host=%s port=%s transport=%s auth_user=%s from_address=%s",
meta["label"],
meta["smtp_host"],
meta["smtp_port"],
meta["transport"],
meta["auth_user"],
from_address,
)
return {
"from_address": from_address,
"recipient_count": len(recipients),
"attachment_count": len(attachment_list),
"subject": subject,
"authenticated_as": smtp_username or None,
"selected_mode": {
"label": meta["label"],
"smtp_host": meta["smtp_host"],
"smtp_port": meta["smtp_port"],
"use_tls": meta["use_tls"],
"use_ssl": meta["use_ssl"],
},
"attempts": attempts,
}
raise RuntimeError(last_error or "SMTP delivery failed")
backend/src/media_library_viewer_api/services/media_index.py
+1 -458
View File
@@ -1,458 +1 @@
"""SQLite-backed media inventory service.
This is the main step toward a frontend-agnostic architecture. The Streamlit UI
asks this service to build/query an index, but the same class could be exposed
through FastAPI to a React frontend without rewriting Jellyfin indexing logic.
"""
from __future__ import annotations
import logging
import sqlite3
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Iterable
from media_library_viewer_api.clients.jellyfin import JellyfinClient
from media_library_viewer_api.domain.media import display_media_row, normalize_media_item
from media_library_viewer_api.path_utils import resolve_remote_media_path
logger = logging.getLogger(__name__)
# Local generated database. It is ignored by git and can be rebuilt from
# Jellyfin metadata whenever needed.
DEFAULT_INDEX_PATH = Path(".cache/media_library_viewer/media_index.sqlite")
MEDIA_TYPES = "Movie,Episode,Video"
# Only values from this whitelist are interpolated into ORDER BY. User-selected
# sort keys map to these known SQL snippets to avoid SQL injection.
SORT_COLUMNS = {
"title": "title COLLATE NOCASE",
"series": "series COLLATE NOCASE",
"season": "season_number",
"episode": "episode",
"type": "type COLLATE NOCASE",
"year": "year",
"runtime": "runtime_min",
"size": "size_bytes",
"bitrate": "bitrate_bps",
"hdr": "hdr",
"video": "video COLLATE NOCASE",
"resolution": "height",
"date_added": "date_added_ts",
"library": "library_name COLLATE NOCASE",
"path": "path COLLATE NOCASE",
}
def _estimate_remaining_seconds(elapsed_seconds: float, progress: float | None) -> float | None:
if progress is None:
return None
progress = max(0.0, min(1.0, progress))
if progress <= 0.0:
return None
return max(0.0, elapsed_seconds * (1.0 - progress) / progress)
class MediaIndexBuildCancelled(Exception):
"""Raised when a media index build is requested to stop."""
@dataclass(frozen=True)
class MediaIndexStatus:
"""Lightweight status object displayed by the Media tab."""
exists: bool
item_count: int = 0
updated_at: int | None = None
updated_at_label: str = ""
build_duration_seconds: float | None = None
build_running: bool = False
build_stage: str = ""
build_message: str = ""
build_progress: float | None = None
build_items_processed: int = 0
build_items_total: int = 0
build_current_library: str = ""
build_library_index: int = 0
build_libraries_total: int = 0
build_library_progress: float | None = None
build_library_items_processed: int = 0
build_library_items_total: int = 0
build_elapsed_seconds: float | None = None
build_eta_seconds: float | None = None
build_library_elapsed_seconds: float | None = None
build_library_eta_seconds: float | None = None
build_cancel_requested: bool = False
build_pid: int | None = None
build_error: str = ""
class MediaIndex:
"""SQLite-backed media inventory.
This class is UI-framework independent. Streamlit, a future FastAPI backend,
or a React-facing API can all use this service.
"""
def __init__(self, db_path: Path | str = DEFAULT_INDEX_PATH):
self.db_path = Path(db_path)
self.db_path.parent.mkdir(parents=True, exist_ok=True)
def connect(self) -> sqlite3.Connection:
"""Open a sqlite connection configured to return Row objects."""
conn = sqlite3.connect(self.db_path, timeout=30)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=30000")
return conn
def init_schema(self) -> None:
"""Create tables/indexes if this is the first use of the index."""
with self.connect() as conn:
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS media_items (
id TEXT PRIMARY KEY,
title TEXT,
series TEXT,
season TEXT,
season_number INTEGER,
episode INTEGER,
type TEXT,
year INTEGER,
runtime_ticks INTEGER,
runtime_min INTEGER,
size_bytes INTEGER,
bitrate_bps INTEGER,
hdr INTEGER,
video TEXT,
width INTEGER,
height INTEGER,
resolution TEXT,
date_added TEXT,
date_added_ts INTEGER,
path TEXT,
library_id TEXT,
library_name TEXT
);
CREATE TABLE IF NOT EXISTS index_metadata (
key TEXT PRIMARY KEY,
value TEXT
);
CREATE INDEX IF NOT EXISTS idx_media_type ON media_items(type);
CREATE INDEX IF NOT EXISTS idx_media_library ON media_items(library_id);
CREATE INDEX IF NOT EXISTS idx_media_title ON media_items(title COLLATE NOCASE);
CREATE INDEX IF NOT EXISTS idx_media_series ON media_items(series COLLATE NOCASE);
CREATE INDEX IF NOT EXISTS idx_media_date_added ON media_items(date_added_ts);
CREATE INDEX IF NOT EXISTS idx_media_size ON media_items(size_bytes);
CREATE INDEX IF NOT EXISTS idx_media_bitrate ON media_items(bitrate_bps);
"""
)
def set_metadata(self, key: str, value: str | int | float) -> None:
"""Store a small string metadata value, e.g. build duration."""
self.init_schema()
with self.connect() as conn:
conn.execute(
"INSERT OR REPLACE INTO index_metadata (key, value) VALUES (?, ?)",
(key, str(value)),
)
def replace_items(self, rows: Iterable[dict[str, Any]]) -> int:
"""Atomically replace indexed media rows with a freshly built set."""
self.init_schema()
row_list = list(rows)
columns = [
"id",
"title",
"series",
"season",
"season_number",
"episode",
"type",
"year",
"runtime_ticks",
"runtime_min",
"size_bytes",
"bitrate_bps",
"hdr",
"video",
"width",
"height",
"resolution",
"date_added",
"date_added_ts",
"path",
"library_id",
"library_name",
]
placeholders = ",".join(["?"] * len(columns))
with self.connect() as conn:
conn.execute("DELETE FROM media_items")
conn.executemany(
f"INSERT OR REPLACE INTO media_items ({','.join(columns)}) VALUES ({placeholders})",
[[row.get(column) for column in columns] for row in row_list],
)
conn.execute(
"INSERT OR REPLACE INTO index_metadata (key, value) VALUES ('updated_at', ?)",
(str(int(time.time())),),
)
return len(row_list)
def status(self) -> MediaIndexStatus:
"""Return existence, count, update time, and last build duration."""
if not self.db_path.exists():
return MediaIndexStatus(exists=False)
try:
with self.connect() as conn:
item_count = int(conn.execute("SELECT COUNT(*) FROM media_items").fetchone()[0])
meta = {
row[0]: row[1]
for row in conn.execute("SELECT key, value FROM index_metadata").fetchall()
}
except sqlite3.Error:
return MediaIndexStatus(exists=False)
updated_at_raw = meta.get("updated_at", "")
updated_at = int(updated_at_raw) if str(updated_at_raw).isdigit() else None
label = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(updated_at)) if updated_at else ""
duration_raw = meta.get("build_duration_seconds")
build_duration = None
if duration_raw is not None:
try:
build_duration = float(duration_raw)
except (TypeError, ValueError):
build_duration = None
def _bool(key: str, default: bool = False) -> bool:
value = str(meta.get(key, str(default))).strip().lower()
return value in {"1", "true", "yes", "on"}
def _int(key: str, default: int = 0) -> int:
value = meta.get(key, default)
try:
return int(value)
except (TypeError, ValueError):
return default
def _float(key: str) -> float | None:
value = meta.get(key)
if value in (None, ""):
return None
try:
return float(value)
except (TypeError, ValueError):
return None
return MediaIndexStatus(
exists=True,
item_count=item_count,
updated_at=updated_at,
updated_at_label=label,
build_duration_seconds=build_duration,
build_running=_bool("build_running"),
build_stage=str(meta.get("build_stage", "")),
build_message=str(meta.get("build_message", "")),
build_progress=_float("build_progress"),
build_items_processed=_int("build_items_processed"),
build_items_total=_int("build_items_total"),
build_current_library=str(meta.get("build_current_library", "")),
build_library_index=_int("build_library_index"),
build_libraries_total=_int("build_libraries_total"),
build_library_progress=_float("build_library_progress"),
build_library_items_processed=_int("build_library_items_processed"),
build_library_items_total=_int("build_library_items_total"),
build_elapsed_seconds=_float("build_elapsed_seconds"),
build_eta_seconds=_float("build_eta_seconds"),
build_library_elapsed_seconds=_float("build_library_elapsed_seconds"),
build_library_eta_seconds=_float("build_library_eta_seconds"),
build_cancel_requested=_bool("build_cancel_requested"),
build_pid=_int("build_pid") or None,
build_error=str(meta.get("build_error", "")),
)
def query(
self,
library_id: str | None = None,
library_ids: list[str] | None = None,
media_types: list[str] | None = None,
search: str = "",
hdr_filter: str = "All",
sort_key: str = "title",
sort_order: str = "Ascending",
limit: int = 100,
offset: int = 0,
) -> tuple[list[dict[str, Any]], int]:
"""Query indexed media with full-index filters, sorting, and pagination."""
self.init_schema()
where = []
params: list[Any] = []
if library_ids:
where.append("library_id IN (" + ",".join(["?"] * len(library_ids)) + ")")
params.extend(library_ids)
elif library_id:
where.append("library_id = ?")
params.append(library_id)
if media_types:
where.append("type IN (" + ",".join(["?"] * len(media_types)) + ")")
params.extend(media_types)
if search:
needle = f"%{search.lower()}%"
where.append("(LOWER(title) LIKE ? OR LOWER(series) LIKE ? OR LOWER(path) LIKE ?)")
params.extend([needle, needle, needle])
if hdr_filter == "HDR only":
where.append("hdr = 1")
elif hdr_filter == "SDR/unknown only":
where.append("(hdr IS NULL OR hdr = 0)")
where_sql = " WHERE " + " AND ".join(where) if where else ""
sort_sql = SORT_COLUMNS.get(sort_key, SORT_COLUMNS["title"])
direction = "DESC" if sort_order == "Descending" else "ASC"
# Always add stable tie-breakers.
order_sql = f" ORDER BY {sort_sql} {direction}, series COLLATE NOCASE ASC, season_number ASC, episode ASC, title COLLATE NOCASE ASC"
with self.connect() as conn:
total = int(conn.execute("SELECT COUNT(*) FROM media_items" + where_sql, params).fetchone()[0])
rows = conn.execute(
"SELECT * FROM media_items" + where_sql + order_sql + " LIMIT ? OFFSET ?",
[*params, int(limit), int(offset)],
).fetchall()
return [display_media_row(dict(row)) for row in rows], total
def build_media_index(
client: JellyfinClient,
user_id: str,
libraries: list[dict[str, Any]],
index: MediaIndex | None = None,
page_size: int = 500,
media_root: str = "",
fallback_prefix: str = "",
progress_callback: Callable[[dict[str, Any]], None] | None = None,
should_cancel: Callable[[], bool] | None = None,
) -> int:
"""Fetch Jellyfin pages for all selected libraries and rebuild the index."""
index = index or MediaIndex()
started_at = time.perf_counter()
normalized_rows: list[dict[str, Any]] = []
processed_total = 0
expected_total = 0
current_library_name = ""
current_library_index = 0
current_library_processed = 0
current_library_total = 0
current_library_started_at = started_at
def ensure_not_cancelled() -> None:
if should_cancel and should_cancel():
raise MediaIndexBuildCancelled()
def emit(stage: str, message: str) -> None:
if not progress_callback:
return
elapsed_seconds = time.perf_counter() - started_at
library_elapsed_seconds = time.perf_counter() - current_library_started_at
overall_progress = (processed_total / expected_total) if expected_total else None
library_progress = (current_library_processed / current_library_total) if current_library_total else None
progress_callback(
{
"stage": stage,
"message": message,
"processed": processed_total,
"total": expected_total,
"progress": overall_progress,
"elapsed_seconds": elapsed_seconds,
"eta_seconds": _estimate_remaining_seconds(elapsed_seconds, overall_progress),
"library": current_library_name,
"library_index": current_library_index,
"libraries_total": len(libraries),
"library_processed": current_library_processed,
"library_total": current_library_total,
"library_progress": library_progress,
"library_elapsed_seconds": library_elapsed_seconds if current_library_total else None,
"library_eta_seconds": _estimate_remaining_seconds(library_elapsed_seconds, library_progress),
}
)
ensure_not_cancelled()
logger.info("Media index build starting libraries=%s page_size=%s", len(libraries), page_size)
emit("starting", "Starting media index build")
for library_index, library in enumerate(libraries, start=1):
library_id = library.get("Id")
current_library_name = library.get("Name", "")
current_library_index = library_index
current_library_processed = 0
current_library_total = 0
current_library_started_at = time.perf_counter()
if not library_id:
continue
ensure_not_cancelled()
logger.info(
"Media index scanning library index=%s/%s name=%s id=%s",
library_index,
len(libraries),
current_library_name or "Library",
library_id,
)
emit("library-starting", f"Scanning {current_library_name or 'Library'}")
start = 0
discovered_library_total = None
while True:
ensure_not_cancelled()
response = client.items(
user_id=user_id,
parent_id=library_id,
start_index=start,
limit=page_size,
include_item_types=MEDIA_TYPES,
recursive=True,
sort_by="SortName",
sort_order="Ascending",
)
ensure_not_cancelled()
items = response.get("Items", [])
if discovered_library_total is None:
discovered_library_total = int(response.get("TotalRecordCount", len(items)))
current_library_total = max(discovered_library_total, 0)
expected_total += current_library_total
normalized_rows.extend(
{
**row,
"path": resolve_remote_media_path(row.get("path", ""), media_root, fallback_prefix),
}
for row in (
normalize_media_item(item, library_id, current_library_name)
for item in items
)
)
processed_total += len(items)
current_library_processed += len(items)
start += len(items)
ensure_not_cancelled()
emit(
"building",
f"{current_library_name or 'Library'}: {current_library_processed} / {current_library_total or '?'} items",
)
logger.debug(
"Media index progress library=%s processed=%s/%s total_processed=%s",
current_library_name or "Library",
current_library_processed,
current_library_total,
processed_total,
)
total = int(response.get("TotalRecordCount", start))
if not items or start >= total:
break
ensure_not_cancelled()
logger.info("Media index finalizing rows=%s", len(normalized_rows))
emit("finalizing", "Writing index to disk")
ensure_not_cancelled()
count = index.replace_items(normalized_rows)
duration = time.perf_counter() - started_at
index.set_metadata("build_duration_seconds", f"{duration:.3f}")
processed_total = count
current_library_processed = current_library_total
emit("completed", f"Indexed {count} items in {duration:.1f}s")
logger.info("Media index build completed count=%s duration=%.2fs", count, duration)
return count
from media_library_viewer_api.services.media_index_impl import * # noqa: F401,F403
backend/src/media_library_viewer_api/services/media_index_impl.py
+458
View File
@@ -0,0 +1,458 @@
"""SQLite-backed media inventory service.
This is the main step toward a frontend-agnostic architecture. The Streamlit UI
asks this service to build/query an index, but the same class could be exposed
through FastAPI to a React frontend without rewriting Jellyfin indexing logic.
"""
from __future__ import annotations
import logging
import sqlite3
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Iterable
from media_library_viewer_api.clients.jellyfin import JellyfinClient
from media_library_viewer_api.domain.media import display_media_row, normalize_media_item
from media_library_viewer_api.path_utils import resolve_remote_media_path
logger = logging.getLogger(__name__)
# Local generated database. It is ignored by git and can be rebuilt from
# Jellyfin metadata whenever needed.
DEFAULT_INDEX_PATH = Path(".cache/media_library_viewer/media_index.sqlite")
MEDIA_TYPES = "Movie,Episode,Video"
# Only values from this whitelist are interpolated into ORDER BY. User-selected
# sort keys map to these known SQL snippets to avoid SQL injection.
SORT_COLUMNS = {
"title": "title COLLATE NOCASE",
"series": "series COLLATE NOCASE",
"season": "season_number",
"episode": "episode",
"type": "type COLLATE NOCASE",
"year": "year",
"runtime": "runtime_min",
"size": "size_bytes",
"bitrate": "bitrate_bps",
"hdr": "hdr",
"video": "video COLLATE NOCASE",
"resolution": "height",
"date_added": "date_added_ts",
"library": "library_name COLLATE NOCASE",
"path": "path COLLATE NOCASE",
}
def _estimate_remaining_seconds(elapsed_seconds: float, progress: float | None) -> float | None:
if progress is None:
return None
progress = max(0.0, min(1.0, progress))
if progress <= 0.0:
return None
return max(0.0, elapsed_seconds * (1.0 - progress) / progress)
class MediaIndexBuildCancelled(Exception):
"""Raised when a media index build is requested to stop."""
@dataclass(frozen=True)
class MediaIndexStatus:
"""Lightweight status object displayed by the Media tab."""
exists: bool
item_count: int = 0
updated_at: int | None = None
updated_at_label: str = ""
build_duration_seconds: float | None = None
build_running: bool = False
build_stage: str = ""
build_message: str = ""
build_progress: float | None = None
build_items_processed: int = 0
build_items_total: int = 0
build_current_library: str = ""
build_library_index: int = 0
build_libraries_total: int = 0
build_library_progress: float | None = None
build_library_items_processed: int = 0
build_library_items_total: int = 0
build_elapsed_seconds: float | None = None
build_eta_seconds: float | None = None
build_library_elapsed_seconds: float | None = None
build_library_eta_seconds: float | None = None
build_cancel_requested: bool = False
build_pid: int | None = None
build_error: str = ""
class MediaIndex:
"""SQLite-backed media inventory.
This class is UI-framework independent. Streamlit, a future FastAPI backend,
or a React-facing API can all use this service.
"""
def __init__(self, db_path: Path | str = DEFAULT_INDEX_PATH):
self.db_path = Path(db_path)
self.db_path.parent.mkdir(parents=True, exist_ok=True)
def connect(self) -> sqlite3.Connection:
"""Open a sqlite connection configured to return Row objects."""
conn = sqlite3.connect(self.db_path, timeout=30)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=30000")
return conn
def init_schema(self) -> None:
"""Create tables/indexes if this is the first use of the index."""
with self.connect() as conn:
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS media_items (
id TEXT PRIMARY KEY,
title TEXT,
series TEXT,
season TEXT,
season_number INTEGER,
episode INTEGER,
type TEXT,
year INTEGER,
runtime_ticks INTEGER,
runtime_min INTEGER,
size_bytes INTEGER,
bitrate_bps INTEGER,
hdr INTEGER,
video TEXT,
width INTEGER,
height INTEGER,
resolution TEXT,
date_added TEXT,
date_added_ts INTEGER,
path TEXT,
library_id TEXT,
library_name TEXT
);
CREATE TABLE IF NOT EXISTS index_metadata (
key TEXT PRIMARY KEY,
value TEXT
);
CREATE INDEX IF NOT EXISTS idx_media_type ON media_items(type);
CREATE INDEX IF NOT EXISTS idx_media_library ON media_items(library_id);
CREATE INDEX IF NOT EXISTS idx_media_title ON media_items(title COLLATE NOCASE);
CREATE INDEX IF NOT EXISTS idx_media_series ON media_items(series COLLATE NOCASE);
CREATE INDEX IF NOT EXISTS idx_media_date_added ON media_items(date_added_ts);
CREATE INDEX IF NOT EXISTS idx_media_size ON media_items(size_bytes);
CREATE INDEX IF NOT EXISTS idx_media_bitrate ON media_items(bitrate_bps);
"""
)
def set_metadata(self, key: str, value: str | int | float) -> None:
"""Store a small string metadata value, e.g. build duration."""
self.init_schema()
with self.connect() as conn:
conn.execute(
"INSERT OR REPLACE INTO index_metadata (key, value) VALUES (?, ?)",
(key, str(value)),
)
def replace_items(self, rows: Iterable[dict[str, Any]]) -> int:
"""Atomically replace indexed media rows with a freshly built set."""
self.init_schema()
row_list = list(rows)
columns = [
"id",
"title",
"series",
"season",
"season_number",
"episode",
"type",
"year",
"runtime_ticks",
"runtime_min",
"size_bytes",
"bitrate_bps",
"hdr",
"video",
"width",
"height",
"resolution",
"date_added",
"date_added_ts",
"path",
"library_id",
"library_name",
]
placeholders = ",".join(["?"] * len(columns))
with self.connect() as conn:
conn.execute("DELETE FROM media_items")
conn.executemany(
f"INSERT OR REPLACE INTO media_items ({','.join(columns)}) VALUES ({placeholders})",
[[row.get(column) for column in columns] for row in row_list],
)
conn.execute(
"INSERT OR REPLACE INTO index_metadata (key, value) VALUES ('updated_at', ?)",
(str(int(time.time())),),
)
return len(row_list)
def status(self) -> MediaIndexStatus:
"""Return existence, count, update time, and last build duration."""
if not self.db_path.exists():
return MediaIndexStatus(exists=False)
try:
with self.connect() as conn:
item_count = int(conn.execute("SELECT COUNT(*) FROM media_items").fetchone()[0])
meta = {
row[0]: row[1]
for row in conn.execute("SELECT key, value FROM index_metadata").fetchall()
}
except sqlite3.Error:
return MediaIndexStatus(exists=False)
updated_at_raw = meta.get("updated_at", "")
updated_at = int(updated_at_raw) if str(updated_at_raw).isdigit() else None
label = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(updated_at)) if updated_at else ""
duration_raw = meta.get("build_duration_seconds")
build_duration = None
if duration_raw is not None:
try:
build_duration = float(duration_raw)
except (TypeError, ValueError):
build_duration = None
def _bool(key: str, default: bool = False) -> bool:
value = str(meta.get(key, str(default))).strip().lower()
return value in {"1", "true", "yes", "on"}
def _int(key: str, default: int = 0) -> int:
value = meta.get(key, default)
try:
return int(value)
except (TypeError, ValueError):
return default
def _float(key: str) -> float | None:
value = meta.get(key)
if value in (None, ""):
return None
try:
return float(value)
except (TypeError, ValueError):
return None
return MediaIndexStatus(
exists=True,
item_count=item_count,
updated_at=updated_at,
updated_at_label=label,
build_duration_seconds=build_duration,
build_running=_bool("build_running"),
build_stage=str(meta.get("build_stage", "")),
build_message=str(meta.get("build_message", "")),
build_progress=_float("build_progress"),
build_items_processed=_int("build_items_processed"),
build_items_total=_int("build_items_total"),
build_current_library=str(meta.get("build_current_library", "")),
build_library_index=_int("build_library_index"),
build_libraries_total=_int("build_libraries_total"),
build_library_progress=_float("build_library_progress"),
build_library_items_processed=_int("build_library_items_processed"),
build_library_items_total=_int("build_library_items_total"),
build_elapsed_seconds=_float("build_elapsed_seconds"),
build_eta_seconds=_float("build_eta_seconds"),
build_library_elapsed_seconds=_float("build_library_elapsed_seconds"),
build_library_eta_seconds=_float("build_library_eta_seconds"),
build_cancel_requested=_bool("build_cancel_requested"),
build_pid=_int("build_pid") or None,
build_error=str(meta.get("build_error", "")),
)
def query(
self,
library_id: str | None = None,
library_ids: list[str] | None = None,
media_types: list[str] | None = None,
search: str = "",
hdr_filter: str = "All",
sort_key: str = "title",
sort_order: str = "Ascending",
limit: int = 100,
offset: int = 0,
) -> tuple[list[dict[str, Any]], int]:
"""Query indexed media with full-index filters, sorting, and pagination."""
self.init_schema()
where = []
params: list[Any] = []
if library_ids:
where.append("library_id IN (" + ",".join(["?"] * len(library_ids)) + ")")
params.extend(library_ids)
elif library_id:
where.append("library_id = ?")
params.append(library_id)
if media_types:
where.append("type IN (" + ",".join(["?"] * len(media_types)) + ")")
params.extend(media_types)
if search:
needle = f"%{search.lower()}%"
where.append("(LOWER(title) LIKE ? OR LOWER(series) LIKE ? OR LOWER(path) LIKE ?)")
params.extend([needle, needle, needle])
if hdr_filter == "HDR only":
where.append("hdr = 1")
elif hdr_filter == "SDR/unknown only":
where.append("(hdr IS NULL OR hdr = 0)")
where_sql = " WHERE " + " AND ".join(where) if where else ""
sort_sql = SORT_COLUMNS.get(sort_key, SORT_COLUMNS["title"])
direction = "DESC" if sort_order == "Descending" else "ASC"
# Always add stable tie-breakers.
order_sql = f" ORDER BY {sort_sql} {direction}, series COLLATE NOCASE ASC, season_number ASC, episode ASC, title COLLATE NOCASE ASC"
with self.connect() as conn:
total = int(conn.execute("SELECT COUNT(*) FROM media_items" + where_sql, params).fetchone()[0])
rows = conn.execute(
"SELECT * FROM media_items" + where_sql + order_sql + " LIMIT ? OFFSET ?",
[*params, int(limit), int(offset)],
).fetchall()
return [display_media_row(dict(row)) for row in rows], total
def build_media_index(
client: JellyfinClient,
user_id: str,
libraries: list[dict[str, Any]],
index: MediaIndex | None = None,
page_size: int = 500,
media_root: str = "",
fallback_prefix: str = "",
progress_callback: Callable[[dict[str, Any]], None] | None = None,
should_cancel: Callable[[], bool] | None = None,
) -> int:
"""Fetch Jellyfin pages for all selected libraries and rebuild the index."""
index = index or MediaIndex()
started_at = time.perf_counter()
normalized_rows: list[dict[str, Any]] = []
processed_total = 0
expected_total = 0
current_library_name = ""
current_library_index = 0
current_library_processed = 0
current_library_total = 0
current_library_started_at = started_at
def ensure_not_cancelled() -> None:
if should_cancel and should_cancel():
raise MediaIndexBuildCancelled()
def emit(stage: str, message: str) -> None:
if not progress_callback:
return
elapsed_seconds = time.perf_counter() - started_at
library_elapsed_seconds = time.perf_counter() - current_library_started_at
overall_progress = (processed_total / expected_total) if expected_total else None
library_progress = (current_library_processed / current_library_total) if current_library_total else None
progress_callback(
{
"stage": stage,
"message": message,
"processed": processed_total,
"total": expected_total,
"progress": overall_progress,
"elapsed_seconds": elapsed_seconds,
"eta_seconds": _estimate_remaining_seconds(elapsed_seconds, overall_progress),
"library": current_library_name,
"library_index": current_library_index,
"libraries_total": len(libraries),
"library_processed": current_library_processed,
"library_total": current_library_total,
"library_progress": library_progress,
"library_elapsed_seconds": library_elapsed_seconds if current_library_total else None,
"library_eta_seconds": _estimate_remaining_seconds(library_elapsed_seconds, library_progress),
}
)
ensure_not_cancelled()
logger.info("Media index build starting libraries=%s page_size=%s", len(libraries), page_size)
emit("starting", "Starting media index build")
for library_index, library in enumerate(libraries, start=1):
library_id = library.get("Id")
current_library_name = library.get("Name", "")
current_library_index = library_index
current_library_processed = 0
current_library_total = 0
current_library_started_at = time.perf_counter()
if not library_id:
continue
ensure_not_cancelled()
logger.info(
"Media index scanning library index=%s/%s name=%s id=%s",
library_index,
len(libraries),
current_library_name or "Library",
library_id,
)
emit("library-starting", f"Scanning {current_library_name or 'Library'}")
start = 0
discovered_library_total = None
while True:
ensure_not_cancelled()
response = client.items(
user_id=user_id,
parent_id=library_id,
start_index=start,
limit=page_size,
include_item_types=MEDIA_TYPES,
recursive=True,
sort_by="SortName",
sort_order="Ascending",
)
ensure_not_cancelled()
items = response.get("Items", [])
if discovered_library_total is None:
discovered_library_total = int(response.get("TotalRecordCount", len(items)))
current_library_total = max(discovered_library_total, 0)
expected_total += current_library_total
normalized_rows.extend(
{
**row,
"path": resolve_remote_media_path(row.get("path", ""), media_root, fallback_prefix),
}
for row in (
normalize_media_item(item, library_id, current_library_name)
for item in items
)
)
processed_total += len(items)
current_library_processed += len(items)
start += len(items)
ensure_not_cancelled()
emit(
"building",
f"{current_library_name or 'Library'}: {current_library_processed} / {current_library_total or '?'} items",
)
logger.debug(
"Media index progress library=%s processed=%s/%s total_processed=%s",
current_library_name or "Library",
current_library_processed,
current_library_total,
processed_total,
)
total = int(response.get("TotalRecordCount", start))
if not items or start >= total:
break
ensure_not_cancelled()
logger.info("Media index finalizing rows=%s", len(normalized_rows))
emit("finalizing", "Writing index to disk")
ensure_not_cancelled()
count = index.replace_items(normalized_rows)
duration = time.perf_counter() - started_at
index.set_metadata("build_duration_seconds", f"{duration:.3f}")
processed_total = count
current_library_processed = current_library_total
emit("completed", f"Indexed {count} items in {duration:.1f}s")
logger.info("Media index build completed count=%s duration=%.2fs", count, duration)
return count