asot_player/asot_player/feed.py

"""Fetch and parse the miroppb.com/ASOT RSS feed."""

from __future__ import annotations

import re
import urllib.request
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from datetime import datetime
from email.utils import parsedate_to_datetime

BASE = "https://miroppb.com"
FEED_URL = BASE + "/rss"
USER_AGENT = (
    "asot-player/0.1 (a terminal ASOT player; reads episodes for playback) "
    "https://github.com/asot-player"
)

_EP_RE = re.compile(r"Episode\s+(\d+)", re.IGNORECASE)


@dataclass
class Episode:
    number: int
    url: str
    pubdate: datetime | None = None
    title: str = ""

    @property
    def date_str(self) -> str:
        return self.pubdate.strftime("%d %b %Y") if self.pubdate else ""


def _ua_request(url: str, headers: dict[str, str] | None = None):
    req = urllib.request.Request(
        url,
        headers={**{"User-Agent": USER_AGENT}, **(headers or {})},
        method="GET",
    )
    try:
        with urllib.request.urlopen(req, timeout=20) as r:
            return r.status, r.headers, r.read()
    except Exception as e:  # noqa: BLE001
        raise ConnectionError(f"fetch failed: {url} ({e})") from e


def _parse_date(raw: str) -> datetime | None:
    if not raw:
        return None
    try:
        return parsedate_to_datetime(raw.strip())
    except Exception:
        return None


def fetch_episodes(limit: int | None = None) -> list[Episode]:
    """Return episodes newest-first from the miroppb RSS feed."""
    _, _, body = _ua_request(FEED_URL)
    root = ET.fromstring(body)
    eps: list[Episode] = []
    for item in root.iter("item"):
        title = item.findtext("title", "") or ""
        m = _EP_RE.search(title)
        if not m:
            continue
        num = int(m.group(1))
        enc = item.find("enclosure")
        url = (enc.get("url") if enc is not None else "") or f"{BASE}/ASOT/{num}"
        num_from_url = re.search(r"ASOT_Ep_(\d+)\.mp3$", url)
        if num_from_url:
            num = int(num_from_url.group(1))
        eps.append(
            Episode(
                number=num,
                url=url,
                pubdate=_parse_date(item.findtext("pubDate", "")),
                title=title,
            )
        )
    # de-dupe on number (feed can repeat items after a rebuild)
    seen: dict[int, Episode] = {}
    for ep in eps:
        if ep.number not in seen:
            seen[ep.number] = ep
    out = sorted(seen.values(), key=lambda e: -e.number)
    return out[:limit] if limit else out


def episode_by_number(episodes: list[Episode], number: int) -> Episode | None:
    for e in episodes:
        if e.number == number:
            return e
    return None


def latest_number(episodes: list[Episode]) -> int:
    return episodes[0].number