Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 35 additions & 14 deletions cogs/lyrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,11 +8,17 @@
from discord.ext import commands

from services import lyrics_api, synced_lyrics
from services.synced_lyrics import Lyrics, index_at
# Deliberately not `from ... import Lyrics`: the cog class below is called
# Lyrics too, and importing the bare name let it shadow the dataclass. The
# Genius fallback then built a Cog instead of a result and raised TypeError,
# which the command reported as nothing at all.
from services.synced_lyrics import index_at, search_terms
from utils.embeds import (error_embed, info_embed, lyrics_embed, lyrics_pages,
synced_lyrics_embed)
from utils.player import players

LyricsResult = synced_lyrics.Lyrics

log = logging.getLogger("loopify.lyrics")

# Discord allows roughly five message edits per five seconds per channel, and a
Expand All @@ -28,7 +34,7 @@
MAX_SLEEP = 2.0
MIN_SLEEP = 0.25

Loader = Callable[[dict], Awaitable[Optional[Lyrics]]]
Loader = Callable[[dict], Awaitable[Optional[LyricsResult]]]


class LyricsFollower:
Expand All @@ -55,7 +61,7 @@ def stop(self) -> None:
async def run(self) -> None:
"""Follow the music until the song, the player or the message runs out."""
track: Optional[dict] = None
lyrics: Optional[Lyrics] = None
lyrics: Optional[LyricsResult] = None
shown: Optional[int] = None
last_edit = float("-inf")

Expand All @@ -80,7 +86,7 @@ async def run(self) -> None:

await self._sleep(self._until_next_line(lyrics.lines, index, position))

async def _show(self, lyrics: Lyrics, index: int, position: float,
async def _show(self, lyrics: LyricsResult, index: int, position: float,
track: Optional[dict]) -> bool:
"""Redraw the message. False means it is gone and we should stop."""
try:
Expand Down Expand Up @@ -183,21 +189,36 @@ def _split_query(query: str) -> tuple[str, str]:
return query.strip(), ""

async def _find(self, title: str, artist: str,
duration: Optional[float]) -> Optional[Lyrics]:
duration: Optional[float]) -> Optional[LyricsResult]:
"""LRCLIB first, since it is the only source with timings, then Genius."""
found = await synced_lyrics.fetch(self._session, title, artist, duration)
if found is not None:
return found
fallback = await lyrics_api.fetch(title, artist)
if fallback is None:
return None
return Lyrics(title=fallback["title"], artist=fallback["artist"],
plain=fallback["lyrics"])
return LyricsResult(title=fallback["title"], artist=fallback["artist"],
plain=fallback["lyrics"])

async def _search(self, query: str) -> Optional[LyricsResult]:
"""
Look up a typed query, in either order.

async def _for_track(self, track: dict) -> Optional[Lyrics]:
return await self._find(track.get("title", ""),
track.get("uploader") or "",
track.get("duration"))
The help documents ``<title> - <artist>``, but ``The Police - Every
Breath You Take`` is how people actually type it. The documented order
is tried first, and the other only if it found nothing.
"""
title, artist = self._split_query(query)
found = await self._find(title, artist, None)
if found is None and artist:
found = await self._find(artist, title, None)
return found

async def _for_track(self, track: dict) -> Optional[LyricsResult]:
"""Look a playing track up, with its title tidied into search terms."""
title, artist = search_terms(track.get("title") or "",
track.get("uploader") or "")
return await self._find(title, artist, track.get("duration"))

# -- Following -----------------------------------------------------

Expand All @@ -211,7 +232,7 @@ def stop_following(self, guild_id: int) -> None:
if not task.done():
task.cancel()

async def _follow(self, ctx, player, lyrics: Lyrics) -> None:
async def _follow(self, ctx, player, lyrics: LyricsResult) -> None:
"""Post the live message and start keeping it up to date."""
self.stop_following(ctx.guild.id) # one per guild; the newest wins
index = index_at(lyrics.lines, player.position)
Expand Down Expand Up @@ -243,8 +264,8 @@ async def lyrics(self, ctx, *, query: str = None):
async with ctx.typing():
player = players.get(ctx.guild.id)
if query:
title, artist = self._split_query(query)
found = await self._find(title, artist, None)
title = self._split_query(query)[0]
found = await self._search(query)
elif player and player.current:
title = player.current.get("title", "")
found = await self._for_track(player.current)
Expand Down
53 changes: 51 additions & 2 deletions services/synced_lyrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,12 +10,12 @@

import logging
import re

import aiohttp
from bisect import bisect_right
from dataclasses import dataclass
from typing import Optional

import aiohttp

log = logging.getLogger("loopify.synced")

