danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Fix #768: playlist sync editor refusing to match certain tracks

Browse source 
Three compounding bugs hit tracks whose source metadata is YouTube/streaming-
shaped — title "Artist - Song", artist "Official Artist"/"Artist - Topic"/
"ArtistVEVO" (reported: "Arctic Monkeys - Do I Wanna Know?" by "Official Arctic
Monkeys"). Server-agnostic — affects Plex/Jellyfin/Navidrome, not just the
reporter's Navidrome.

Bug A — the match fails. The confidence scorer and the editor's reconcile both
compared the raw "Artist - Song" title against the library's clean "Song"; the
length-ratio penalty + floor drove it to ~0.18 (NO-MATCH), so the track showed
unmatched while its server copy showed as an orphan "extra". New pure
core/text/source_title.py (clean_source_artist / strip_artist_prefix /
canonical_source_track) strips the channel/video decoration, applied at BOTH
matching seams: services/sync_service._find_track_in_media_server (tries raw
then canonical, keeps the best) and the editor reconcile. Conservative: a title
prefix is stripped only when it equals the artist, so "Self-Titled", "Jay-Z",
and "Marvin Gaye" (by another artist) are untouched, and the canonical form is
an additional best-of candidate so it can only help.

Bug B — manual matches never persisted. get_server_playlist_tracks built the
per-source entry WITHOUT source_track_id, so "Find & add" posted an empty id
and _persist_find_and_add_match returned early. The match reverted to "extra"
on reload and re-adding looped. The editor's 3-pass matcher is now lifted to a
pure, tested core.sync.playlist_reconcile.reconcile_playlist that includes
source_track_id (the frontend at pages-extra.js:1836 already reads + sends it).

Bug C — manual match duplicated + delete wiped all copies. "Find & add" always
inserted, so linking a source to an already-present server track appended a
duplicate (pos 72, 73...); remove filtered out EVERY entry with the target id.
New pure core.sync.playlist_edit (plan_playlist_add: link-don't-duplicate when
the target is already present; remove_one_occurrence: drop a single copy) wired
into the Plex/Jellyfin/Navidrome add + remove branches.

Tests (extreme): tests/test_source_title.py (35), tests/test_playlist_reconcile.py
(11 — incl. the reported case, parity for override/exact/fuzzy/extra, and
duplicate-server handling), tests/test_playlist_edit.py (12). 286 matching/sync
tests still pass.

Caveats: the sync_service change and the add/remove/editor endpoints are
read-verified, not executed against a live media server (none in CI). The pure
cores they call are exhaustively unit-tested; output-shape parity of the
reconcile lift is covered. Delete removes the first matching copy (duplicates
are identical, so harmless).
This commit is contained in:
BoulderBadgeDad 2026-06-02 10:16:21 -07:00
parent 174513d351
commit bba0836324
8 changed files with 761 additions and 128 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

74
core/sync/playlist_edit.py Normal file
View file

@ -0,0 +1,74 @@
"""Pure planners for sync-editor playlist mutations (#768 Bug C).
The sync editor's "Find & add" and remove actions rewrite the whole server
playlist from a flat list of track IDs (Subsonic/Navidrome + Jellyfin have no
position-level ops). Two bugs lived in the inline endpoint logic:
* **Duplicate on manual match.** "Find & add" always *inserted* the chosen
track — but when the user is matching an UNMATCHED source to a server track
that's already in the playlist (an orphan "extra"), the intent is to LINK
them, not add a second copy. Each attempt appended another duplicate
(positions 72, 73, 74…). ``plan_playlist_add`` skips the insert when it's a
link to an already-present track (the caller still persists the override).
* **Delete removes ALL copies.** The inline remove filtered out *every* entry
with the target ID. With duplicates present, deleting one removed them all.
``remove_one_occurrence`` drops a single entry (duplicates are the same
track, so removing any one is correct).
Pure, no I/O — the caller fetches the current track-id list and applies the
returned plan to the media-server client.
"""
from __future__ import annotations
from typing import List, Optional, Tuple
def plan_playlist_add(
current_ids: List[str],
track_id: str,
*,
is_link: bool,
position: Optional[int] = None,
) -> dict:
"""Plan a "Find & add" against a flat track-id playlist.
``is_link`` is True when the add carries a ``source_track_id`` (i.e. the
user is matching an unmatched source to this server track). In that case,
if the track is ALREADY in the playlist, return ``should_insert=False`` so
the caller only records the override and never duplicates it.
Returns ``{'should_insert': bool, 'new_ids': [...]}``. ``new_ids`` equals
the input (stringified) when no insert is needed."""
tid = str(track_id)
current = [str(t) for t in current_ids]
if is_link and tid in current:
return {"should_insert": False, "new_ids": current}
pos = len(current) if position is None else max(0, min(int(position), len(current)))
new_ids = current[:pos] + [tid] + current[pos:]
return {"should_insert": True, "new_ids": new_ids}
def remove_one_occurrence(
track_ids: List[str],
target_id: str,
position: Optional[int] = None,
) -> Tuple[List[str], bool]:
"""Remove a SINGLE occurrence of ``target_id`` from a flat id list.
If ``position`` is given and the id there matches, that exact entry is
removed (so the user removes the row they clicked); otherwise the first
matching id is removed. Returns ``(new_ids, removed)``. ``removed`` is
False when the id isn't present (caller should 404)."""
target = str(target_id)
ids = [str(t) for t in track_ids]
if position is not None and 0 <= position < len(ids) and ids[position] == target:
return ids[:position] + ids[position + 1:], True
for idx, tid in enumerate(ids):
if tid == target:
return ids[:idx] + ids[idx + 1:], True
return ids, False
__all__ = ["plan_playlist_add", "remove_one_occurrence"]

186
core/sync/playlist_reconcile.py Normal file
View file

@ -0,0 +1,186 @@
"""Reconcile a source playlist against a media-server playlist (pure).
Lifted verbatim from the inline three-pass matcher in
``web_server.get_server_playlist_tracks`` so it can be unit-tested and so the
two #768 fixes live in importable, covered code:
Pass 0 user-confirmed overrides (``sync_match_cache``), applied first.
Pass 1 exact normalized-title match.
Pass 2 fuzzy match on ``"artist title"`` (SequenceMatcher >= 0.75).
Extra server tracks no source claimed -> ``match_status='extra'``.
Two bug fixes over the original inline version:
* #768 Bug A — the source side is YouTube/streaming-shaped (title
``"Artist - Song"``, artist ``"Official Artist"``). The original passes
compared the raw title, so ``"Arctic Monkeys - Do I Wanna Know?"`` never
matched the library's ``"Do I Wanna Know?"`` and the track showed as
unmatched while its server copy showed as an orphan "extra". We now also try
the canonicalized source title/artist (see ``core.text.source_title``).
* #768 Bug B — the original built the per-source ``src_entry`` WITHOUT
``source_track_id``, so the editor UI never received it; "Find & add" then
posted an empty id and the manual match was never persisted (it reverted to
"extra" on reload, and re-adding duplicated the track). ``src_entry`` now
carries ``source_track_id``.
Pure, no I/O. ``override_pairs`` (``{source_idx: server_idx}``) is computed by
the caller via ``core.sync.match_overrides.resolve_match_overrides`` so the DB
lookup stays out of this module.
"""
from __future__ import annotations
import re
from difflib import SequenceMatcher
from typing import Any, Dict, List, Optional
from core.text.source_title import canonical_source_track
_FUZZY_THRESHOLD = 0.75
_FEAT_RE = re.compile(r'\s*[\(\[](?:feat|ft)\.?[^\)\]]*[\)\]]', re.IGNORECASE)
_REMASTER_RE = re.compile(
r'\s*[\(\[](?:\d{4}\s+)?remaster(?:ed)?(?:\s+version)?\s*[\)\]]', re.IGNORECASE)
_EDITION_RE = re.compile(
r'\s*[\(\[](?:deluxe|special|anniversary|legacy|expanded|limited)(?:\s+edition)?\s*[\)\]]',
re.IGNORECASE)
def norm_title(t: str) -> str:
"""Strip feat./ft., remaster, and edition qualifiers for comparison only.
Byte-faithful to web_server's inline ``_norm_title``."""
t = _FEAT_RE.sub('', t or '')
t = _REMASTER_RE.sub('', t)
t = _EDITION_RE.sub('', t)
return t.lower().strip()
def _src_entry(src: Dict[str, Any], position_fallback: int) -> Dict[str, Any]:
return {
'name': src.get('name', ''),
'artist': src.get('artist', ''),
'album': src.get('album', ''),
'image_url': src.get('image_url', ''),
'duration_ms': src.get('duration_ms', 0),
'position': src.get('position', position_fallback),
# #768 Bug B: echo the source id back so the editor can persist a
# manual "Find & add" override against it.
'source_track_id': src.get('source_track_id', '') or '',
}
def _resolved_artist(src: Dict[str, Any]) -> str:
"""Artist string, falling back to the first of an ``artists`` list."""
artist = src.get('artist', '')
if not artist and src.get('artists'):
a = src['artists'][0] if src['artists'] else ''
artist = a.get('name', a) if isinstance(a, dict) else str(a)
return artist or ''
def reconcile_playlist(
source_tracks: List[Dict[str, Any]],
server_tracks: List[Dict[str, Any]],
override_pairs: Optional[Dict[int, int]] = None,
) -> List[Dict[str, Any]]:
"""Return the combined matched/missing/extra view (list of dicts).
Each combined entry has ``source_track``, ``server_track``,
``match_status`` ('matched'|'missing'|'extra'), ``confidence``, and
``override: True`` on override hits."""
override_pairs = override_pairs or {}
combined: List[Dict[str, Any]] = []
used_server_indices: set[int] = set()
unmatched_source: List[tuple[int, Dict[str, Any], str]] = []
# Precompute normalized server titles once.
server_norm = [norm_title(svr.get('title', '')) for svr in server_tracks]
for i, src in enumerate(source_tracks):
src_artist = _resolved_artist(src)
src_name = src.get('name', '')
src_entry = _src_entry({**src, 'artist': src_artist}, i)
# Pass 0: user-confirmed override.
if i in override_pairs:
j = override_pairs[i]
used_server_indices.add(j)
combined.append({
'source_track': src_entry,
'server_track': server_tracks[j],
'match_status': 'matched',
'confidence': 1.0,
'override': True,
})
continue
# Pass 1: exact normalized-title match — try the raw source title AND
# the canonicalized one (strips "Artist - " prefix / channel artist).
canon_title, _canon_artist = canonical_source_track(src_name, src_artist)
candidates = {norm_title(src_name), norm_title(canon_title)}
best_idx = -1
for j, svr_norm in enumerate(server_norm):
if j in used_server_indices:
continue
if svr_norm in candidates:
best_idx = j
break
if best_idx >= 0:
used_server_indices.add(best_idx)
combined.append({
'source_track': src_entry,
'server_track': server_tracks[best_idx],
'match_status': 'matched',
'confidence': 1.0,
})
else:
idx = len(combined)
combined.append({
'source_track': src_entry,
'server_track': None,
'match_status': 'missing',
'confidence': 0.0,
})
# Carry the canonical artist for the fuzzy pass.
unmatched_source.append((idx, src_entry, _canon_artist or src_artist))
# Pass 2: fuzzy match on remaining unmatched source tracks. Build the key
# from the canonicalized title/artist so YouTube-shaped sources can pair.
for combo_idx, src_entry, canon_artist in unmatched_source:
canon_title, _ = canonical_source_track(src_entry['name'], src_entry['artist'])
src_key = f"{canon_artist} {norm_title(canon_title)}".strip().lower()
best_score = 0.0
best_j = -1
for j, svr in enumerate(server_tracks):
if j in used_server_indices:
continue
svr_key = f"{svr.get('artist', '')} {norm_title(svr.get('title', ''))}".strip().lower()
score = SequenceMatcher(None, src_key, svr_key).ratio()
if score > best_score and score >= _FUZZY_THRESHOLD:
best_score = score
best_j = j
if best_j >= 0:
used_server_indices.add(best_j)
combined[combo_idx] = {
'source_track': src_entry,
'server_track': server_tracks[best_j],
'match_status': 'matched',
'confidence': round(best_score, 3),
}
# Extra: server tracks no source claimed.
for j, svr in enumerate(server_tracks):
if j not in used_server_indices:
combined.append({
'source_track': None,
'server_track': svr,
'match_status': 'extra',
'confidence': 0.0,
})
return combined
__all__ = ["reconcile_playlist", "norm_title"]

105
core/text/source_title.py Normal file
View file

@ -0,0 +1,105 @@
"""Normalize streaming/YouTube-style source track metadata for matching.
Issue #768: source playlists — especially ones seeded from YouTube — carry
video-style metadata: the title is ``"Artist - Song"`` and the artist is a
channel name like ``"Official Arctic Monkeys"``, ``"Arctic Monkeys - Topic"``,
or ``"ColdplayVEVO"``. The library/media-server side has the clean ``"Song"`` /
``"Arctic Monkeys"``. Both matching paths (the sync confidence scorer and the
playlist-editor reconcile) then fail to pair them — the track is reported
"not matched" / shows up as an orphan "extra" even though it exists.
These helpers strip that channel/video decoration so the cleaned source can be
compared against the clean library metadata. Pure, no I/O.
Conservative by construction:
- ``strip_artist_prefix`` removes a leading ``"<artist><sep>"`` only when the
prefix EQUALS the artist we're matching against. So ``"Death - Pull the
Plug"`` by ``"Death"`` is helped, while ``"Marvin Gaye"`` by Charlie Puth
(title is not ``"Charlie Puth - ..."``) is left untouched, and a hyphenated
word like ``"Self-Titled"`` is never split (a separator needs surrounding
whitespace, or a colon).
- ``clean_source_artist`` only removes well-known channel decorations.
Both are intended to be applied as ADDITIONAL match candidates (best-of), so
an over-eager strip can only add a comparison, never remove the original.
"""
from __future__ import annotations
import re
from core.text.normalize import normalize_for_comparison
# Artist/title separator: a dash/pipe/tilde flanked by whitespace, OR a colon
# (with optional trailing space). Whitespace-flanking keeps "Self-Titled" and
# "Jay-Z" intact while still splitting "Artist - Title".
_SEP_SPLIT = re.compile(r"\s+[-–—|~]\s+|\s*:\s+")
# YouTube auto-generated artist channel: "Arctic Monkeys - Topic".
_TOPIC_SUFFIX = re.compile(r"\s*-\s*topic\s*$", re.IGNORECASE)
# "Official " / "The Official " channel prefix.
_OFFICIAL_PREFIX = re.compile(r"^\s*(?:the\s+)?official\s+", re.IGNORECASE)
# Trailing VEVO, attached ("ColdplayVEVO") or spaced ("Coldplay VEVO").
_VEVO_SUFFIX = re.compile(r"\s*vevo\s*$", re.IGNORECASE)
def clean_source_artist(artist: str) -> str:
"""Strip well-known streaming-channel decoration from an artist name.
``"Official Arctic Monkeys"`` → ``"Arctic Monkeys"``;
``"Arctic Monkeys - Topic"`` → ``"Arctic Monkeys"``;
``"ColdplayVEVO"`` → ``"Coldplay"``. Returns the input unchanged when
nothing matches, and never returns empty for non-empty input."""
if not artist:
return artist
s = artist.strip()
topic = _TOPIC_SUFFIX.sub("", s).strip()
if topic and topic != s:
s = topic
official = _OFFICIAL_PREFIX.sub("", s).strip()
if official:
s = official
# Only strip VEVO if at least 2 chars of name remain (don't empty "VEVO").
vevo = _VEVO_SUFFIX.sub("", s).strip()
if len(vevo) >= 2 and vevo != s:
s = vevo
return s or artist
def strip_artist_prefix(title: str, artist: str) -> str:
"""Remove a leading ``"<artist><separator>"`` from ``title`` when the prefix
equals ``artist`` (accent/case-folded). Otherwise return ``title`` unchanged.
``("Arctic Monkeys - Do I Wanna Know?", "Arctic Monkeys")`` → ``"Do I Wanna
Know?"``. Never returns an empty string."""
if not title or not artist:
return title
na = normalize_for_comparison(artist)
if not na:
return title
parts = _SEP_SPLIT.split(title, maxsplit=1)
if len(parts) == 2:
left, right = parts
right = right.strip()
if right and normalize_for_comparison(left) == na:
return right
return title
def canonical_source_track(title: str, artist: str) -> tuple[str, str]:
"""Best-effort clean (title, artist) for matching a streaming/YouTube source
against clean library metadata. Cleans the artist first, then strips a
leading artist prefix from the title using EITHER the cleaned or the raw
artist (YouTube titles prepend the real artist, not the channel name)."""
cleaned_artist = clean_source_artist(artist)
new_title = strip_artist_prefix(title, cleaned_artist)
if new_title == title and cleaned_artist != artist:
new_title = strip_artist_prefix(title, artist)
return new_title, cleaned_artist
__all__ = ["clean_source_artist", "strip_artist_prefix", "canonical_source_track"]

34
services/sync_service.py
View file

@ -586,17 +586,35 @@ class PlaylistSyncService:
artist_name = _artist_name(artist)
# YouTube/streaming sources arrive video-shaped: title
# "Artist - Song", artist "Official Artist"/"Artist - Topic"/
# "ArtistVEVO". Try the raw (title, artist) first, then a
# canonicalized variant so those still match the clean library
# metadata instead of being reported missing (#768).
from core.text.source_title import canonical_source_track
_attempts = [(original_title, artist_name)]
_canon_title, _canon_artist = canonical_source_track(original_title, artist_name)
if (_canon_title, _canon_artist) != (original_title, artist_name):
_attempts.append((_canon_title, _canon_artist))
# Use the improved database check_track_exists method with server awareness
try:
db = MusicDatabase()
artist_candidates = self._get_or_fetch_artist_candidates(
candidate_pool, db, artist_name, active_server,
)
db_track, confidence = db.check_track_exists(
original_title, artist_name,
confidence_threshold=0.7, server_source=active_server,
candidate_tracks=artist_candidates,
)
# Try every candidate form and keep the BEST — don't stop at
# the first that clears the threshold, or a marginal raw
# match could mask a stronger (correct) canonical one.
db_track, confidence = None, 0.0
for _try_title, _try_artist in _attempts:
artist_candidates = self._get_or_fetch_artist_candidates(
candidate_pool, db, _try_artist, active_server,
)
_cand_track, _cand_conf = db.check_track_exists(
_try_title, _try_artist,
confidence_threshold=0.7, server_source=active_server,
candidate_tracks=artist_candidates,
)
if _cand_conf > confidence:
db_track, confidence = _cand_track, _cand_conf
if db_track and confidence >= 0.7:
logger.debug(f"Database match found for '{original_title}' by '{artist_name}': '{db_track.title}' with confidence {confidence:.2f}")

80
tests/test_playlist_edit.py Normal file
View file

@ -0,0 +1,80 @@
"""Extreme battery for sync-editor add/remove planners (#768 Bug C)."""
from __future__ import annotations
from core.sync.playlist_edit import plan_playlist_add, remove_one_occurrence
# ── plan_playlist_add: link must not duplicate ────────────────────────────
def test_link_to_existing_track_does_not_insert():
# The reported loop: matching an unmatched source to a track already in
# the playlist (an "extra") must NOT add a second copy.
plan = plan_playlist_add(["a", "b", "nv72"], "nv72", is_link=True)
assert plan["should_insert"] is False
assert plan["new_ids"] == ["a", "b", "nv72"] # unchanged
def test_link_to_absent_track_inserts():
plan = plan_playlist_add(["a", "b"], "nv99", is_link=True, position=1)
assert plan["should_insert"] is True
assert plan["new_ids"] == ["a", "nv99", "b"]
def test_non_link_add_always_inserts_even_if_present():
# A plain add (no source link) may legitimately duplicate.
plan = plan_playlist_add(["a", "b"], "a", is_link=False)
assert plan["should_insert"] is True
assert plan["new_ids"].count("a") == 2
def test_add_appends_when_no_position():
plan = plan_playlist_add(["a", "b"], "c", is_link=False)
assert plan["new_ids"] == ["a", "b", "c"]
def test_add_clamps_out_of_range_position():
assert plan_playlist_add(["a"], "c", is_link=False, position=99)["new_ids"] == ["a", "c"]
assert plan_playlist_add(["a"], "c", is_link=False, position=-5)["new_ids"] == ["c", "a"]
def test_add_stringifies_ids():
plan = plan_playlist_add([1, 2, 72], 72, is_link=True)
assert plan["should_insert"] is False
# ── remove_one_occurrence: remove ONE, not all ────────────────────────────
def test_removes_only_one_of_duplicates():
# The #768 delete bug: two copies (pos 72, 73) — removing must drop ONE.
new_ids, removed = remove_one_occurrence(["a", "nv72", "nv72", "b"], "nv72")
assert removed is True
assert new_ids == ["a", "nv72", "b"] # one copy survives
def test_removes_exact_position_when_given():
new_ids, removed = remove_one_occurrence(["x", "x", "x"], "x", position=1)
assert removed is True
assert new_ids == ["x", "x"]
def test_falls_back_to_first_when_position_mismatches():
new_ids, removed = remove_one_occurrence(["a", "b", "c"], "b", position=0)
assert removed is True
assert new_ids == ["a", "c"]
def test_remove_absent_id_reports_not_removed():
new_ids, removed = remove_one_occurrence(["a", "b"], "zzz")
assert removed is False
assert new_ids == ["a", "b"]
def test_remove_single_occurrence():
new_ids, removed = remove_one_occurrence(["a", "b", "c"], "b")
assert (new_ids, removed) == (["a", "c"], True)
def test_remove_stringifies():
new_ids, removed = remove_one_occurrence([1, 2, 2, 3], 2)
assert removed and new_ids == ["1", "2", "3"]

122
tests/test_playlist_reconcile.py Normal file
View file

@ -0,0 +1,122 @@
"""Extreme battery for the playlist sync-editor reconcile (#768).
Covers: the reported YouTube failure (Bug A — "Artist - Title" source matching
its clean server copy instead of showing unmatched + orphan extra), the
source_track_id echo (Bug B), and parity with the original three-pass behavior
(override → exact → fuzzy → extra), plus duplicate-server-track handling.
"""
from __future__ import annotations
from core.sync.playlist_reconcile import norm_title, reconcile_playlist
def _src(name, artist, sid="", **kw):
return {"name": name, "artist": artist, "source_track_id": sid, **kw}
def _svr(title, artist, tid):
return {"title": title, "artist": artist, "id": tid, "ratingKey": tid}
def _status(combined):
return [(c["match_status"],
(c["source_track"] or {}).get("name"),
(c["server_track"] or {}).get("title")) for c in combined]
# ── Bug A: the reported YouTube case ──────────────────────────────────────
def test_youtube_artist_title_source_matches_clean_server_track():
source = [_src("Arctic Monkeys - Do I Wanna Know?", "Official Arctic Monkeys", "sp1")]
server = [_svr("Do I Wanna Know?", "Arctic Monkeys", "nv72")]
combined = reconcile_playlist(source, server)
assert len(combined) == 1
assert combined[0]["match_status"] == "matched"
assert combined[0]["server_track"]["id"] == "nv72"
# ...and the server track is NOT left as an orphan extra.
assert not any(c["match_status"] == "extra" for c in combined)
def test_youtube_match_does_not_leave_unmatched_or_extra():
# Before the fix this produced one 'missing' + one 'extra'.
source = [_src("The Killers - Mr. Brightside", "The KillersVEVO", "sp2")]
server = [_svr("Mr. Brightside", "The Killers", "nv5")]
statuses = [c["match_status"] for c in reconcile_playlist(source, server)]
assert statuses == ["matched"]
# ── Bug B: source_track_id is echoed back ─────────────────────────────────
def test_source_track_id_present_on_matched_entry():
source = [_src("Do I Wanna Know?", "Arctic Monkeys", "spotify:track:abc")]
server = [_svr("Do I Wanna Know?", "Arctic Monkeys", "nv1")]
combined = reconcile_playlist(source, server)
assert combined[0]["source_track"]["source_track_id"] == "spotify:track:abc"
def test_source_track_id_present_on_missing_entry():
# A genuinely-missing source must still carry its id so it can be
# manually matched and persisted (the #768 manual-match loop).
source = [_src("Some Obscure B-Side", "Some Artist", "spotify:track:xyz")]
server = [_svr("Completely Different", "Other Artist", "nv9")]
combined = reconcile_playlist(source, server)
missing = [c for c in combined if c["match_status"] == "missing"]
assert missing and missing[0]["source_track"]["source_track_id"] == "spotify:track:xyz"
# ── parity: override / exact / fuzzy / extra ──────────────────────────────
def test_override_pair_wins_first():
source = [_src("Anything", "Whoever", "s1")]
server = [_svr("Totally Different Title", "Nobody", "nvX")]
combined = reconcile_playlist(source, server, override_pairs={0: 0})
assert combined[0]["match_status"] == "matched"
assert combined[0]["confidence"] == 1.0
assert combined[0].get("override") is True
def test_exact_normalized_match_strips_feat():
source = [_src("Stay (feat. Justin Bieber)", "The Kid LAROI", "s1")]
server = [_svr("Stay", "The Kid LAROI", "nv1")]
assert reconcile_playlist(source, server)[0]["match_status"] == "matched"
def test_fuzzy_match_above_threshold():
source = [_src("Mr Brightside", "The Killers", "s1")]
server = [_svr("Mr. Brightside", "The Killers", "nv1")]
c = reconcile_playlist(source, server)[0]
assert c["match_status"] == "matched"
assert c["confidence"] >= 0.75
def test_truly_absent_track_is_missing_and_unrelated_server_is_extra():
source = [_src("Nonexistent Song", "Ghost Artist", "s1")]
server = [_svr("Yellow", "Coldplay", "nv1")]
statuses = sorted(c["match_status"] for c in reconcile_playlist(source, server))
assert statuses == ["extra", "missing"]
def test_each_server_track_claimed_once_no_double_match():
# Two identical source rows must not both claim the single server track.
source = [_src("Yellow", "Coldplay", "s1"), _src("Yellow", "Coldplay", "s2")]
server = [_svr("Yellow", "Coldplay", "nv1")]
combined = reconcile_playlist(source, server)
matched = [c for c in combined if c["match_status"] == "matched"]
missing = [c for c in combined if c["match_status"] == "missing"]
assert len(matched) == 1 and len(missing) == 1
def test_duplicate_server_tracks_one_matched_one_extra():
# The #768 duplicate scenario: two copies of the same track on the server.
source = [_src("Do I Wanna Know?", "Arctic Monkeys", "s1")]
server = [_svr("Do I Wanna Know?", "Arctic Monkeys", "nv72"),
_svr("Do I Wanna Know?", "Arctic Monkeys", "nv73")]
combined = reconcile_playlist(source, server)
assert sorted(c["match_status"] for c in combined) == ["extra", "matched"]
def test_norm_title_helper_parity():
assert norm_title("Stay (feat. X)") == "stay"
assert norm_title("Song (2019 Remaster)") == "song"
assert norm_title("Album (Deluxe Edition)") == "album"

113
tests/test_source_title.py Normal file
View file

@ -0,0 +1,113 @@
"""Extreme test battery for source-track normalization (#768).
YouTube/streaming sources carry "Artist - Song" titles and "Official Artist"/
"Artist - Topic"/"ArtistVEVO" artist names; the library has clean metadata, so
matching fails and tracks are reported missing. These helpers strip the
decoration. The batteries below pin both the positives (must clean) and the
negatives (must NOT mangle real titles/artists).
"""
from __future__ import annotations
import pytest
from core.text.source_title import (
canonical_source_track,
clean_source_artist,
strip_artist_prefix,
)
# ── clean_source_artist ───────────────────────────────────────────────────
@pytest.mark.parametrize("raw,expected", [
("Official Arctic Monkeys", "Arctic Monkeys"),
("The Official Weeknd", "Weeknd"),
("Arctic Monkeys - Topic", "Arctic Monkeys"),
("Coldplay - Topic", "Coldplay"),
("ColdplayVEVO", "Coldplay"),
("Coldplay VEVO", "Coldplay"),
("EminemVEVO", "Eminem"),
(" Official Radiohead ", "Radiohead"),
])
def test_clean_source_artist_strips_decoration(raw, expected):
assert clean_source_artist(raw) == expected
@pytest.mark.parametrize("raw", [
"Arctic Monkeys", # already clean
"Coldplay",
"Twenty One Pilots",
"Death",
"U2",
"AJR", # would be emptied by a naive vevo/official strip
"",
])
def test_clean_source_artist_leaves_clean_names(raw):
assert clean_source_artist(raw) == raw
def test_clean_source_artist_never_empties():
# Pathological: artist that is ONLY decoration must not become "".
assert clean_source_artist("VEVO") == "VEVO"
assert clean_source_artist("Official ") == "Official"
# ── strip_artist_prefix ───────────────────────────────────────────────────
@pytest.mark.parametrize("title,artist,expected", [
("Arctic Monkeys - Do I Wanna Know?", "Arctic Monkeys", "Do I Wanna Know?"),
("Death - Pull the Plug", "Death", "Pull the Plug"),
("Coldplay – Yellow", "Coldplay", "Yellow"), # en dash
("Coldplay — Yellow", "Coldplay", "Yellow"), # em dash
("Eminem: Lose Yourself", "Eminem", "Lose Yourself"), # colon
("Daft Punk | Get Lucky", "Daft Punk", "Get Lucky"), # pipe
("ARCTIC MONKEYS - 505", "arctic monkeys", "505"), # case-fold
])
def test_strip_artist_prefix_strips_when_prefix_is_artist(title, artist, expected):
assert strip_artist_prefix(title, artist) == expected
@pytest.mark.parametrize("title,artist", [
("Marvin Gaye", "Charlie Puth"), # title is not "artist - ..."
("Do I Wanna Know?", "Arctic Monkeys"), # already clean
("Self-Titled", "Whoever"), # hyphen w/o spaces — not a sep
("Jay-Z Anthem", "Somebody"), # hyphen inside a word
("Song - Live", "Coldplay"), # prefix "Song" != artist
("Stay With Me", "Sam Smith"),
("", "Arctic Monkeys"),
("Arctic Monkeys -", "Arctic Monkeys"), # nothing after sep -> unchanged
])
def test_strip_artist_prefix_leaves_others_untouched(title, artist):
assert strip_artist_prefix(title, artist) == title
def test_strip_only_first_separator():
# "Artist - Song - Remix" -> strip only the leading artist segment.
assert strip_artist_prefix("Gorillaz - Feel Good Inc - Remix", "Gorillaz") == "Feel Good Inc - Remix"
# ── canonical_source_track (combined) ─────────────────────────────────────
def test_canonical_handles_youtube_channel_and_prefix():
# The reported case: channel-name artist + "Artist - Title" title.
title, artist = canonical_source_track(
"Arctic Monkeys - Do I Wanna Know?", "Official Arctic Monkeys",
)
assert title == "Do I Wanna Know?"
assert artist == "Arctic Monkeys"
def test_canonical_strips_prefix_using_raw_artist_when_clean_differs():
# Title prefixed with the channel-style raw artist itself.
title, artist = canonical_source_track(
"Official Arctic Monkeys - 505", "Official Arctic Monkeys",
)
# cleaned artist is "Arctic Monkeys"; raw prefix "Official Arctic Monkeys"
# also stripped via the raw-artist fallback.
assert title == "505"
assert artist == "Arctic Monkeys"
def test_canonical_noop_on_clean_input():
assert canonical_source_track("Yellow", "Coldplay") == ("Yellow", "Coldplay")

175
web_server.py
View file

@ -18185,23 +18185,11 @@ def get_server_playlist_tracks(playlist_id):
pass
break
# Build combined view with two-pass matching (exact then fuzzy)
import re as _re
from difflib import SequenceMatcher
def _norm_title(t):
"""Strip feat./ft., remaster, and edition qualifiers for comparison only."""
# feat./ft. — e.g. (feat. Artist), [ft. Artist]
t = _re.sub(r'\s*[\(\[](?:feat|ft)\.?[^\)\]]*[\)\]]', '', t, flags=_re.IGNORECASE)
# Remaster/Remastered — e.g. (2019 Remaster), (Remastered), (2019 Remastered Version)
t = _re.sub(r'\s*[\(\[](?:\d{4}\s+)?remaster(?:ed)?(?:\s+version)?\s*[\)\]]', '', t, flags=_re.IGNORECASE)
# Edition qualifiers — e.g. (Deluxe Edition), (Special Edition), [Anniversary Edition]
t = _re.sub(r'\s*[\(\[](?:deluxe|special|anniversary|legacy|expanded|limited)(?:\s+edition)?\s*[\)\]]', '', t, flags=_re.IGNORECASE)
return t.lower().strip()
combined = []
used_server_indices = set()
unmatched_source = [] # (index_in_combined, src_dict) for fuzzy second pass
# Reconcile source vs server playlist. Three-pass matcher lifted to
# core.sync.playlist_reconcile (pure + tested) — fixes #768 (YouTube
# "Artist - Title" sources now match, and source_track_id is echoed
# back so manual "Find & add" overrides persist).
from core.sync.playlist_reconcile import reconcile_playlist
# Pass 0: User-confirmed match overrides from sync_match_cache.
# When a user previously picked a local file via "Find & Add",
@ -18218,92 +18206,7 @@ def get_server_playlist_tracks(playlist_id):
lambda src_id: ((_db_for_overrides.read_sync_match_cache(src_id, active_server) or {}).get('server_track_id')),
)
# Pass 1: Exact title match (normalized — strips feat./ft. qualifiers)
for i, src in enumerate(source_tracks):
src_name = src.get('name', '')
src_artist = src.get('artist', '')
if not src_artist and src.get('artists'):
a = src['artists'][0] if src['artists'] else ''
src_artist = a.get('name', a) if isinstance(a, dict) else str(a)
src_entry = {
'name': src_name, 'artist': src_artist,
'album': src.get('album', ''), 'image_url': src.get('image_url', ''),
'duration_ms': src.get('duration_ms', 0), 'position': src.get('position', i),
}
# Override hit — paired by user, skip exact/fuzzy matching.
if i in _override_pairs:
j_override = _override_pairs[i]
used_server_indices.add(j_override)
combined.append({
'source_track': src_entry,
'server_track': server_tracks[j_override],
'match_status': 'matched',
'confidence': 1.0,
'override': True,
})
continue
src_norm = _norm_title(src_name)
best_idx = -1
for j, svr in enumerate(server_tracks):
if j in used_server_indices:
continue
if _norm_title(svr['title']) == src_norm:
best_idx = j
break
if best_idx >= 0:
used_server_indices.add(best_idx)
combined.append({
'source_track': src_entry,
'server_track': server_tracks[best_idx],
'match_status': 'matched',
'confidence': 1.0,
})
else:
idx = len(combined)
combined.append({
'source_track': src_entry,
'server_track': None,
'match_status': 'missing',
'confidence': 0.0,
})
unmatched_source.append((idx, src_entry))
# Pass 2: Fuzzy match on remaining unmatched source tracks (normalized keys)
for combo_idx, src_entry in unmatched_source:
src_key = f"{src_entry['artist']} {_norm_title(src_entry['name'])}".strip()
best_score = 0.0
best_j = -1
for j, svr in enumerate(server_tracks):
if j in used_server_indices:
continue
svr_key = f"{svr['artist']} {_norm_title(svr['title'])}".strip().lower()
score = SequenceMatcher(None, src_key.lower(), svr_key).ratio()
if score > best_score and score >= 0.75:
best_score = score
best_j = j
if best_j >= 0:
used_server_indices.add(best_j)
combined[combo_idx] = {
'source_track': src_entry,
'server_track': server_tracks[best_j],
'match_status': 'matched',
'confidence': round(best_score, 3),
}
# Add server tracks that aren't in the source (extra tracks on server)
for j, svr in enumerate(server_tracks):
if j not in used_server_indices:
combined.append({
'source_track': None,
'server_track': svr,
'match_status': 'extra',
'confidence': 0.0,
})
combined = reconcile_playlist(source_tracks, server_tracks, _override_pairs)
return jsonify({
"success": True,
@ -18490,6 +18393,18 @@ def server_playlist_add_track(playlist_id):
if not new_item:
return jsonify({"success": False, "error": "Track not found on server"}), 404
# Link, don't duplicate: matching an unmatched source to a track
# already in the playlist should only record the override, never
# append a second copy (#768).
if source_track_id:
try:
_existing = {str(it.ratingKey) for it in raw_playlist.items()}
except Exception:
_existing = set()
if str(track_id) in _existing:
_persist_find_and_add_match(source_track_id, active_server, track_id, server_track_title or new_item.title, source_title, source_artist)
return jsonify({"success": True, "message": "Track linked"})
logger.info(f"[ServerPlaylist] Adding track: '{new_item.title}' (ratingKey={new_item.ratingKey}) to playlist '{playlist_name}'")
raw_playlist.addItems([new_item])
@ -18515,24 +18430,30 @@ def server_playlist_add_track(playlist_id):
return jsonify({"success": True, "message": "Track added", "new_playlist_id": new_id})
elif active_server == 'jellyfin' and media_server_engine.client('jellyfin'):
from core.sync.playlist_edit import plan_playlist_add
current_tracks = media_server_engine.client('jellyfin').get_playlist_tracks(playlist_id) or []
track_ids = [str(t.ratingKey) for t in current_tracks]
pos = max(0, min(int(position), len(track_ids))) if position is not None else len(track_ids)
track_ids.insert(pos, track_id)
new_track_objs = [type('T', (), {'ratingKey': tid, 'title': ''})() for tid in track_ids]
media_server_engine.client('jellyfin').update_playlist(playlist_name, new_track_objs)
# Matching an unmatched source to a track already in the playlist
# is a LINK, not a second copy — don't duplicate it (#768).
plan = plan_playlist_add(track_ids, track_id, is_link=bool(source_track_id), position=position)
if plan['should_insert']:
new_track_objs = [type('T', (), {'ratingKey': tid, 'title': ''})() for tid in plan['new_ids']]
media_server_engine.client('jellyfin').update_playlist(playlist_name, new_track_objs)
_persist_find_and_add_match(source_track_id, active_server, track_id, server_track_title, source_title, source_artist)
return jsonify({"success": True, "message": "Track added"})
return jsonify({"success": True, "message": "Track linked" if not plan['should_insert'] else "Track added"})
elif active_server == 'navidrome' and media_server_engine.client('navidrome'):
from core.sync.playlist_edit import plan_playlist_add
current_tracks = media_server_engine.client('navidrome').get_playlist_tracks(playlist_id) or []
track_ids = [str(t.ratingKey) for t in current_tracks]
pos = max(0, min(int(position), len(track_ids))) if position is not None else len(track_ids)
track_ids.insert(pos, track_id)
new_track_objs = [type('T', (), {'ratingKey': tid, 'title': ''})() for tid in track_ids]
media_server_engine.client('navidrome').create_playlist(playlist_name, new_track_objs, playlist_id=playlist_id)
# Matching an unmatched source to a track already in the playlist
# is a LINK, not a second copy — don't duplicate it (#768).
plan = plan_playlist_add(track_ids, track_id, is_link=bool(source_track_id), position=position)
if plan['should_insert']:
new_track_objs = [type('T', (), {'ratingKey': tid, 'title': ''})() for tid in plan['new_ids']]
media_server_engine.client('navidrome').create_playlist(playlist_name, new_track_objs, playlist_id=playlist_id)
_persist_find_and_add_match(source_track_id, active_server, track_id, server_track_title, source_title, source_artist)
return jsonify({"success": True, "message": "Track added"})
return jsonify({"success": True, "message": "Track linked" if not plan['should_insert'] else "Track added"})
return jsonify({"success": False, "error": f"Unsupported server: {active_server}"}), 400
except Exception as e:
@ -18573,10 +18494,17 @@ def server_playlist_remove_track(playlist_id):
logger.warning(f"[ServerPlaylist] remove-track: playlist not found by id={playlist_id} or name='{playlist_name}'")
return jsonify({"success": False, "error": "Playlist not found"}), 404
# Rebuild without the target track
# Rebuild without ONE copy of the target track — deleting one
# duplicate must not wipe every copy (#768).
current_items = list(raw_playlist.items())
new_items = [item for item in current_items if str(item.ratingKey) != str(remove_track_id)]
if len(new_items) == len(current_items):
new_items = list(current_items)
_removed = False
for _i, _it in enumerate(new_items):
if str(_it.ratingKey) == str(remove_track_id):
del new_items[_i]
_removed = True
break
if not _removed:
return jsonify({"success": False, "error": "Track not found in playlist"}), 404
raw_playlist.delete()
if new_items:
@ -18586,18 +18514,25 @@ def server_playlist_remove_track(playlist_id):
return jsonify({"success": True, "message": "Track removed (playlist now empty)"})
elif active_server == 'jellyfin' and media_server_engine.client('jellyfin'):
from core.sync.playlist_edit import remove_one_occurrence
current_tracks = media_server_engine.client('jellyfin').get_playlist_tracks(playlist_id) or []
new_ids = [str(t.ratingKey) for t in current_tracks if str(t.ratingKey) != str(remove_track_id)]
if len(new_ids) == len(current_tracks):
track_ids = [str(t.ratingKey) for t in current_tracks]
# Remove ONE occurrence, not every copy — duplicates are the same
# track, so deleting one must not wipe them all (#768).
new_ids, removed = remove_one_occurrence(track_ids, remove_track_id)
if not removed:
return jsonify({"success": False, "error": "Track not found in playlist"}), 404
new_track_objs = [type('T', (), {'ratingKey': tid, 'title': ''})() for tid in new_ids]
media_server_engine.client('jellyfin').update_playlist(playlist_name, new_track_objs)
return jsonify({"success": True, "message": "Track removed"})
elif active_server == 'navidrome' and media_server_engine.client('navidrome'):
from core.sync.playlist_edit import remove_one_occurrence
current_tracks = media_server_engine.client('navidrome').get_playlist_tracks(playlist_id) or []
new_ids = [str(t.ratingKey) for t in current_tracks if str(t.ratingKey) != str(remove_track_id)]
if len(new_ids) == len(current_tracks):
track_ids = [str(t.ratingKey) for t in current_tracks]
# Remove ONE occurrence, not every copy (#768).
new_ids, removed = remove_one_occurrence(track_ids, remove_track_id)
if not removed:
return jsonify({"success": False, "error": "Track not found in playlist"}), 404
new_track_objs = [type('T', (), {'ratingKey': tid, 'title': ''})() for tid in new_ids]
media_server_engine.client('navidrome').create_playlist(playlist_name, new_track_objs, playlist_id=playlist_id)