feat(kexp): download kexp playlist from rest api

pyproject.toml

@@ -6,15 +6,22 @@ readme = "README.md"
 authors = [{ name = "publicmatt", email = "git@publicmatt.com" }]
 requires-python = ">=3.12"
 dependencies = [
+    "audioop-lts>=0.2.2 ; python_full_version >= '3.13'",
     "click>=8.1.8",
+    "duckdb>=1.1.0",
     "mpv>=1.0.7",
     "mutagen>=1.47.0",
+    "pydantic>=2.0",
+    "pydantic-settings>=2.12.0",
     "requests>=2.32.3",
+    "shazamio>=0.8.1",
+    "structlog>=25.5.0",
     "yt-dlp>=2025.1.15",
 ]
 
 [project.scripts]
 music = "music.__main__:cli"
+stream = "music.__main__:stream"
 
 [build-system]
 requires = ["hatchling"]

src/music/__main__.py

@@ -1,53 +1,15 @@
-from datetime import datetime, timezone
-from typing import Optional
+from pydantic_settings import CliApp
 
-import click
-
-from .save import yt
-from .stream import Stream, local, play
+from .cli import Cli
+from .commands import Stream
 
 
-@click.group()
 def cli():
-    pass
+    CliApp.run(Cli)
 
 
-@cli.command("stream")
-@click.argument(
-    "stream", type=click.Choice([e.value for e in Stream]), default=Stream.KEXP
-)
-@click.option(
-    "-t",
-    "--timestamp",
-    "t",
-    type=click.DateTime(formats=["%Y-%m-%d %H:%M"]),
-    required=False,
-)
-def _stream(stream, t: Optional[datetime]):
-    """play streams."""
-
-    if t is not None:
-        time = t.astimezone(tz=timezone.utc)
-    else:
-        time = None
-    play(stream, time)
-
-
-@cli.command("random")
-def _random():
-    """play randomly from local music."""
-    local()
-
-
-@cli.command("yt")
-@click.argument("url", required=True)
-def _save(url):
-    """
-    download a youtube song from URL to current quarter dir.
-
-    URL: youtube url to download
-    """
-    yt(url)
+def stream():
+    CliApp.run(Stream)
 
 
 if __name__ == "__main__":

src/music/cli.py

@@ -0,0 +1,26 @@
+from pydantic import Field
+from pydantic_settings import BaseSettings, CliApp, CliSubCommand, get_subcommand
+
+from music.commands import Kexp, Playlist, Random, Stream, YtDownload
+
+
+class Cli(BaseSettings):
+    model_config = {
+        "env_file": [".env"],
+        "cli_kebab_case": True,
+        "cli_use_class_docs_for_groups": True,
+    }
+
+    download: CliSubCommand[YtDownload] = Field(alias="yt")
+    stream: CliSubCommand[Stream] = Field(alias="stream")
+    random: CliSubCommand[Random] = Field(alias="random")
+    kexp: CliSubCommand[Kexp] = Field(alias="kexp")
+    playlist: CliSubCommand[Playlist] = Field(alias="playlist")
+
+    def cli_cmd(self) -> None:
+        if (cmd := get_subcommand(self, is_required=False)) is not None:
+            CliApp.run_subcommand(self, cli_cmd_method_name="run")
+        else:
+            CliApp.run(Cli, cli_args=["--help"])
+
+    pass

src/music/commands/__init__.py

@@ -0,0 +1,7 @@
+from .download import YtDownload
+from .kexp import Kexp
+from .playlist import Playlist
+from .random import Random
+from .stream import Stream
+
+__all__ = ["YtDownload", "Kexp", "Playlist", "Random", "Stream"]

src/music/commands/download.py