API_URL = "https://lrclib.net/api/get"
Expand Down Expand Up @@ -56,6 +56,55 @@ def parse_lrc(body: str) -> tuple[Line, ...]:
return tuple(sorted(lines, key=lambda line: line[0]))


# Words that mark a bracketed chunk as video furniture rather than part of the
# song's name. "(Acoustic Version)" is the song; "(Official Video)" is not.
_FURNITURE = re.compile(
r"[(\[][^)\]]*\b(?:official|video|audio|lyrics?|visuali[sz]er"
r"|remaster(?:ed)?|hd|hq|4k|8k|mv|explicit|clean|live|oficial|letra"
r"|legendado|sub(?:titulado)?|full album|color coded)\b[^)\]]*[)\]]",
re.IGNORECASE)
# "Song ft. Someone" — databases file a track under its lead artist alone.
_CREDITS = re.compile(r"\s*\b(?:ft|feat)\.?\s.*$", re.IGNORECASE)
# A trailing "| Album Name" or "| Official Video".
_TRAILING_PIPE = re.compile(r"\s*\|.*$")
# What separates an artist from a title. A run of two or more spaces counts:
# plenty of uploads write "The Police Every Breath You Take" with no dash.
_SEPARATOR = re.compile(r"\s[-–—]\s|\s{2,}")
# Suffixes YouTube itself adds to a channel name.
_CHANNEL_NOISE = re.compile(r"\s*-\s*Topic$|\s*VEVO$|\s*Official$", re.IGNORECASE)


def _tidy(text: str) -> str:
return " ".join(text.split())


def search_terms(title: str, uploader: str = "") -> tuple[str, str]:
"""
Turn a video title into the ``(track, artist)`` a lyrics database expects.

Raw YouTube titles are not song names: they carry video furniture, featured
credits and album tags, and the channel is often an aggregator rather than
the artist. Searching with them as-is is why some songs came back empty
while others worked — measured against LRCLIB, the raw form 404s where the
tidied one returns synced lyrics.

The title is trusted over the channel when it names both, because a channel
called "Music n Lyrics" is not who recorded the song.
"""
# Whitespace is collapsed only at the end: a run of two or more spaces is
# itself a separator, and tidying first would erase it.
cleaned = _CREDITS.sub("", _TRAILING_PIPE.sub("", _FURNITURE.sub("", title))).strip()
if not cleaned:
cleaned = title.strip() # all furniture: better to search it than nothing

artist = _CHANNEL_NOISE.sub("", uploader)
halves = _SEPARATOR.split(cleaned, maxsplit=1)
if len(halves) == 2 and halves[0].strip() and halves[1].strip():
artist, cleaned = halves

return _tidy(cleaned).strip(" -–—"), _tidy(artist.split(",")[0])


def index_at(lines: tuple[Line, ...], position: float) -> int:
"""
Index of the line playing at ``position``, or -1 before the first one.
Expand Down
21 changes: 17 additions & 4 deletions tests/test_errors.py
Original file line number Diff line number Diff line change
Expand Up @@ -138,11 +138,24 @@ async def test_a_command_with_its_own_handler_is_left_alone(ctx):
ctx.send.assert_not_awaited()


async def test_an_unexpected_error_is_logged_not_shown(ctx, caplog):
"""Internal failures are the operator's problem, not the user's."""
async def test_an_unexpected_error_is_acknowledged_and_logged(ctx, caplog):
"""
Staying quiet here was a mistake, and a real one: a TypeError in !lyrics was
logged and never answered, so from the channel the bot looked dead. The
details stay in the log, but the user gets told something happened.
"""
await errors.handle(ctx, RuntimeError("something broke internally"))
ctx.send.assert_not_awaited()
assert "something broke internally" in caplog.text

assert "went wrong" in sent_text(ctx).lower()
assert "something broke internally" in caplog.text, "the operator still needs it"


async def test_the_acknowledgement_leaks_no_internals(ctx):
"""A stack trace or an exception message in the channel helps nobody."""
await errors.handle(ctx, RuntimeError("AttributeError at /srv/secret/path.py"))

reply = sent_text(ctx)
assert "AttributeError" not in reply and "/srv" not in reply


async def test_the_original_exception_is_unwrapped(ctx):
Expand Down
150 changes: 150 additions & 0 deletions tests/test_lyrics_cog.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,150 @@
"""
How the command decides what to look up, and what it does with the answer.

