feat(kexp): download kexp playlist from rest api

.gitignore

@@ -0,0 +1,175 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# UV
+#   Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#uv.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/latest/usage/project/#working-with-version-control
+.pdm.toml
+.pdm-python
+.pdm-build/
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+
+# Ruff stuff:
+.ruff_cache/
+
+# PyPI configuration file
+.pypirc
+

pyproject.toml

@@ -4,15 +4,24 @@ version = "0.1.0"
 description = "music utilies"
 readme = "README.md"
 authors = [{ name = "publicmatt", email = "git@publicmatt.com" }]
-requires-python = ">=3.13"
+requires-python = ">=3.12"
 dependencies = [
+    "audioop-lts>=0.2.2 ; python_full_version >= '3.13'",
     "click>=8.1.8",
-    "joblib>=1.4.2",
+    "duckdb>=1.1.0",
-    "python-dotenv>=1.0.1",
+    "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"]

requirements.txt

@@ -1,7 +0,0 @@
-asyncio
-shazamio
-click
-python_dotenv
-tqdm
-joblib
-mutagen

src/music/__init__.py

@@ -1,2 +0,0 @@
-def main() -> None:
-    print("Hello from music!")

src/music/__main__.py

@@ -1,53 +1,15 @@
-from enum import Enum
+from pydantic_settings import CliApp
-import click
-import subprocess
-# from .shazam import rename
 
-from click.exceptions import ClickException
+from .cli import Cli
+from .commands import Stream
 
 
-@click.group()
 def cli():
-    pass
+    CliApp.run(Cli)
 
 
-class Stream(str, Enum):
+def stream():
-    KEXP = "kexp"
+    CliApp.run(Stream)
-    KUGS = "kugs"
-    CLIS = "clis"
-
-
-@cli.command("stream")
-@click.option(
-    "-s", "--stream", "stream", type=click.Choice(Stream), default=Stream.KEXP
-)
-def play_stream(stream):
-    """Play the KEXP stream using mpv."""
-    match stream:
-        case Stream.KEXP:
-            url = "https://kexp-mp3-128.streamguys1.com/kexp128.mp3"
-        case Stream.KUGS:
-            url = "https://peridot.streamguys1.com:7175/kugs-mp3""
-        case Stream.CLIS:
-            url = "https://stream2.statsradio.com:8012/stream?=&&___cb=759135934160766"
-        case _:
-            raise ClickException(f"unrecognized stream: {stream}")
-    subprocess.run(["mpv", url])
-
-@cli.command("random")
-def play_random():
-  """todo."""
-    raise NotImplementedError("todo")
-    # fd -t file -e mp3 -p -a "fall|winter|spring|summer" $HOME/Music/ | mpv --playlist=- --shuffle --no-video
-
-@cli.command("save")
-@click.option( "-u", "--url", "url", required=True)
-def save_song(url):
-  """todo."""
-    raise NotImplementedError("todo")
-
-    # FILENAME=$(yt-dlp {{url}} -x --audio-format mp3 | tee /dev/tty | grep 'ExtractAudio' | cut -d ' ' -f 3-)
-    # $HOME/.local/lib/shazam/bin/shazam.py rename --song "$FILENAME"
 
 
 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

@@ -0,0 +1,53 @@
+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:
+        for_date = datetime.now()
+    year = for_date.year
+    month = for_date.month
+    if month in [12, 1, 2]:
+        quarter = "winter"
+    elif month in [3, 4, 5]:
+        quarter = "spring"
+    elif month in [6, 7, 8]:
+        quarter = "summer"
+    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/paths.py

@@ -0,0 +1,21 @@
+from pathlib import Path
+from typing import Generator
+
+from music.dates import quarter_year
+
+
+def seasons() -> Generator[Path, None, None]:
+    prefix = ("fall_", "winter_", "spring_", "summer_")
+    music_dir = Path().home() / "Music"
+    for d in music_dir.iterdir():
+        if not d.is_dir():
+            continue
+        if not d.name.startswith(prefix):
+            continue
+        yield d
+    yield from []
+
+
+def current_quarter() -> Path:
+    quarter, year = quarter_year()
+    return Path().home() / "Music" / f"{quarter}_{year}"

src/music/save.py

@@ -0,0 +1,48 @@
+from mutagen.easyid3 import EasyID3
+import yt_dlp
+from .dates import quarter_year
+from .paths import current_quarter
+import os
+
+from pathlib import Path
+
+
+def yt(url):
+    ydl_opts = {
+        "format": "mp3/bestaudio/best",
+        # ℹ️ See help(yt_dlp.postprocessor) for a list of available Postprocessors and their arguments
+        "postprocessors": [
+            {  # Extract audio using ffmpeg
+                "key": "FFmpegExtractAudio",
+                "preferredcodec": "mp3",
+            }
+        ],
+    }
+
+    os.chdir(Path.home() / "Downloads")
+    # Download the audio and get the filename
+    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+        info_dict = ydl.extract_info(url, download=True)
+        _ = ydl.prepare_filename(info_dict)
+    if info_dict is None:
+        raise ValueError("error downloading")
+    filename = info_dict["requested_downloads"][0]["filepath"]
+
+    track = info_dict.get("track")
+    artist = info_dict.get("artist")
+
+    parent = current_quarter()
+    if not parent.exists():
+        parent.mkdir(parents=True)
+    if track is not None and artist is not None:
+        name = f"{artist} - {track}.mp3"
+        new_filepath = parent / name
+        os.rename(filename, new_filepath)
+        tags = EasyID3(new_filepath)
+        tags["title"] = track
+        tags["artist"] = artist
+        tags.save()
+
+    else:
+        new_filepath = parent / filename
+        os.rename(filename, new_filepath)