@@ -0,0 +1,143 @@
+import asyncio
+import os
+import re
+from pathlib import Path
+from typing import ClassVar
+
+from pydantic_settings import CliImplicitFlag, CliPositionalArg
+import structlog
+import yt_dlp
+from mutagen.easyid3 import EasyID3
+from pydantic import Field, computed_field
+from shazamio import Shazam
+
+from music.dates import DateParse
+from music.models import Command
+
+log = structlog.get_logger()
+
+
+def _clean(value) -> str | None:
+    if value is None:
+        return None
+    if isinstance(value, list):
+        value = ", ".join(v for v in value if v)
+    value = str(value).strip()
+    return value or None
+
+
+def _prompt(label: str, default: str | None = None) -> str | None:
+    suffix = f" [{default}]" if default else ""
+    try:
+        value = input(f"{label}{suffix}: ").strip()
+    except EOFError:
+        return default
+    return value or default
+
+
+def _extract_artist_track(info: dict) -> tuple[str | None, str | None]:
+    artist = (
+        _clean(info.get("artists"))
+        or _clean(info.get("artist"))
+        or _clean(info.get("creators"))
+        or _clean(info.get("creator"))
+    )
+    track = _clean(info.get("track")) or _clean(info.get("alt_title"))
+
+    title = _clean(info.get("title"))
+    if (artist is None or track is None) and title and " - " in title:
+        parsed_artist, parsed_track = title.split(" - ", 1)
+        parsed_track = re.sub(r"\s*[\(\[][^\)\]]*[\)\]]\s*$", "", parsed_track).strip()
+        if artist is None:
+            artist = _clean(parsed_artist)
+        if track is None:
+            track = _clean(parsed_track)
+    return artist, track
+
+
+def _shazam_lookup(filepath: str) -> tuple[str | None, str | None]:
+    async def recognize() -> dict:
+        return await Shazam().recognize(filepath)
+
+    try:
+        result = asyncio.run(recognize())
+    except Exception as e:
+        log.warning("shazam lookup failed", error=str(e))
+        return None, None
+    track_info = result.get("track") or {}
+    return _clean(track_info.get("subtitle")), _clean(track_info.get("title"))
+
+
+def current_quarter() -> Path:
+    date = DateParse()
+    return Path().home() / "Music" / f"{date.quarter}_{date.year}"
+
+
+class YtDownload(Command):
+    """
+    download a youtube song from {URL} to current quarter dir.
+    """
+
+    url: CliPositionalArg[str] = Field(description="youtube url to download")
+    ydl_opts: ClassVar = {
+        "format": "mp3/bestaudio/best",
+        "postprocessors": [
+            {
+                "key": "FFmpegExtractAudio",
+                "preferredcodec": "mp3",
+            }
+        ],
+        "remote_components": ["ejs:github"],
+    }
+    parents: CliImplicitFlag[bool] = Field(
+        default=True, description="create the parent dirs if not exists"
+    )
+
+    @computed_field()
+    @property
+    def quarter_dir(self) -> Path:
+        date = DateParse()
+        return Path().home() / "Music" / f"{date.quarter}_{date.year}"
+
+    def run(self) -> None:
+        os.chdir(Path.home() / "Downloads")
+        with yt_dlp.YoutubeDL(self.ydl_opts) as ydl:
+            info_dict = ydl.extract_info(self.url, download=True)
+            _ = ydl.prepare_filename(info_dict)
+        if info_dict is None:
+            raise ValueError("error downloading")
+        filename = info_dict["requested_downloads"][0]["filepath"]
+
+        artist, track = _extract_artist_track(info_dict)
+
+        if artist is None or track is None:
+            log.info("trying shazam", file=filename)
+            shazam_artist, shazam_track = _shazam_lookup(filename)
+            artist = artist or shazam_artist
+            track = track or shazam_track
+            if shazam_artist or shazam_track:
+                log.info("shazam matched", artist=shazam_artist, track=shazam_track)
+
+        if artist is None or track is None:
+            title = info_dict.get("title") or Path(filename).stem
+            uploader = info_dict.get("uploader") or info_dict.get("channel")
+            log.info("could not parse metadata", title=title, uploader=uploader)
+            artist = _prompt("artist", default=artist or uploader)
+            track = _prompt("track", default=track or title)
+
+        base = self.quarter_dir
+        if not base.exists():
+            base.mkdir(parents=self.parents)
+        if track and artist:
+            name = f"{artist} - {track}.mp3"
+            new_filepath = base / name
+            os.rename(filename, new_filepath)
+            tags = EasyID3(new_filepath)
+            tags["title"] = track
+            tags["artist"] = artist
+            tags.save()
+        else:
+            new_filepath = base / Path(filename).name
+            os.rename(filename, new_filepath)
+
+        log.info("downloaded", path=str(new_filepath))

