MIL-006: Step 4 - Search and add songs #48
@@ -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) |
|
||||
| 2. YouTube Music authentication | 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
|
||||
|
||||
```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.
|
||||
@@ -128,7 +128,17 @@ The screenshot shows the tests of this step, which use a fake YouTube Music serv
|
||||
|
||||

|
||||
|
||||
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
|
||||
|
||||
|
||||
@@ -17,7 +17,7 @@ document of a type. `Primary File` may contain a glob (e.g.
|
||||
| MIL | Milestone / Gateway | docs/milestones/mil-*.md | 008 |
|
||||
| GOV | Governance / ARB Workflow | docs/sqa/governance.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
|
||||
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
# 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 | [9f0e66d] |
|
||||
|
||||
---
|
||||
|
||||
## 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
|
||||
[9f0e66d]: https://git.tirsystem.com/Tirsvad-Udemy-100-days-of-code/046-scraping-the-billboard/commit/9f0e66d20a704b1610cf2dea07c4d4acd6fb3dab
|
||||
@@ -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-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-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] |
|
||||
| [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-014]: ./reviews/rc-014-mil-004-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.
|
||||
"""
|
||||
...
|
||||
|
||||
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.
|
||||
"""
|
||||
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
|
||||
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.song import Song
|
||||
|
||||
@@ -25,6 +28,34 @@ class PlaylistResult:
|
||||
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]:
|
||||
"""@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)
|
||||
playlist_id = service.create_playlist(title, playlist_description(chart_date), PLAYLIST_PRIVACY)
|
||||
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))
|
||||
|
||||
@@ -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.
|
||||
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):
|
||||
"""@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 ytmusicapi import YTMusic
|
||||
from ytmusicapi.exceptions import YTMusicError
|
||||
from ytmusicapi.exceptions import YTMusicError, YTMusicServerError
|
||||
|
||||
from musical_time_machine.constants import (
|
||||
AUTH_HTTP_CODES,
|
||||
BROWSER_FILE_MISSING_REASON,
|
||||
BROWSER_FILE_NAME,
|
||||
BROWSER_RENEWAL_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 (
|
||||
AuthenticationError,
|
||||
BrowserFileMissingError,
|
||||
PlaylistCreationError,
|
||||
SongAddError,
|
||||
SongSearchError,
|
||||
)
|
||||
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."""
|
||||
...
|
||||
|
||||
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:
|
||||
"""@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))
|
||||
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:
|
||||
if self._client is not None:
|
||||
return self._client
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
@@ -12,6 +13,7 @@ from musical_time_machine.application.use_cases import (
|
||||
count_library_playlists,
|
||||
create_playlist,
|
||||
scrape_songs,
|
||||
time_travel,
|
||||
)
|
||||
from musical_time_machine.constants import (
|
||||
BROWSER_FILE_FLAG,
|
||||
@@ -24,10 +26,14 @@ from musical_time_machine.constants import (
|
||||
DATE_PROMPT,
|
||||
EXIT_FAILURE,
|
||||
EXIT_OK,
|
||||
LIST_ONLY_FLAG,
|
||||
LIST_ONLY_HELP,
|
||||
PLAYLIST_COUNT_MESSAGE,
|
||||
PLAYLIST_CREATED_MESSAGE,
|
||||
PLAYLIST_EXISTS_MESSAGE,
|
||||
SKIPPED_SONG_MESSAGE,
|
||||
SONG_LINE_TEMPLATE,
|
||||
SUMMARY_MESSAGE,
|
||||
)
|
||||
from musical_time_machine.domain.chart_date import parse_chart_date
|
||||
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.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(LIST_ONLY_FLAG, action="store_true", help=LIST_ONLY_HELP)
|
||||
parser.add_argument(
|
||||
BROWSER_FILE_FLAG, type=Path, default=Path(BROWSER_FILE_NAME), help=BROWSER_FILE_HELP
|
||||
)
|
||||
@@ -68,6 +75,45 @@ def _create_playlist(
|
||||
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(
|
||||
argv: list[str] | None = None,
|
||||
*,
|
||||
@@ -75,14 +121,17 @@ def main(
|
||||
music_service: MusicService | None = None,
|
||||
read: Callable[[str], str] = input,
|
||||
write: Callable[[str], None] = print,
|
||||
pause: Callable[[float], None] = time.sleep,
|
||||
) -> 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 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 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.
|
||||
"""
|
||||
args = _parse_args(argv)
|
||||
@@ -93,12 +142,7 @@ def main(
|
||||
service = music_service or YouTubeMusicService(browser_file=args.browser_file)
|
||||
return _create_playlist(service, read, write)
|
||||
chart_source = source if source is not None else BakeboardChartSource()
|
||||
try:
|
||||
chart_date = parse_chart_date(read(DATE_PROMPT))
|
||||
songs = scrape_songs(chart_date, 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
|
||||
if args.list_only:
|
||||
return _list_only(chart_source, read, write)
|
||||
service = music_service or YouTubeMusicService(browser_file=args.browser_file)
|
||||
return _full_run(chart_source, service, read, write, pause)
|
||||
|
||||
+24
-1
@@ -38,11 +38,21 @@ class FakeMusicService:
|
||||
"""@brief MusicService returning fixed playlists, or raising a given error."""
|
||||
|
||||
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:
|
||||
self._playlists = list(playlists or [])
|
||||
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.searched: list[tuple[str, str]] = []
|
||||
self.added: list[tuple[str, str]] = []
|
||||
|
||||
def create_playlist(self, title: str, description: str, privacy: str) -> str:
|
||||
if self._error is not None:
|
||||
@@ -56,3 +66,16 @@ class FakeMusicService:
|
||||
if self._error is not None:
|
||||
raise self._error
|
||||
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))
|
||||
|
||||
@@ -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")]
|
||||
@@ -6,12 +6,14 @@ from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from ytmusicapi.exceptions import YTMusicUserError
|
||||
from ytmusicapi.exceptions import YTMusicServerError, YTMusicUserError
|
||||
|
||||
from musical_time_machine.domain.errors import (
|
||||
AuthenticationError,
|
||||
BrowserFileMissingError,
|
||||
PlaylistCreationError,
|
||||
SongAddError,
|
||||
SongSearchError,
|
||||
)
|
||||
from musical_time_machine.domain.playlist import Playlist
|
||||
from musical_time_machine.infrastructure.ytmusic import YouTubeMusicService, YTMusicLike
|
||||
@@ -23,7 +25,13 @@ class _FakeClient:
|
||||
items: list[dict[str, Any]],
|
||||
error: Exception | None = None,
|
||||
created: str | dict[str, Any] = "new-id",
|
||||
results: list[dict[str, Any]] | None = None,
|
||||
added: str | dict[str, Any] = "ignored",
|
||||
) -> 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._error = error
|
||||
self._created = created
|
||||
@@ -38,6 +46,25 @@ class _FakeClient:
|
||||
raise self._error
|
||||
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]]:
|
||||
self.limits.append(limit)
|
||||
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):
|
||||
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")
|
||||
|
||||
@@ -26,7 +26,7 @@ def _run(answer: str, source: FakeChartSource) -> tuple[int, list[str], list[str
|
||||
prompts.append(prompt)
|
||||
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
|
||||
|
||||
|
||||
@@ -97,12 +97,12 @@ def test_main_check_login_with_missing_real_file_returns_one(tmp_path: Path) ->
|
||||
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] = []
|
||||
service = FakeMusicService(error=BrowserFileMissingError("must not be called"))
|
||||
|
||||
code = main(
|
||||
[],
|
||||
["--list-only"],
|
||||
source=FakeChartSource([Song("T", "A")]),
|
||||
music_service=service,
|
||||
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 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."]
|
||||
|
||||
Reference in New Issue
Block a user