src/music/shazam/shazam.py

@@ -1,137 +0,0 @@
-#! /home/user/.venv/bin/python
-
-import asyncio
-from shazamio import Shazam, Serialize
-from pathlib import Path
-import click
-from dotenv import load_dotenv
-import os
-from tqdm import tqdm
-from typing import Optional
-import sys
-from collections import namedtuple
-from joblib import Memory
-from pydantic import BaseModel, Field
-
-base_path = Path(__file__).parent.parent
-memory = Memory(base_path / "./.cache", verbose=0)
-
-
-class Match(BaseModel):
-    old: Path
-    new: Optional[Path] = Field(default=None)
-    title: Optional[str] = Field(default=None)
-    artist: Optional[str] = Field(default=None)
-
-
-@click.group()
-def cli():
-    pass
-
-
-@cli.command()
-@click.option(
-    "-s",
-    "--song",
-    required=True,
-    type=click.Path(exists=True, dir_okay=False, path_type=Path),
-)
-def rename(song: Path):
-    change = get_match(song)
-    if change and change.new is not None:
-        prompt_rename(change)
-
-
-def songs(root: Path):
-    seasons = ["winter", "spring", "summer", "fall"]
-    pattern = "{}_*/*.mp3"
-    matches = []
-    for season in seasons:
-        for file_path in root.rglob(pattern.format(season)):
-            matches.append(file_path)
-
-    def small(f: Path, max_bytes: int = 10_000_000):
-        return f.stat().st_size < max_bytes
-
-    return filter(small, matches)
-
-
-def needs_rename(song: Match):
-    if song.new is None:
-        print(f"\nno match: {song.old.name}\n")
-        return False
-    if song.old.name == song.new.name:
-        print(f"\nname correct: {song.old.name}\n")
-        return False
-    return True
-
-
-@cli.command("main")
-def main():
-    root = Path(os.getenv("MUSIC_PATH", "~/Music"))
-    if not root.exists():
-        raise ValueError(f"{str(root)} does not exist")
-
-    renames = []
-    for song in tqdm(songs(root), position=0, leave=True):
-        rename = get_match(song)
-        if rename and rename.new is not None:
-            renames.append(rename)
-
-    for song in filter(needs_rename, renames):
-        prompt_rename(song)
-
-    sys.exit(127)
-
-
-def prompt_rename(song: Match):
-    old = song.old
-    answer = input(
-        f'\nrename: "{old.name}"\nto: "{song.new.name}"?\n[y[es]/n[o]/m[anual]] > '
-    )
-    if answer in ["n", "N"]:
-        print(f"\nskipping: {old.name}\n")
-        return
-    if answer in ["m"]:
-        manual = input("\nnew name: ")
-        new = old.parent / manual
-        assert (
-            new.suffix == old.suffix
-        ), f"filetype changed: {old.suffix} -> {new.suffix}"
-    else:
-        new = song.new
-    try:
-        old.rename(new)
-        from mutagen.easyid3 import EasyID3
-
-        tags = EasyID3(new)
-        tags["title"] = song.title
-        tags["artist"] = song.artist
-        tags.save()
-    except FileNotFoundError as e:
-        print(f"error renaming: {e}")
-    print(f'old: "{old}"\nnew: "{new}"')
-    return
-
-
-@memory.cache
-def get_match(original: Path) -> Match:
-    loop = asyncio.get_event_loop()
-    shazam = Shazam()
-    try:
-        out = loop.run_until_complete(shazam.recognize(str(original)))
-        serialized = Serialize.track(out["track"])
-        rename = f"{serialized.title} - {serialized.subtitle}{original.suffix}"
-        new = Path(original.parent) / rename
-        return Match(
-            old=original, new=new, title=serialized.title, artist=serialized.subtitle
-        )
-    except:
-        return Match(old=original, new=None)
-
-
-if __name__ == "__main__":
-    env = Path(__file__).parent.parent / ".env"
-    if not load_dotenv(env):
-        raise ValueError(".env not found: {env}")
-    cli()

src/music/stream.py

@@ -0,0 +1,80 @@
+import subprocess
+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"
+    KUGS = "kugs"
+    CLIS = "clis"
+
+
+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
+
+
+def play(stream, t=None):
+    """Play streams using mpv."""
+    match stream:
+        case Stream.KEXP:
+            url = "https://kexp-mp3-128.streamguys1.com/kexp128.mp3"
+            if t:
+                url = url_for_time(t)
+        case Stream.KUGS:
+            url = "https://peridot.streamguys1.com:7175/kugs-mp3"
+        case Stream.CLIS:
+            url = "https://stream2.statsradio.com:8012/stream?=&&___cb=759135934160766"
+        case _:
+            raise ValueError(f"unrecognized stream: {stream}")
+    subprocess.run(["mpv", url])
+
+
+def local():
+    """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)