mcma-backend/app/api/rest/serializers.py

"""Entity → Subsonic child-dict mappers (presentation only).

Pure functions turning domain entities into the attribute dicts the envelope
serializer renders as ``<artist>`` / ``<album>`` / ``<song>`` elements (or their
JSON equivalents). No business logic — they only reshape and rename.
"""

import datetime as dt
import re
from typing import Any

from app.api.rest.ids import encode_album, encode_artist, encode_track
from app.domain.entities import Album, Artist, Track
from app.domain.entities.lyrics import Lyrics

# One LRC timecode: ``[mm:ss.xx]`` / ``[mm:ss.xxx]`` (fraction optional). A line
# may carry several (the same words repeat at multiple times); metadata tags like
# ``[ar:..]`` don't match, so they're ignored.
_LRC_TAG_RE = re.compile(r"\[(\d+):(\d{1,2})(?:[.:](\d{1,3}))?\]")

# Suffix → MIME, for the ``contentType``/``suffix`` song attributes. A
# presentation detail (mirrors StreamingService's content-type negotiation).
_CONTENT_TYPE: dict[str, str] = {
    "mp3": "audio/mpeg",
    "flac": "audio/flac",
    "m4a": "audio/mp4",
    "aac": "audio/aac",
    "ogg": "audio/ogg",
    "opus": "audio/ogg",
    "wav": "audio/wav",
    "aiff": "audio/aiff",
    "aif": "audio/aiff",
}


def iso(value: dt.datetime) -> str:
    return value.astimezone(dt.UTC).strftime("%Y-%m-%dT%H:%M:%S.000Z")


def content_type_for(file_format: str) -> str:
    return _CONTENT_TYPE.get(file_format.lower(), "application/octet-stream")


def artist_dict(artist: Artist, *, album_count: int) -> dict[str, Any]:
    return {
        "id": encode_artist(artist.id),
        "name": artist.name,
        "albumCount": album_count,
        "coverArt": encode_artist(artist.id),
    }


def album_dict(
    album: Album,
    artist: Artist | None,
    *,
    song_count: int,
    duration: int | None = None,
) -> dict[str, Any]:
    return {
        "id": encode_album(album.id),
        "name": album.title,
        "title": album.title,
        "artist": artist.name if artist is not None else None,
        "artistId": encode_artist(album.artist_id),
        "coverArt": encode_album(album.id),
        "songCount": song_count,
        "duration": duration,
        "created": iso(album.created_at),
        "year": album.year,
    }


def song_dict(
    track: Track,
    artist: Artist | None,
    album: Album | None,
) -> dict[str, Any]:
    cover = encode_album(track.album_id) if track.album_id is not None else encode_track(track.id)
    return {
        "id": encode_track(track.id),
        "parent": encode_album(track.album_id) if track.album_id is not None else None,
        "isDir": False,
        "title": track.title,
        "album": album.title if album is not None else None,
        "artist": artist.name if artist is not None else None,
        "albumId": encode_album(track.album_id) if track.album_id is not None else None,
        "artistId": encode_artist(track.artist_id),
        "coverArt": cover,
        "size": track.file_size or 0,
        "contentType": content_type_for(track.file_format or ""),
        "suffix": track.file_format,
        "duration": track.duration_seconds,
        "year": track.year,
        "genre": track.genre,
        "created": iso(track.created_at),
        "type": "music",
        "isVideo": False,
    }


def _parse_lrc(synced: str) -> list[dict[str, Any]]:
    """LRC text → OpenSubsonic ``line`` dicts (``start`` in ms, ``value`` text),
    ordered by time. Lines with no timecode (blank lines, metadata tags) drop out;
    a timecode carrying several stamps yields one line per stamp."""
    lines: list[tuple[int, str]] = []
    for raw in synced.splitlines():
        stamps = list(_LRC_TAG_RE.finditer(raw))
        if not stamps:
            continue
        text = _LRC_TAG_RE.sub("", raw).strip()
        for m in stamps:
            minutes, seconds = int(m.group(1)), int(m.group(2))
            # LRC fractions are centiseconds (2 digits) or ms (3); pad to ms.
            ms = int((m.group(3) or "0").ljust(3, "0")[:3])
            lines.append(((minutes * 60 + seconds) * 1000 + ms, text))
    lines.sort(key=lambda pair: pair[0])
    return [{"start": start, "value": text} for start, text in lines]


def structured_lyrics(
    lyrics: Lyrics, *, display_artist: str, display_title: str
) -> dict[str, Any]:
    """OpenSubsonic ``getLyricsBySongId`` payload. Prefers synced (LRC) lines and
    falls back to plain text; an empty ``lyricsList`` when the track has none."""
    lines: list[dict[str, Any]] = []
    synced = False
    if lyrics.synced:
        lines = _parse_lrc(lyrics.synced)
        synced = bool(lines)
    if not lines and lyrics.plain:
        lines = [{"value": line} for line in lyrics.plain.splitlines()]
    if not lines:
        return {"lyricsList": {}}
    return {
        "lyricsList": {
            "structuredLyrics": [
                {
                    "displayArtist": display_artist,
                    "displayTitle": display_title,
                    "lang": "xxx",
                    "offset": 0,
                    "synced": synced,
                    "line": lines,
                }
            ]
        }
    }