MIL-006: Step 4 - Search and add songs #48

Merged
Tirsvad merged 2 commits from mil-006-step-4-search-and-add-songs into main 2026-10-28 07:36:30 +01:00
16 changed files with 650 additions and 23 deletions
Showing only changes of commit 9f0e66d20a - Show all commits
+13 -3
View File
@@ -84,12 +84,12 @@ The program is added step by step (one pull request per lecture step):
| 1. Scrape the Billboard Hot 100 | done (below) | | 1. Scrape the Billboard Hot 100 | done (below) |
| 2. YouTube Music authentication | code done; live screenshot waits for `browser.json` | | 2. YouTube Music authentication | code done; live screenshot waits for `browser.json` |
| 3. Create the playlist | code done; live screenshot waits for `browser.json` | | 3. Create the playlist | code done; live screenshot waits for `browser.json` |
| 4. Search and add songs | planned in milestone 6 | | 4. Search and add songs | code done; live screenshot waits for `browser.json` |
### Step 1 - Scrape the Billboard Hot 100 ### Step 1 - Scrape the Billboard Hot 100
```bash ```bash
python -m musical_time_machine python -m musical_time_machine --list-only
``` ```
The program asks for a date in the format `YYYY-MM-DD` and prints the 100 songs of that chart as `rank. title - artist`. Use a date that Bakeboard has, for example `2026-04-18`. `scrape_titles` returns the titles as a Python list; a date in the wrong format or a chart that does not exist ends the program with a clear message. The program asks for a date in the format `YYYY-MM-DD` and prints the 100 songs of that chart as `rank. title - artist`. Use a date that Bakeboard has, for example `2026-04-18`. `scrape_titles` returns the titles as a Python list; a date in the wrong format or a chart that does not exist ends the program with a clear message.
@@ -128,7 +128,17 @@ The screenshot shows the tests of this step, which use a fake YouTube Music serv
![Tests of step 3: the playlist is created once](docs/wiki/images/step3-create-playlist.png) ![Tests of step 3: the playlist is created once](docs/wiki/images/step3-create-playlist.png)
The screenshots and the instructions of the next steps are added with each step. ### Step 4 - Search and add songs
```bash
python -m musical_time_machine
```
The full program: it asks for the date, scrapes the chart, creates the playlist (or reuses the existing one), then searches YouTube Music for each song (`yt.search(..., filter="songs")`) and adds the top result with `yt.add_playlist_items()`. A song that is not found or fails is skipped and the run goes on; the summary says how many songs were added and which were skipped. A failed login stops the run, because every following song would fail too. Use `--list-only` to print the chart without logging in (step 1).
The screenshot shows the tests of this step with a fake YouTube Music service. A screenshot of the live playlist is added once `browser.json` exists (issue 34 stays open until then).
![Tests of step 4: songs are added and a failing song is skipped](docs/wiki/images/step4-add-songs.png)
## 🧪 Tests ## 🧪 Tests
+1 -1
View File
@@ -17,7 +17,7 @@ document of a type. `Primary File` may contain a glob (e.g.
| MIL | Milestone / Gateway | docs/milestones/mil-*.md | 008 | | MIL | Milestone / Gateway | docs/milestones/mil-*.md | 008 |
| GOV | Governance / ARB Workflow | docs/sqa/governance.md | 002 | | GOV | Governance / ARB Workflow | docs/sqa/governance.md | 002 |
| TM | Traceability Matrix | docs/sqa/traceability-matrix.md | 002 | | TM | Traceability Matrix | docs/sqa/traceability-matrix.md | 002 |
| RC | SQA Review Record | docs/sqa/reviews/rc-*.md | 016 | | RC | SQA Review Record | docs/sqa/reviews/rc-*.md | 017 |
## Languages ## Languages
+61
View File
@@ -0,0 +1,61 @@
# Review of the code of MIL-006
## Metadata
| Key | Value |
| --- | --- |
| ID | RC-016 |
| CrossReference | [MIL-006], [QC-PY-001] |
## Version History
| Date | Status | Author | Reviewer | Change | Commit |
| --- | --- | --- | --- | --- | --- |
| 2026-10-28 | Proposed | Jens Tirsvad Nielsen | AI as stand-in for S01 | Initial version | pending |
---
## Artifact Under Review
- Instance reviewed: [MIL-006] (the code of its pull request, `src/` and `tests/`)
- Checklist used: [QC-PY-001] (`QC-PY-001`)
- Scope: full review
- Language and domain: n/a
- Language reviewer: none
- Mode: autonomous, not independent (reviewer is `AI as stand-in for S01`)
## Checklist Results
| # | Criterion | Status | Evidence/Notes |
| --- | --- | --- | --- |
| 1 | Packages, modules, functions, variables, classes and constants follow PEP 8 casing (`snake_case`, `PascalCase`, `UPPER_SNAKE`) | Pass | Casing follows PEP 8. |
| 2 | Names state purpose in the domain's language; no unexplained abbreviations, no single-letter names outside tiny scopes | Pass | Names state purpose (add_songs, time_travel, SongAddError). |
| 3 | Code is produced by the project's formatter and passes its linter with no unexplained suppressions | Pass | ruff format and ruff check report no finding; four `noqa: N803` are explained: parameter names must match ytmusicapi. |
| 4 | Every function and method signature is type-annotated, including `-> None` | Pass | All functions annotated including -> None. |
| 5 | No bare `except:`, no swallowed exceptions; specific exceptions are raised and the cause is kept (`raise ... from`) | Pass | Song errors are skipped by type, never by a bare except; authentication errors propagate; causes are kept with `from`. |
| 6 | No mutable default arguments and no shadowed builtins | Pass | Checked in the diff. |
| 7 | Files, locks and connections are managed with context managers | N-A | No files or connections are opened by the new code. |
| 8 | Public modules, classes and functions have docstrings that say what, not how | Pass | Checked in the diff. |
| 9 | Logging uses `logging`, not `print`; no secrets or personal data in log output | Pass | Output goes through the injected writer; nothing secret is logged. |
| 10 | Classes and operations trace to the Design Class Diagram they implement; deviations are recorded | N-A | No Design Class Diagram exists; all tasks are technical. |
| 11 | Tests exist for new behaviour, are named for the behaviour, and do not depend on order or the network | Pass | 92 tests, 99 % coverage, no network, no real sleep (pause is injected). |
| 12 | Type checker runs in strict mode without errors; `Any` is justified in a comment | N-A | mypy is not configured. Optional criterion. |
| 13 | Dependencies are declared and pinned in the project's dependency file, none unused | Pass | Checked in the diff. |
| 14 | Each piece of knowledge (a business rule, constant, format, validation or query) is defined in one place; code is merged only where it expresses the same knowledge, not where it merely looks alike | Pass | Messages, pause and status text are constants defined once. |
| 15 | Imports point inward: business-rule modules import no framework, database, UI or delivery-mechanism code, and there are no import cycles | Pass | ytmusicapi is imported only in infrastructure; use cases depend on the MusicService port. |
| 16 | SOLID holds: a class or module has one reason to change; new behaviour is added by extension, not by editing a type test in several places; subclasses honour the contract of their base; protocols are narrow; high-level code depends on protocols or abstract types and receives its implementations | Pass | time_travel composes use cases; MusicService stays small (five methods, each used). |
| 17 | Source is laid out in the layers and folders of the clean architecture folder structure (`domain`, `application`, `infrastructure`, `presentation` packages under `src/`, and `unit`, `integration` and `e2e` under `tests/`), in `snake_case`, and imports between the layers point inward | Pass | Layer folders and tests/unit/<layer> exist. |
| 18 | File and folder names are English, contain no space and no character outside letters, digits, `-`, `_` and `.`, and follow the casing of the language for files | Pass | English snake_case file names. |
## Overall Verdict
Go — every Mandatory criterion passes or does not apply; no blocking finding.
## Action Items
| Action | Owner | Due |
| --- | --- | --- |
| None | S01 | - |
---
[MIL-006]: ../../milestones/mil-006-step-4-search-and-add-songs.md
[QC-PY-001]: ../../../framework/qc/code/qc-programming-python.md
+2 -1
View File
@@ -30,7 +30,7 @@ Tracks backward and forward links between artifact instances so that the traceab
| [MIL-003] | MIL | en | it | [MIL-002] | [MIL-004] | [RC-007], [RC-013] | | [MIL-003] | MIL | en | it | [MIL-002] | [MIL-004] | [RC-007], [RC-013] |
| [MIL-004] | MIL | en | it | [MIL-003] | [MIL-005] | [RC-008], [RC-014] | | [MIL-004] | MIL | en | it | [MIL-003] | [MIL-005] | [RC-008], [RC-014] |
| [MIL-005] | MIL | en | it | [MIL-004] | [MIL-006] | [RC-009], [RC-015] | | [MIL-005] | MIL | en | it | [MIL-004] | [MIL-006] | [RC-009], [RC-015] |
| [MIL-006] | MIL | en | it | [MIL-005] | [MIL-007] | [RC-010] | | [MIL-006] | MIL | en | it | [MIL-005] | [MIL-007] | [RC-010], [RC-016] |
| [MIL-007] | MIL | en | it | [MIL-006] | - | [RC-011] | | [MIL-007] | MIL | en | it | [MIL-006] | - | [RC-011] |
| [TM-001] | TM | - | - | [BC-001] | - | - | | [TM-001] | TM | - | - | [BC-001] | - | - |
@@ -85,3 +85,4 @@ Tracks backward and forward links between artifact instances so that the traceab
[RC-013]: ./reviews/rc-013-mil-003-code.md [RC-013]: ./reviews/rc-013-mil-003-code.md
[RC-014]: ./reviews/rc-014-mil-004-code.md [RC-014]: ./reviews/rc-014-mil-004-code.md
[RC-015]: ./reviews/rc-015-mil-005-code.md [RC-015]: ./reviews/rc-015-mil-005-code.md
[RC-016]: ./reviews/rc-016-mil-006-code.md
Binary file not shown.

Before

Width:  |  Height:  |  Size: 124 KiB

After

Width:  |  Height:  |  Size: 124 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 64 KiB

@@ -40,3 +40,20 @@ class MusicService(Protocol):
@return The identifier of the new playlist. @return The identifier of the new playlist.
""" """
... ...
def search_song(self, title: str, artist: str) -> str | None:
"""@brief Find a song.
@param title Song title.
@param artist Performing artist.
@return The video id of the top song result, or None when nothing was found.
"""
...
def add_song(self, playlist_id: str, video_id: str) -> None:
"""@brief Add a song to a playlist.
@param playlist_id Identifier of the playlist.
@param video_id Video id of the song.
"""
...
@@ -2,11 +2,14 @@
@brief Use cases of the Musical Time Machine. @brief Use cases of the Musical Time Machine.
""" """
import time
from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from datetime import date from datetime import date
from musical_time_machine.application.ports import ChartSource, MusicService from musical_time_machine.application.ports import ChartSource, MusicService
from musical_time_machine.constants import PLAYLIST_PRIVACY from musical_time_machine.constants import ADD_PAUSE_SECONDS, PLAYLIST_PRIVACY
from musical_time_machine.domain.errors import SongAddError, SongSearchError
from musical_time_machine.domain.playlist import playlist_description, playlist_name from musical_time_machine.domain.playlist import playlist_description, playlist_name
from musical_time_machine.domain.song import Song from musical_time_machine.domain.song import Song
@@ -25,6 +28,34 @@ class PlaylistResult:
created: bool created: bool
@dataclass(frozen=True, slots=True)
class AddSongsResult:
"""@brief Outcome of add_songs.
@param added Number of songs added to the playlist.
@param skipped Songs that were not found or could not be added.
"""
added: int
skipped: list[Song]
@dataclass(frozen=True, slots=True)
class TimeTravelResult:
"""@brief Outcome of time_travel.
@param playlist The new or existing playlist.
@param added Number of songs added.
@param skipped Songs that were not found or could not be added.
@param total Number of songs on the chart.
"""
playlist: PlaylistResult
added: int
skipped: list[Song]
total: int
def scrape_songs(chart_date: date, source: ChartSource) -> list[Song]: def scrape_songs(chart_date: date, source: ChartSource) -> list[Song]:
"""@brief Get the songs of the chart of a date. """@brief Get the songs of the chart of a date.
@@ -67,3 +98,60 @@ def create_playlist(chart_date: date, service: MusicService) -> PlaylistResult:
return PlaylistResult(playlist.playlist_id, title, created=False) return PlaylistResult(playlist.playlist_id, title, created=False)
playlist_id = service.create_playlist(title, playlist_description(chart_date), PLAYLIST_PRIVACY) playlist_id = service.create_playlist(title, playlist_description(chart_date), PLAYLIST_PRIVACY)
return PlaylistResult(playlist_id, title, created=True) return PlaylistResult(playlist_id, title, created=True)
def add_songs(
songs: list[Song],
playlist_id: str,
service: MusicService,
*,
pause: Callable[[float], None] = time.sleep,
) -> AddSongsResult:
"""@brief Search every song and add the found ones to a playlist.
A song that is not found, or whose search or add fails, is skipped and the
run continues. Authentication errors are not caught: they stop the run.
@param songs The songs in chart order.
@param playlist_id Identifier of the target playlist.
@param service The music service to ask.
@param pause Called with the seconds to wait between two songs.
@return The number added and the skipped songs.
"""
added = 0
skipped: list[Song] = []
for index, song in enumerate(songs):
if index > 0:
pause(ADD_PAUSE_SECONDS)
try:
video_id = service.search_song(song.title, song.artist)
if video_id is None:
skipped.append(song)
continue
service.add_song(playlist_id, video_id)
except (SongSearchError, SongAddError):
skipped.append(song)
continue
added += 1
return AddSongsResult(added, skipped)
def time_travel(
chart_date: date,
chart_source: ChartSource,
service: MusicService,
*,
pause: Callable[[float], None] = time.sleep,
) -> TimeTravelResult:
"""@brief Scrape the chart, make its playlist and fill it.
@param chart_date Date of the chart.
@param chart_source Where the chart comes from.
@param service The music service to ask.
@param pause Called with the seconds to wait between two songs.
@return The playlist and what was added or skipped.
"""
songs = scrape_songs(chart_date, chart_source)
playlist = create_playlist(chart_date, service)
added = add_songs(songs, playlist.playlist_id, service, pause=pause)
return TimeTravelResult(playlist, added.added, added.skipped, len(songs))
+34
View File
@@ -147,3 +147,37 @@ PLAYLIST_EXISTS_MESSAGE: Final[str] = 'Playlist "{title}" already exists; nothin
## @brief Message when YouTube Music returned no playlist id; format with the answer. ## @brief Message when YouTube Music returned no playlist id; format with the answer.
PLAYLIST_CREATION_FAILED_MESSAGE: Final[str] = "The playlist could not be created: {reason}" PLAYLIST_CREATION_FAILED_MESSAGE: Final[str] = "The playlist could not be created: {reason}"
## @} ## @}
## @name Add songs
## @{
## @brief Seconds to wait between two songs, to stay gentle with YouTube Music.
ADD_PAUSE_SECONDS: Final[float] = 0.2
## @brief Number of search results to ask for; only the first is used.
SEARCH_LIMIT: Final[int] = 1
## @brief Status text ytmusicapi reports when songs were added.
STATUS_SUCCEEDED: Final[str] = "STATUS_SUCCEEDED"
## @brief HTTP status codes that mean the login was rejected.
AUTH_HTTP_CODES: Final[tuple[str, ...]] = ("HTTP 401", "HTTP 403")
## @brief Message when a search fails; format with the reason.
SONG_SEARCH_FAILED_MESSAGE: Final[str] = "The song search failed: {reason}"
## @brief Message when a song is not added; format with the answer.
SONG_ADD_FAILED_MESSAGE: Final[str] = "The song could not be added: {reason}"
## @brief Command line flag that only prints the chart songs.
LIST_ONLY_FLAG: Final[str] = "--list-only"
## @brief Help text of the list-only flag.
LIST_ONLY_HELP: Final[str] = "Only print the numbered chart songs; no YouTube Music login."
## @brief Summary line; format with added, total, title and skipped.
SUMMARY_MESSAGE: Final[str] = 'Added {added} of {total} songs to "{title}"; skipped {skipped}.'
## @brief One line per skipped song; format with title and artist.
SKIPPED_SONG_MESSAGE: Final[str] = "Skipped: {title} - {artist}"
## @}
@@ -29,3 +29,11 @@ class AuthenticationError(MusicalTimeMachineError):
class PlaylistCreationError(MusicalTimeMachineError): class PlaylistCreationError(MusicalTimeMachineError):
"""@brief YouTube Music did not create the playlist.""" """@brief YouTube Music did not create the playlist."""
class SongSearchError(MusicalTimeMachineError):
"""@brief YouTube Music could not search for a song."""
class SongAddError(MusicalTimeMachineError):
"""@brief YouTube Music did not add a song to the playlist."""
@@ -8,18 +8,26 @@ from pathlib import Path
from typing import Any, Protocol from typing import Any, Protocol
from ytmusicapi import YTMusic from ytmusicapi import YTMusic
from ytmusicapi.exceptions import YTMusicError from ytmusicapi.exceptions import YTMusicError, YTMusicServerError
from musical_time_machine.constants import ( from musical_time_machine.constants import (
AUTH_HTTP_CODES,
BROWSER_FILE_MISSING_REASON, BROWSER_FILE_MISSING_REASON,
BROWSER_FILE_NAME, BROWSER_FILE_NAME,
BROWSER_RENEWAL_MESSAGE, BROWSER_RENEWAL_MESSAGE,
PLAYLIST_CREATION_FAILED_MESSAGE, PLAYLIST_CREATION_FAILED_MESSAGE,
SEARCH_FILTER,
SEARCH_LIMIT,
SONG_ADD_FAILED_MESSAGE,
SONG_SEARCH_FAILED_MESSAGE,
STATUS_SUCCEEDED,
) )
from musical_time_machine.domain.errors import ( from musical_time_machine.domain.errors import (
AuthenticationError, AuthenticationError,
BrowserFileMissingError, BrowserFileMissingError,
PlaylistCreationError, PlaylistCreationError,
SongAddError,
SongSearchError,
) )
from musical_time_machine.domain.playlist import Playlist from musical_time_machine.domain.playlist import Playlist
@@ -48,6 +56,24 @@ class YTMusicLike(Protocol):
"""@brief Create a playlist; returns its id or an error dictionary.""" """@brief Create a playlist; returns its id or an error dictionary."""
... ...
def search(
self,
query: str,
filter: str | None = ...,
limit: int = ...,
) -> list[dict[str, Any]]: # ytmusicapi returns untyped JSON dictionaries
"""@brief Search YouTube Music; returns the raw results."""
...
def add_playlist_items(
self,
playlistId: str, # noqa: N803 - must match the ytmusicapi parameter names
videoIds: list[str] | None = ..., # noqa: N803
duplicates: bool = ...,
) -> str | dict[str, Any]: # a status dictionary, or the raw answer on failure
"""@brief Add videos to a playlist; returns the status answer."""
...
class YouTubeMusicService: class YouTubeMusicService:
"""@brief MusicService backed by ytmusicapi and a browser headers file.""" """@brief MusicService backed by ytmusicapi and a browser headers file."""
@@ -104,6 +130,57 @@ class YouTubeMusicService:
raise PlaylistCreationError(PLAYLIST_CREATION_FAILED_MESSAGE.format(reason=result)) raise PlaylistCreationError(PLAYLIST_CREATION_FAILED_MESSAGE.format(reason=result))
return result return result
def search_song(self, title: str, artist: str) -> str | None:
"""@brief Find the top song result for a title and artist.
@param title Song title.
@param artist Performing artist.
@return The video id of the first song result, or None when nothing was found.
@throws BrowserFileMissingError when the browser file does not exist.
@throws AuthenticationError when YouTube Music rejects the headers.
@throws SongSearchError when the search fails.
"""
client = self._connect()
try:
results = client.search(f"{title} {artist}", filter=SEARCH_FILTER, limit=SEARCH_LIMIT)
except _CLIENT_ERRORS as error:
raise self._failure(error, SongSearchError, SONG_SEARCH_FAILED_MESSAGE) from error
for result in results:
video_id = result.get("videoId")
if video_id:
return str(video_id)
return None
def add_song(self, playlist_id: str, video_id: str) -> None:
"""@brief Add a song to a playlist, without duplicates.
@param playlist_id Identifier of the playlist.
@param video_id Video id of the song.
@throws BrowserFileMissingError when the browser file does not exist.
@throws AuthenticationError when YouTube Music rejects the headers.
@throws SongAddError when YouTube Music does not confirm the addition.
"""
client = self._connect()
try:
answer = client.add_playlist_items(playlist_id, [video_id], duplicates=False)
except _CLIENT_ERRORS as error:
raise self._failure(error, SongAddError, SONG_ADD_FAILED_MESSAGE) from error
status = answer.get("status") if isinstance(answer, dict) else None
if status != STATUS_SUCCEEDED:
raise SongAddError(SONG_ADD_FAILED_MESSAGE.format(reason=answer))
def _failure(
self,
error: Exception,
song_error: type[SongSearchError] | type[SongAddError],
template: str,
) -> AuthenticationError | SongSearchError | SongAddError:
# A server error is a failure of this song; a 401/403 or a parse error is a rejected login.
is_server_error = isinstance(error, YTMusicServerError)
if is_server_error and not any(code in str(error) for code in AUTH_HTTP_CODES):
return song_error(template.format(reason=error))
return self._rejected(error)
def _connect(self) -> YTMusicLike: def _connect(self) -> YTMusicLike:
if self._client is not None: if self._client is not None:
return self._client return self._client
+55 -11
View File
@@ -3,6 +3,7 @@
""" """
import argparse import argparse
import time
from collections.abc import Callable from collections.abc import Callable
from pathlib import Path from pathlib import Path
@@ -12,6 +13,7 @@ from musical_time_machine.application.use_cases import (
count_library_playlists, count_library_playlists,
create_playlist, create_playlist,
scrape_songs, scrape_songs,
time_travel,
) )
from musical_time_machine.constants import ( from musical_time_machine.constants import (
BROWSER_FILE_FLAG, BROWSER_FILE_FLAG,
@@ -24,10 +26,14 @@ from musical_time_machine.constants import (
DATE_PROMPT, DATE_PROMPT,
EXIT_FAILURE, EXIT_FAILURE,
EXIT_OK, EXIT_OK,
LIST_ONLY_FLAG,
LIST_ONLY_HELP,
PLAYLIST_COUNT_MESSAGE, PLAYLIST_COUNT_MESSAGE,
PLAYLIST_CREATED_MESSAGE, PLAYLIST_CREATED_MESSAGE,
PLAYLIST_EXISTS_MESSAGE, PLAYLIST_EXISTS_MESSAGE,
SKIPPED_SONG_MESSAGE,
SONG_LINE_TEMPLATE, SONG_LINE_TEMPLATE,
SUMMARY_MESSAGE,
) )
from musical_time_machine.domain.chart_date import parse_chart_date from musical_time_machine.domain.chart_date import parse_chart_date
from musical_time_machine.domain.errors import MusicalTimeMachineError from musical_time_machine.domain.errors import MusicalTimeMachineError
@@ -39,6 +45,7 @@ def _parse_args(argv: list[str] | None) -> argparse.Namespace:
parser = argparse.ArgumentParser(prog="musical_time_machine") parser = argparse.ArgumentParser(prog="musical_time_machine")
parser.add_argument(CHECK_LOGIN_FLAG, action="store_true", help=CHECK_LOGIN_HELP) parser.add_argument(CHECK_LOGIN_FLAG, action="store_true", help=CHECK_LOGIN_HELP)
parser.add_argument(CREATE_PLAYLIST_FLAG, action="store_true", help=CREATE_PLAYLIST_HELP) parser.add_argument(CREATE_PLAYLIST_FLAG, action="store_true", help=CREATE_PLAYLIST_HELP)
parser.add_argument(LIST_ONLY_FLAG, action="store_true", help=LIST_ONLY_HELP)
parser.add_argument( parser.add_argument(
BROWSER_FILE_FLAG, type=Path, default=Path(BROWSER_FILE_NAME), help=BROWSER_FILE_HELP BROWSER_FILE_FLAG, type=Path, default=Path(BROWSER_FILE_NAME), help=BROWSER_FILE_HELP
) )
@@ -68,6 +75,45 @@ def _create_playlist(
return EXIT_OK return EXIT_OK
def _list_only(
chart_source: ChartSource, read: Callable[[str], str], write: Callable[[str], None]
) -> int:
try:
songs = scrape_songs(parse_chart_date(read(DATE_PROMPT)), chart_source)
except MusicalTimeMachineError as error:
write(str(error))
return EXIT_FAILURE
for position, song in enumerate(songs, start=1):
write(SONG_LINE_TEMPLATE.format(position=position, title=song.title, artist=song.artist))
return EXIT_OK
def _full_run(
chart_source: ChartSource,
service: MusicService,
read: Callable[[str], str],
write: Callable[[str], None],
pause: Callable[[float], None],
) -> int:
try:
chart_date = parse_chart_date(read(DATE_PROMPT))
result = time_travel(chart_date, chart_source, service, pause=pause)
except MusicalTimeMachineError as error:
write(str(error))
return EXIT_FAILURE
write(
SUMMARY_MESSAGE.format(
added=result.added,
total=result.total,
title=result.playlist.title,
skipped=len(result.skipped),
)
)
for song in result.skipped:
write(SKIPPED_SONG_MESSAGE.format(title=song.title, artist=song.artist))
return EXIT_OK
def main( def main(
argv: list[str] | None = None, argv: list[str] | None = None,
*, *,
@@ -75,14 +121,17 @@ def main(
music_service: MusicService | None = None, music_service: MusicService | None = None,
read: Callable[[str], str] = input, read: Callable[[str], str] = input,
write: Callable[[str], None] = print, write: Callable[[str], None] = print,
pause: Callable[[float], None] = time.sleep,
) -> int: ) -> int:
"""@brief Ask for a date and print that chart, check the login or create a playlist. """@brief Ask for a date and fill a YouTube Music playlist with that chart.
@param argv Arguments: `--check-login`, `--create-playlist`, `--browser-file PATH`. @param argv Arguments: `--check-login`, `--create-playlist`, `--list-only`,
`--browser-file PATH`.
@param source Chart source; the Bakeboard site when omitted. @param source Chart source; the Bakeboard site when omitted.
@param music_service Music service for the YouTube Music flags; YouTube Music when omitted. @param music_service Music service for the YouTube Music flags; YouTube Music when omitted.
@param read Function that prompts and returns the user's answer. @param read Function that prompts and returns the user's answer.
@param write Function that shows one line to the user. @param write Function that shows one line to the user.
@param pause Function that waits the given seconds between songs.
@return Process exit code: 0 on success, 1 when a domain error stopped the run. @return Process exit code: 0 on success, 1 when a domain error stopped the run.
""" """
args = _parse_args(argv) args = _parse_args(argv)
@@ -93,12 +142,7 @@ def main(
service = music_service or YouTubeMusicService(browser_file=args.browser_file) service = music_service or YouTubeMusicService(browser_file=args.browser_file)
return _create_playlist(service, read, write) return _create_playlist(service, read, write)
chart_source = source if source is not None else BakeboardChartSource() chart_source = source if source is not None else BakeboardChartSource()
try: if args.list_only:
chart_date = parse_chart_date(read(DATE_PROMPT)) return _list_only(chart_source, read, write)
songs = scrape_songs(chart_date, chart_source) service = music_service or YouTubeMusicService(browser_file=args.browser_file)
except MusicalTimeMachineError as error: return _full_run(chart_source, service, read, write, pause)
write(str(error))
return EXIT_FAILURE
for position, song in enumerate(songs, start=1):
write(SONG_LINE_TEMPLATE.format(position=position, title=song.title, artist=song.artist))
return EXIT_OK
+24 -1
View File
@@ -38,11 +38,21 @@ class FakeMusicService:
"""@brief MusicService returning fixed playlists, or raising a given error.""" """@brief MusicService returning fixed playlists, or raising a given error."""
def __init__( def __init__(
self, playlists: list[Playlist] | None = None, error: Exception | None = None self,
playlists: list[Playlist] | None = None,
error: Exception | None = None,
video_ids: dict[str, str] | None = None,
search_errors: dict[str, Exception] | None = None,
add_errors: dict[str, Exception] | None = None,
) -> None: ) -> None:
self._playlists = list(playlists or []) self._playlists = list(playlists or [])
self._error = error self._error = error
self._video_ids = video_ids
self._search_errors = search_errors or {}
self._add_errors = add_errors or {}
self.created: list[tuple[str, str, str]] = [] self.created: list[tuple[str, str, str]] = []
self.searched: list[tuple[str, str]] = []
self.added: list[tuple[str, str]] = []
def create_playlist(self, title: str, description: str, privacy: str) -> str: def create_playlist(self, title: str, description: str, privacy: str) -> str:
if self._error is not None: if self._error is not None:
@@ -56,3 +66,16 @@ class FakeMusicService:
if self._error is not None: if self._error is not None:
raise self._error raise self._error
return self._playlists return self._playlists
def search_song(self, title: str, artist: str) -> str | None:
self.searched.append((title, artist))
if title in self._search_errors:
raise self._search_errors[title]
if self._video_ids is None:
return f"vid-{title}"
return self._video_ids.get(title)
def add_song(self, playlist_id: str, video_id: str) -> None:
if video_id in self._add_errors:
raise self._add_errors[video_id]
self.added.append((playlist_id, video_id))
+104
View File
@@ -0,0 +1,104 @@
"""@file
@brief Tests of the add_songs and time_travel use cases with fake ports.
"""
from datetime import date
import pytest
from conftest import FakeChartSource, FakeMusicService
from musical_time_machine.application.use_cases import add_songs, time_travel
from musical_time_machine.constants import ADD_PAUSE_SECONDS
from musical_time_machine.domain.errors import (
AuthenticationError,
BrowserFileMissingError,
SongAddError,
SongSearchError,
)
from musical_time_machine.domain.song import Song
ONE = Song("One", "A")
TWO = Song("Two", "B")
THREE = Song("Three", "C")
def _no_pause(_seconds: float) -> None:
return None
def test_add_songs_adds_every_found_song() -> None:
service = FakeMusicService()
result = add_songs([ONE, TWO], "pl", service, pause=_no_pause)
assert (result.added, result.skipped) == (2, [])
assert service.added == [("pl", "vid-One"), ("pl", "vid-Two")]
assert service.searched == [("One", "A"), ("Two", "B")]
def test_add_songs_skips_song_not_found() -> None:
service = FakeMusicService(video_ids={"One": "v1", "Three": "v3"})
result = add_songs([ONE, TWO, THREE], "pl", service, pause=_no_pause)
assert result.added == 2
assert result.skipped == [TWO]
assert service.added == [("pl", "v1"), ("pl", "v3")]
def test_add_songs_skips_song_whose_add_fails_and_continues() -> None:
service = FakeMusicService(add_errors={"vid-One": SongAddError("no")})
result = add_songs([ONE, TWO], "pl", service, pause=_no_pause)
assert result.added == 1
assert result.skipped == [ONE]
assert service.added == [("pl", "vid-Two")]
def test_add_songs_skips_song_whose_search_fails_and_continues() -> None:
service = FakeMusicService(search_errors={"One": SongSearchError("no")})
result = add_songs([ONE, TWO], "pl", service, pause=_no_pause)
assert (result.added, result.skipped) == (1, [ONE])
@pytest.mark.parametrize("error", [AuthenticationError("auth"), BrowserFileMissingError("gone")])
def test_add_songs_propagates_login_errors(error: Exception) -> None:
service = FakeMusicService(search_errors={"One": error})
with pytest.raises(type(error)):
add_songs([ONE, TWO], "pl", service, pause=_no_pause)
assert service.searched == [("One", "A")]
def test_add_songs_pauses_between_songs_only() -> None:
pauses: list[float] = []
add_songs([ONE, TWO, THREE], "pl", FakeMusicService(), pause=pauses.append)
assert pauses == [ADD_PAUSE_SECONDS, ADD_PAUSE_SECONDS]
def test_add_songs_with_no_songs_adds_nothing() -> None:
pauses: list[float] = []
result = add_songs([], "pl", FakeMusicService(), pause=pauses.append)
assert (result.added, result.skipped, pauses) == (0, [], [])
def test_time_travel_scrapes_creates_playlist_and_adds() -> None:
source = FakeChartSource([ONE, TWO, THREE])
service = FakeMusicService(video_ids={"One": "v1", "Three": "v3"})
result = time_travel(date(2026, 4, 18), source, service, pause=_no_pause)
playlist_id = result.playlist.playlist_id
assert source.requested == [date(2026, 4, 18)]
assert result.playlist.title == "2026-04-18 Billboard 100"
assert result.playlist.created is True
assert (result.added, result.skipped, result.total) == (2, [TWO], 3)
assert service.added == [(playlist_id, "v1"), (playlist_id, "v3")]
+107 -1
View File
@@ -6,12 +6,14 @@ from pathlib import Path
from typing import Any from typing import Any
import pytest import pytest
from ytmusicapi.exceptions import YTMusicUserError from ytmusicapi.exceptions import YTMusicServerError, YTMusicUserError
from musical_time_machine.domain.errors import ( from musical_time_machine.domain.errors import (
AuthenticationError, AuthenticationError,
BrowserFileMissingError, BrowserFileMissingError,
PlaylistCreationError, PlaylistCreationError,
SongAddError,
SongSearchError,
) )
from musical_time_machine.domain.playlist import Playlist from musical_time_machine.domain.playlist import Playlist
from musical_time_machine.infrastructure.ytmusic import YouTubeMusicService, YTMusicLike from musical_time_machine.infrastructure.ytmusic import YouTubeMusicService, YTMusicLike
@@ -23,7 +25,13 @@ class _FakeClient:
items: list[dict[str, Any]], items: list[dict[str, Any]],
error: Exception | None = None, error: Exception | None = None,
created: str | dict[str, Any] = "new-id", created: str | dict[str, Any] = "new-id",
results: list[dict[str, Any]] | None = None,
added: str | dict[str, Any] = "ignored",
) -> None: ) -> None:
self._results = results or []
self._added = added
self.searches: list[tuple[str, str | None, int]] = []
self.additions: list[tuple[str, list[str] | None, bool]] = []
self._items = items self._items = items
self._error = error self._error = error
self._created = created self._created = created
@@ -38,6 +46,25 @@ class _FakeClient:
raise self._error raise self._error
return self._created return self._created
def search(
self, query: str, filter: str | None = None, limit: int = 20
) -> list[dict[str, Any]]:
self.searches.append((query, filter, limit))
if self._error is not None:
raise self._error
return self._results
def add_playlist_items(
self,
playlistId: str, # noqa: N803 - must match the ytmusicapi parameter names
videoIds: list[str] | None = None, # noqa: N803
duplicates: bool = False,
) -> str | dict[str, Any]:
self.additions.append((playlistId, videoIds, duplicates))
if self._error is not None:
raise self._error
return self._added
def get_library_playlists(self, limit: int | None = 25) -> list[dict[str, Any]]: def get_library_playlists(self, limit: int | None = 25) -> list[dict[str, Any]]:
self.limits.append(limit) self.limits.append(limit)
if self._error is not None: if self._error is not None:
@@ -162,3 +189,82 @@ def test_create_playlist_raises_when_browser_file_missing(tmp_path: Path) -> Non
with pytest.raises(BrowserFileMissingError): with pytest.raises(BrowserFileMissingError):
service.create_playlist("T", "D", "PRIVATE") service.create_playlist("T", "D", "PRIVATE")
def _service(tmp_path: Path, client: _FakeClient) -> YouTubeMusicService:
return YouTubeMusicService(_browser_file(tmp_path), lambda _auth: client)
def test_search_song_returns_first_video_id_and_filters_songs(tmp_path: Path) -> None:
client = _FakeClient([], results=[{"videoId": "v1"}, {"videoId": "v2"}])
assert _service(tmp_path, client).search_song("Title", "Artist") == "v1"
assert client.searches == [("Title Artist", "songs", 1)]
def test_search_song_returns_none_for_no_results(tmp_path: Path) -> None:
assert _service(tmp_path, _FakeClient([])).search_song("T", "A") is None
def test_search_song_skips_results_without_video_id(tmp_path: Path) -> None:
client = _FakeClient([], results=[{"title": "x"}, {"videoId": "v2"}])
assert _service(tmp_path, client).search_song("T", "A") == "v2"
def test_search_song_maps_server_error_to_song_search_error(tmp_path: Path) -> None:
error = YTMusicServerError("Server returned HTTP 500: oops")
service = _service(tmp_path, _FakeClient([], error))
with pytest.raises(SongSearchError) as info:
service.search_song("T", "A")
assert info.value.__cause__ is error
@pytest.mark.parametrize(
"error",
[YTMusicServerError("Server returned HTTP 401: no"), YTMusicUserError("x"), KeyError("k")],
)
def test_search_song_keeps_login_failures_as_authentication_error(
tmp_path: Path, error: Exception
) -> None:
with pytest.raises(AuthenticationError):
_service(tmp_path, _FakeClient([], error)).search_song("T", "A")
def test_add_song_sends_one_video_without_duplicates(tmp_path: Path) -> None:
client = _FakeClient([], added={"status": "STATUS_SUCCEEDED", "playlistEditResults": []})
_service(tmp_path, client).add_song("pl", "v1")
assert client.additions == [("pl", ["v1"], False)]
@pytest.mark.parametrize("answer", [{"status": "STATUS_FAILED"}, {"error": "dup"}, "oops"])
def test_add_song_raises_when_status_is_not_succeeded(
tmp_path: Path, answer: str | dict[str, Any]
) -> None:
service = _service(tmp_path, _FakeClient([], added=answer))
with pytest.raises(SongAddError):
service.add_song("pl", "v1")
def test_add_song_maps_server_error_to_song_add_error(tmp_path: Path) -> None:
error = YTMusicServerError("Server returned HTTP 400: bad")
with pytest.raises(SongAddError) as info:
_service(tmp_path, _FakeClient([], error)).add_song("pl", "v1")
assert info.value.__cause__ is error
def test_add_song_keeps_login_failure_as_authentication_error(tmp_path: Path) -> None:
with pytest.raises(AuthenticationError):
_service(tmp_path, _FakeClient([], KeyError("cookie"))).add_song("pl", "v1")
def test_search_song_raises_when_browser_file_missing(tmp_path: Path) -> None:
service = YouTubeMusicService(tmp_path / "none.json", lambda _auth: _FakeClient([]))
with pytest.raises(BrowserFileMissingError):
service.search_song("T", "A")
+57 -3
View File
@@ -26,7 +26,7 @@ def _run(answer: str, source: FakeChartSource) -> tuple[int, list[str], list[str
prompts.append(prompt) prompts.append(prompt)
return answer return answer
code = main([], source=source, read=read, write=lines.append) code = main(["--list-only"], source=source, read=read, write=lines.append)
return code, lines, prompts return code, lines, prompts
@@ -97,12 +97,12 @@ def test_main_check_login_with_missing_real_file_returns_one(tmp_path: Path) ->
assert "ytmusicapi browser" in lines[0] assert "ytmusicapi browser" in lines[0]
def test_main_without_flag_does_not_use_the_music_service() -> None: def test_main_list_only_does_not_use_the_music_service() -> None:
lines: list[str] = [] lines: list[str] = []
service = FakeMusicService(error=BrowserFileMissingError("must not be called")) service = FakeMusicService(error=BrowserFileMissingError("must not be called"))
code = main( code = main(
[], ["--list-only"],
source=FakeChartSource([Song("T", "A")]), source=FakeChartSource([Song("T", "A")]),
music_service=service, music_service=service,
read=lambda _prompt: "2026-04-18", read=lambda _prompt: "2026-04-18",
@@ -159,3 +159,57 @@ def test_main_create_playlist_returns_one_on_domain_error() -> None:
assert code == 1 assert code == 1
assert lines == ["failed"] assert lines == ["failed"]
def _full(
source: FakeChartSource, service: FakeMusicService, answer: str = "2026-04-18"
) -> tuple[int, list[str], list[float]]:
lines: list[str] = []
pauses: list[float] = []
code = main(
[],
source=source,
music_service=service,
read=lambda _prompt: answer,
write=lines.append,
pause=pauses.append,
)
return code, lines, pauses
def test_main_full_run_adds_songs_and_prints_summary_with_skipped() -> None:
source = FakeChartSource([Song("One", "A"), Song("Two", "B"), Song("Three", "C")])
service = FakeMusicService(video_ids={"One": "v1", "Three": "v3"})
code, lines, pauses = _full(source, service)
assert code == 0
assert lines == [
'Added 2 of 3 songs to "2026-04-18 Billboard 100"; skipped 1.',
"Skipped: Two - B",
]
assert len(service.created) == 1
assert len(pauses) == 2
def test_main_full_run_without_skips_prints_only_summary() -> None:
code, lines, _ = _full(FakeChartSource([Song("One", "A")]), FakeMusicService())
assert code == 0
assert lines == ['Added 1 of 1 songs to "2026-04-18 Billboard 100"; skipped 0.']
def test_main_full_run_returns_one_for_invalid_date() -> None:
code, lines, _ = _full(FakeChartSource([Song("x", "y")]), FakeMusicService(), "nope")
assert code == 1
assert "YYYY-MM-DD" in lines[0]
def test_main_full_run_returns_one_on_login_error() -> None:
service = FakeMusicService(error=BrowserFileMissingError("browser.json was not found."))
code, lines, _ = _full(FakeChartSource([Song("x", "y")]), service)
assert code == 1
assert lines == ["browser.json was not found."]