embycovers/services/emby_collections.py

"""Favourites collection detection and item operations.

A "favourites collection" is any Emby collection (a ``BoxSet``) named
``"{UserName} Favorites"``. Unlike playlists, collection membership is keyed by
the item's own id (there is no per-entry id), so additions and removals operate
on item ids via ``/Collections/{id}/Items``.
"""

from __future__ import annotations

FAVORITES_SUFFIX = "Favorites"

# Fields requested for every collection/candidate item so the recommendation
# engine and the UI have what they need in one round trip.
ITEM_FIELDS = (
    "Genres,Studios,Tags,People,ProductionYear,RunTimeTicks,"
    "SeriesName,CommunityRating,MediaType,Overview"
)

# Emby stores runtime as 100ns ticks. 1 minute = 60 * 1e7 ticks.
_TICKS_PER_MINUTE = 600_000_000


def parse_favorites_owner(collection_name: str | None) -> str | None:
    """Return the owner name from ``"{Name} Favorites"`` or ``None``.

    Matching is case-insensitive on the suffix but preserves the owner's casing.
    ``"Favorites"`` on its own (no owner) is not a per-user favourites collection.
    """
    if not collection_name:
        return None
    name = collection_name.strip()
    suffix = " " + FAVORITES_SUFFIX
    if len(name) <= len(suffix):
        return None
    if name[-len(suffix):].casefold() != suffix.casefold():
        return None
    owner = name[: -len(suffix)].strip()
    return owner or None


def normalize_item(raw: dict) -> dict:
    """Flatten an Emby item into the shape the feature uses everywhere."""
    user_data = raw.get("UserData") or {}
    ticks = raw.get("RunTimeTicks")
    runtime_minutes = round(ticks / _TICKS_PER_MINUTE) if ticks else None
    people = raw.get("People") or []
    return {
        "id": raw.get("Id", ""),
        "title": raw.get("Name", ""),
        "type": raw.get("Type", ""),
        "media_type": raw.get("MediaType", "") or raw.get("Type", ""),
        "year": raw.get("ProductionYear"),
        "runtime_minutes": runtime_minutes,
        "watched": bool(user_data.get("Played", False)),
        "community_rating": raw.get("CommunityRating"),
        "genres": [g for g in (raw.get("Genres") or []) if g],
        "studios": [s.get("Name") for s in (raw.get("Studios") or []) if s.get("Name")],
        "tags": [t for t in (raw.get("Tags") or []) if t],
        "series_name": raw.get("SeriesName"),
        "directors": [p.get("Name") for p in people if p.get("Type") == "Director" and p.get("Name")],
        "actors": [p.get("Name") for p in people if p.get("Type") == "Actor" and p.get("Name")],
    }


async def find_all_collections(client) -> list[dict]:
    """Return every Emby collection (BoxSet), sorted by name.

    ``owner_name`` is the parsed ``"{Name} Favorites"`` owner (or ``None``), and
    ``is_favorites`` flags whether the collection follows that convention.
    ``[{"collection_id", "collection_name", "owner_name", "is_favorites", "item_count"}]``
    """
    data = await client.get_all(
        "/Items",
        {
            "IncludeItemTypes": "BoxSet",
            "Recursive": "true",
            "Fields": "ChildCount",
        },
    )
    collections = []
    for raw in data:
        owner = parse_favorites_owner(raw.get("Name"))
        collections.append(
            {
                "collection_id": raw.get("Id", ""),
                "collection_name": raw.get("Name", ""),
                "owner_name": owner,
                "is_favorites": owner is not None,
                "item_count": raw.get("ChildCount"),
            }
        )
    collections.sort(key=lambda c: (c["collection_name"] or "").casefold())
    return collections


async def find_favorites_collections(client) -> list[dict]:
    """Return only the detected ``"{Name} Favorites"`` collections."""
    return [c for c in await find_all_collections(client) if c["is_favorites"]]


async def find_user_favorites_collection(client, user_name: str) -> dict | None:
    """Find the ``"{user_name} Favorites"`` collection, or ``None``."""
    if not user_name:
        return None
    target = user_name.strip().casefold()
    for collection in await find_favorites_collections(client):
        if collection["owner_name"].casefold() == target:
            return collection
    return None


async def list_collection_items(client, collection_id: str, user_id: str) -> list[dict]:
    """List a collection's items with watched status resolved for ``user_id``.

    Collection children are retrieved via ``ParentId``. Passing the user scope
    makes Emby populate ``UserData.Played`` for that specific user, which is what
    makes the watched flag user-specific.
    """
    data = await client.get(
        f"/Users/{user_id}/Items",
        {
            "ParentId": collection_id,
            "Fields": ITEM_FIELDS,
            "EnableUserData": "true",
        },
    )
    items = data.get("Items", []) if isinstance(data, dict) else (data or [])
    return [normalize_item(raw) for raw in items]


async def get_collection_item_count(client, collection_id: str) -> int:
    """Ask Emby for the authoritative child count for one collection."""
    data = await client.get(
        "/Items",
        {
            "ParentId": collection_id,
            "Limit": "1",
            "Recursive": "false",
        },
    )
    if isinstance(data, dict):
        try:
            return int(data.get("TotalRecordCount", len(data.get("Items") or [])))
        except (TypeError, ValueError):
            return len(data.get("Items") or [])
    return len(data or [])


async def add_collection_items(client, collection_id: str, item_ids: list[str]):
    """Add items to a collection by id. No-op for an empty list."""
    if not item_ids:
        return None
    return await client.post(
        f"/Collections/{collection_id}/Items",
        {"Ids": ",".join(item_ids)},
    )


async def remove_collection_items(client, collection_id: str, item_ids: list[str]):
    """Remove items from a collection by id. No-op when empty."""
    if not item_ids:
        return None
    return await client.delete(
        f"/Collections/{collection_id}/Items",
        {"Ids": ",".join(item_ids)},
    )