The gap these close is a real one: the cog's own class is called ``Lyrics`` and
it shadowed the dataclass of the same name, so the Genius fallback raised
``TypeError: Lyrics.__init__() got an unexpected keyword argument 'title'`` and
the command answered nothing at all. No test reached that path.
"""

from unittest.mock import AsyncMock, MagicMock

import pytest

from cogs.lyrics import Lyrics as LyricsCog
from services import lyrics_api, synced_lyrics

GENIUS_HIT = {
"title": "Every Breath You Take",
"artist": "The Police",
"lyrics": "Every breath you take",
"url": "https://genius.invalid/song",
}


@pytest.fixture
def cog():
return LyricsCog(MagicMock())


@pytest.fixture
def no_lrclib(monkeypatch):
monkeypatch.setattr(synced_lyrics, "fetch", AsyncMock(return_value=None))


# -- the crash that produced silence -----------------------------------

async def test_the_genius_fallback_produces_lyrics(cog, no_lrclib, monkeypatch):
"""This raised TypeError in production, and the user saw nothing at all."""
monkeypatch.setattr(lyrics_api, "fetch", AsyncMock(return_value=GENIUS_HIT))

found = await cog._find("Every Breath You Take", "The Police", None)

assert found.title == "Every Breath You Take"
assert found.artist == "The Police"
assert found.plain == "Every breath you take"
assert found.synced is False


async def test_nothing_anywhere_is_not_an_error(cog, no_lrclib, monkeypatch):
monkeypatch.setattr(lyrics_api, "fetch", AsyncMock(return_value=None))

assert await cog._find("zxqwv", "", None) is None


async def test_lrclib_wins_when_it_has_the_song(cog, monkeypatch):
"""Genius has no timings, so it is only ever the fallback."""
synced = synced_lyrics.Lyrics("T", "A", lines=((0.0, "line"),))
monkeypatch.setattr(synced_lyrics, "fetch", AsyncMock(return_value=synced))
genius = AsyncMock(return_value=GENIUS_HIT)
monkeypatch.setattr(lyrics_api, "fetch", genius)

found = await cog._find("T", "A", None)

assert found.synced is True
genius.assert_not_awaited()


# -- what gets searched for --------------------------------------------

async def test_a_track_is_looked_up_with_tidied_terms(cog, monkeypatch):
"""
The reported failure: the raw video title and the channel name 404 on
LRCLIB, where the tidied pair returns synced lyrics.
"""
seen = {}

async def spy(session, title, artist, duration):
seen.update(title=title, artist=artist, duration=duration)
return None

monkeypatch.setattr(synced_lyrics, "fetch", spy)
monkeypatch.setattr(lyrics_api, "fetch", AsyncMock(return_value=None))

await cog._for_track({
"title": "The Police Every Breath You Take (Lyrics)",
"uploader": "Music n Lyrics",
"duration": 253,
})

assert seen == {"title": "Every Breath You Take",
"artist": "The Police", "duration": 253}


async def test_a_track_with_no_uploader_still_searches(cog, monkeypatch):
seen = {}

async def spy(session, title, artist, duration):
seen.update(title=title, artist=artist)
return None

monkeypatch.setattr(synced_lyrics, "fetch", spy)
monkeypatch.setattr(lyrics_api, "fetch", AsyncMock(return_value=None))

await cog._for_track({"title": "Some Song", "uploader": None, "duration": None})

assert seen == {"title": "Some Song", "artist": ""}


# -- a typed query, either way round -----------------------------------

async def test_a_typed_query_is_tried_both_ways_round(cog, monkeypatch):
"""
The help says `<title> - <artist>`, but `The Police - Every Breath You Take`
is the way people actually type it. Both should work.
"""
tried = []

async def spy(session, title, artist, duration):
tried.append((title, artist))
return synced_lyrics.Lyrics("Every Breath You Take", "The Police",
plain="words") if artist == "The Police" else None

monkeypatch.setattr(synced_lyrics, "fetch", spy)
monkeypatch.setattr(lyrics_api, "fetch", AsyncMock(return_value=None))

found = await cog._search("The Police - Every Breath You Take")

assert found is not None, "the artist-first order found nothing"
assert tried[0] == ("The Police", "Every Breath You Take"), "documented order first"


async def test_the_documented_order_is_not_searched_twice(cog, monkeypatch):
"""A hit on the first try must not cost a second request."""
hit = synced_lyrics.Lyrics("Bohemian Rhapsody", "Queen", plain="words")
fetch = AsyncMock(return_value=hit)
monkeypatch.setattr(synced_lyrics, "fetch", fetch)

await cog._search("Bohemian Rhapsody - Queen")

assert fetch.await_count == 1


async def test_a_query_without_a_separator_is_searched_once(cog, monkeypatch):
fetch = AsyncMock(return_value=None)
monkeypatch.setattr(synced_lyrics, "fetch", fetch)
monkeypatch.setattr(lyrics_api, "fetch", AsyncMock(return_value=None))

await cog._search("Bohemian Rhapsody")

assert fetch.await_count == 1, "there is no other order to try"
Loading
Loading