src/music/commands/kexp.py

@@ -0,0 +1,225 @@
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+import duckdb
+import requests
+import structlog
+from pydantic import Field
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+
+from music.models import Command
+
+log = structlog.get_logger()
+
+API_URL = "https://api.kexp.org/v1/play/"
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS plays (
+    playid             BIGINT PRIMARY KEY,
+    playtype_id        INTEGER,
+    playtype_name      VARCHAR,
+    airdate            TIMESTAMP,
+    epoch_airdate      BIGINT,
+    artist_id          BIGINT,
+    artist_name        VARCHAR,
+    artist_islocal     BOOLEAN,
+    release_id         BIGINT,
+    release_name       VARCHAR,
+    release_image      VARCHAR,
+    releaseevent_id    BIGINT,
+    releaseevent_year  INTEGER,
+    track_id           BIGINT,
+    track_name         VARCHAR,
+    label_id           BIGINT,
+    label_name         VARCHAR,
+    showid             BIGINT,
+    comments           VARCHAR,
+    fetched_at         TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);
+"""
+
+UPSERT = """
+INSERT INTO plays (
+    playid, playtype_id, playtype_name, airdate, epoch_airdate,
+    artist_id, artist_name, artist_islocal,
+    release_id, release_name, release_image,
+    releaseevent_id, releaseevent_year,
+    track_id, track_name,
+    label_id, label_name,
+    showid, comments
+) VALUES (
+    ?, ?, ?, ?, ?,
+    ?, ?, ?,
+    ?, ?, ?,
+    ?, ?,
+    ?, ?,
+    ?, ?,
+    ?, ?
+)
+ON CONFLICT (playid) DO NOTHING;
+"""
+
+
+def _get(d: dict | None, key: str) -> Any:
+    return d.get(key) if isinstance(d, dict) else None
+
+
+def _flatten(play: dict) -> tuple:
+    playtype = play.get("playtype") or {}
+    artist = play.get("artist") or {}
+    release = play.get("release") or {}
+    revent = play.get("releaseevent") or {}
+    track = play.get("track") or {}
+    label = play.get("label") or {}
+    comments = play.get("comments") or []
+    comment_text = " | ".join(c.get("text", "") for c in comments if c.get("text"))
+
+    airdate = play.get("airdate")
+    if airdate:
+        airdate = datetime.fromisoformat(airdate.replace("Z", "+00:00"))
+
+    return (
+        play.get("playid"),
+        _get(playtype, "playtypeid"),
+        _get(playtype, "name"),
+        airdate,
+        play.get("epoch_airdate"),
+        _get(artist, "artistid"),
+        _get(artist, "name"),
+        _get(artist, "islocal"),
+        _get(release, "releaseid"),
+        _get(release, "name"),
+        _get(release, "largeimageuri") or _get(release, "smallimageuri"),
+        _get(revent, "releaseeventid"),
+        _get(revent, "year"),
+        _get(track, "trackid"),
+        _get(track, "name"),
+        _get(label, "labelid"),
+        _get(label, "name"),
+        play.get("showid"),
+        comment_text or None,
+    )
+
+
+def _make_session() -> requests.Session:
+    session = requests.Session()
+    retry = Retry(
+        total=5,
+        connect=5,
+        read=5,
+        status=5,
+        backoff_factor=1.5,
+        status_forcelist=(429, 500, 502, 503, 504),
+        allowed_methods=("GET",),
+        raise_on_status=False,
+    )
+    session.mount("https://", HTTPAdapter(max_retries=retry))
+    session.mount("http://", HTTPAdapter(max_retries=retry))
+    return session
+
+
+def _fetch_page(session: requests.Session, offset: int, limit: int) -> list[dict]:
+    resp = session.get(
+        API_URL, params={"limit": limit, "offset": offset}, timeout=60
+    )
+    resp.raise_for_status()
+    return resp.json().get("results") or []
+
+
+class Kexp(Command):
+    """batch the KEXP play API into a duckdb table."""
+
+    db: Path = Field(
+        default=Path.home() / "Music" / "kexp.duckdb",
+        description="duckdb file path",
+    )
+    limit: int = Field(default=200, description="page size per request")
+    concurrency: int = Field(default=5, description="parallel HTTP workers")
+    batch_size: int = Field(
+        default=2000, description="rows per DB insert flush"
+    )
+    max_pages: int | None = Field(
+        default=None, description="stop after N pages (default: until exhausted)"
+    )
+    from_page: int = Field(
+        default=0, description="start pagination at this page number"
+    )
+    stop_on_existing: bool = Field(
+        default=False,
+        description="stop scheduling once a chunk yields zero new rows (use for incremental top-up)",
+    )
+
+    def _flush(self, con: duckdb.DuckDBPyConnection, buffer: list[tuple]) -> int:
+        if not buffer:
+            return 0
+        ids = [r[0] for r in buffer]
+        existing = {
+            row[0]
+            for row in con.execute(
+                f"SELECT playid FROM plays WHERE playid IN ({','.join('?' * len(ids))})",
+                ids,
+            ).fetchall()
+        }
+        new_rows = [r for r in buffer if r[0] not in existing]
+        if new_rows:
+            con.executemany(UPSERT, new_rows)
+        buffer.clear()
+        return len(new_rows)
+
+    def run(self) -> None:
+        self.db.parent.mkdir(parents=True, exist_ok=True)
+        con = duckdb.connect(str(self.db))
+        con.execute(SCHEMA)
+
+        total_inserted = 0
+        buffer: list[tuple] = []
+        page = self.from_page
+        end_page = (
+            self.from_page + self.max_pages if self.max_pages is not None else None
+        )
+        stop = False
+        session = _make_session()
+
+        with ThreadPoolExecutor(max_workers=self.concurrency) as pool:
+            while not stop:
+                chunk = []
+                for _ in range(self.concurrency):
+                    if end_page is not None and page >= end_page:
+                        break
+                    chunk.append((page, page * self.limit))
+                    page += 1
+                if not chunk:
+                    break
+
+                offsets = [off for _, off in chunk]
+                log.info("fetching chunk", pages=[p for p, _ in chunk])
+                results = list(
+                    pool.map(lambda o: _fetch_page(session, o, self.limit), offsets)
+                )
+
+                chunk_new = 0
+                empty_page = False
+                for results_page in results:
+                    if not results_page:
+                        empty_page = True
+                        continue
+                    buffer.extend(_flatten(p) for p in results_page)
+                    if len(buffer) >= self.batch_size:
+                        chunk_new += self._flush(con, buffer)
+
+                chunk_new += self._flush(con, buffer)
+                total_inserted += chunk_new
+                log.info("chunk done", new=chunk_new, total=total_inserted)
+
+                if empty_page:
+                    stop = True
+                elif self.stop_on_existing and chunk_new == 0:
+                    log.info("no new rows in chunk, stopping")
+                    stop = True
+
+        count = con.execute("SELECT COUNT(*) FROM plays").fetchone()[0]
+        log.info("done", inserted=total_inserted, table_rows=count, db=str(self.db))
+        con.close()

src/music/commands/playlist.py

@@ -0,0 +1,37 @@
+from pathlib import Path
+
+import structlog
+from pydantic import Field
+from pydantic_settings import CliImplicitFlag
+
+from music.models import Command
+from music.paths import seasons
+
+log = structlog.get_logger()
+
+EXTENSIONS = {".mp3", ".flac", ".m4a", ".ogg", ".opus", ".wav"}
+
+
+class Playlist(Command):
+    """update .m3u8 playlist in every quarterly dir."""
+
+    force: CliImplicitFlag[bool] = Field(
+        default=True, description="overwrite existing playlist files."
+    )
+
+    def run(self) -> None:
+        for d in sorted(seasons()):
+            songs = sorted(f for f in d.iterdir() if f.suffix in EXTENSIONS)
+            if not songs:
+                log.info("skipping empty dir", dir=d.name)
+                continue
+
+            playlist = d / f"{d.name}.m3u8"
+            if playlist.exists() and not self.force:
+                log.info("skipping existing playlist", path=str(playlist))
+                continue
+
+            playlist.write_text(
+                "\n".join(f.name for f in songs) + "\n", encoding="utf-8"
+            )
+            log.info("wrote playlist", path=str(playlist), tracks=len(songs))

src/music/commands/random.py

@@ -0,0 +1,35 @@
+import subprocess
+from music.models import Command
+from music.paths import seasons
+import random
+
+# import mpv
+import tempfile
+import os
+
+
+class Random(Command):
+    def run(self):
+        """play random local songs."""
+        songs = []
+        for d in seasons():
+            for f in d.iterdir():
+                if not f.suffix == ".mp3":
+                    continue
+                songs.append(f)
+        random.shuffle(songs)
+        with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as tmp:
+            for song in songs:
+                tmp.write(f"{song}\n".encode())
+            playlist = tmp.name
+            # fd -t file -e mp3 -p -a "fall|winter|spring|summer" $HOME/Music/ | mpv --playlist=- --shuffle --no-video
+
+        try:
+            subprocess.run(["mpv", f"--playlist={playlist}", "--shuffle", "--no-video"])
+            # process = subprocess.Popen(
+            #     ["mpv", f"--playlist={playlist}", "--shuffle", "--no-video"],
+            # )
+
+            # process.wait()
+        finally:
+            os.remove(playlist)

src/music/commands/stream.py

@@ -0,0 +1,92 @@
+from datetime import datetime, timezone
+from enum import Enum
+from pathlib import Path
+import subprocess
+
+from pydantic import Field
+import requests
+import structlog
+
+from music.dates import DateParse
+from music.models import Command
+
+log = structlog.get_logger()
+
+
+class Source(str, Enum):
+    KEXP = "kexp"
+    KUGS = "kugs"
+    CLIS = "clis"
+
+
+class Show(str, Enum):
+    """
+    TODO: map these to the most recent.
+    TODO: write a parser or use an llm to map english to timedelta
+    --query "last morning show"
+    --query "last friday show"
+    --query "today's midday show"
+    --query "recent roadhouse"
+    """
+
+    FRIDAY_MORNING = ["friday", "friday morning"]
+    MORNING = ["morning show", "morning"]
+    # roadhouse
+    # midday
+    # afternoon
+    # drivetime
+    # el sonito
+
+
+def current_quarter() -> Path:
+    date = DateParse()
+    return Path().home() / "Music" / f"{date.quarter}_{date.year}"
+
+
+def url_for_time(t) -> str:
+    default = "https://kexp-mp3-128.streamguys1.com/kexp128.mp3"
+    tz = t.strftime("%Y-%m-%dT%H:%M:%SZ")
+    log.info("getting archive", time=t)
+    check_url = (
+        f"https://api.kexp.org/get_streaming_url/?bitrate=128&timestamp={tz}&location=1"
+    )
+    response = requests.get(check_url)
+    if response.status_code == 200:
+        found = response.json().get("sg-url")
+        if found is None:
+            log.error("error getting archive", time=t)
+            return default
+    else:
+        log.error("error getting archive", time=t, status=response.status_code)
+        return default
+
+    return found
+
+
+class Stream(Command):
+    """play streams."""
+
+    source: Source = Field(default=Source.KEXP)
+    for_date: datetime | None = Field(default=None)
+
+    def run(self) -> None:
+        if self.for_date is not None:
+            time = self.for_date.astimezone(tz=timezone.utc)
+        else:
+            time = None
+        self.play(self.source, time)
+
+    def play(self, stream, t=None):
+        """Play streams using mpv."""
+        match stream:
+            case Source.KEXP:
+                url = "https://kexp-mp3-128.streamguys1.com/kexp128.mp3"
+                if t:
+                    url = url_for_time(t)
+            case Source.KUGS:
+                url = "https://peridot.streamguys1.com:7175/kugs-mp3"
+            case Source.CLIS:
+                url = "https://stream2.statsradio.com:8012/stream?=&&___cb=759135934160766"
+            case _:
+                raise ValueError(f"unrecognized stream: {stream}")
+        subprocess.run(["mpv", url])

src/music/dates.py

@@ -1,6 +1,11 @@
-from datetime import datetime
+from __future__ import annotations
+
+from datetime import date, datetime
+from pathlib import Path
 from typing import Optional, Tuple
 
+from pydantic import BaseModel, Field, computed_field
+
 
 def quarter_year(for_date: Optional[datetime] = None) -> Tuple[str, int]:
     if for_date is None:
@@ -16,3 +21,33 @@ def quarter_year(for_date: Optional[datetime] = None) -> Tuple[str, int]:
     else:
         quarter = "fall"
     return quarter, year
+
+
+class DateParse(BaseModel):
+    date: datetime = Field(default_factory=datetime.now)
+
+    @computed_field()
+    @property
+    def quarter(self) -> str:
+        match self.date.month:
+            case 12 | 1 | 2:
+                return "winter"
+            case 3 | 4 | 5:
+                return "spring"
+            case 6 | 7 | 8:
+                return "summer"
+            case 9 | 10 | 11:
+                return "fall"
+            case _:
+                raise ValueError(self.date.month)
+
+    @computed_field()
+    @property
+    def year(self) -> int:
+        return self.date.year
+
+    @staticmethod
+    def from_dir(path: Path) -> date | None:
+        # TODO: parse path.name into a date if its {winter|fall|..}_{yyyy}.
+        # TODO: None if can't parse
+        return None

src/music/models.py

@@ -0,0 +1,9 @@
+from abc import ABC, abstractmethod
+
+from pydantic_settings import BaseSettings
+
+
+class Command(BaseSettings, ABC):
+    @abstractmethod
+    def run(self) -> None:
+        pass

src/music/stream.py

@@ -3,11 +3,14 @@ from enum import Enum
 from .paths import seasons
 import random
 import requests
+import structlog
 
 # import mpv
 import tempfile
 import os
 
+log = structlog.get_logger()
+
 
 class Stream(str, Enum):
     KEXP = "kexp"
@@ -18,7 +21,7 @@ class Stream(str, Enum):
 def url_for_time(t) -> str:
     default = "https://kexp-mp3-128.streamguys1.com/kexp128.mp3"
     tz = t.strftime("%Y-%m-%dT%H:%M:%SZ")
-    print(f"getting archive: {t}")
+    log.info("getting archive", time=t)
     check_url = (
         f"https://api.kexp.org/get_streaming_url/?bitrate=128&timestamp={tz}&location=1"
     )
@@ -26,10 +29,10 @@ def url_for_time(t) -> str:
     if response.status_code == 200:
         found = response.json().get("sg-url")
         if found is None:
-            print(f"error getting archive: {t}")
+            log.error("error getting archive", time=t)
             return default
     else:
-        print(f"error getting archive: {t}")
+        log.error("error getting archive", time=t, status=response.status_code)
         return default
 
     return found