feat: add recommendation domain foundations
This commit is contained in:
parent
fad7884c42
commit
5c3ba8d6ec
10 changed files with 446 additions and 0 deletions
73
backend/app/domain/matching.py
Normal file
73
backend/app/domain/matching.py
Normal file
|
|
@ -0,0 +1,73 @@
|
|||
"""Pure normalization and conservative candidate matching."""
|
||||
|
||||
import re
|
||||
import unicodedata
|
||||
from dataclasses import dataclass
|
||||
from difflib import SequenceMatcher
|
||||
|
||||
from app.domain.models import Track, TrackCandidate
|
||||
|
||||
_SUFFIX_PATTERN = re.compile(r"(?:\s*(?:\([^)]*\)|\[[^]]*\]))+\s*$")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MatchVerdict:
|
||||
"""Explain whether a Spotify track safely matches a proposed candidate."""
|
||||
|
||||
is_match: bool
|
||||
title_similarity: float
|
||||
is_artist_match: bool
|
||||
|
||||
|
||||
def normalize_text(value: str) -> str:
|
||||
"""Normalize names for stable comparisons without transport knowledge."""
|
||||
decomposed = unicodedata.normalize("NFKD", value.casefold())
|
||||
without_marks = "".join(
|
||||
character for character in decomposed if not unicodedata.combining(character)
|
||||
)
|
||||
without_suffix = _SUFFIX_PATTERN.sub("", without_marks)
|
||||
words = "".join(character if character.isalnum() else " " for character in without_suffix)
|
||||
return " ".join(words.split())
|
||||
|
||||
|
||||
def title_similarity(candidate_title: str, track_title: str) -> float:
|
||||
"""Return normalized sequence similarity for two track titles."""
|
||||
return SequenceMatcher(
|
||||
None,
|
||||
normalize_text(candidate_title),
|
||||
normalize_text(track_title),
|
||||
).ratio()
|
||||
|
||||
|
||||
def artist_matches(candidate_artist: str, track_artists: tuple[str, ...]) -> bool:
|
||||
"""Require one normalized Spotify artist to equal the proposed artist."""
|
||||
normalized_candidate = normalize_text(candidate_artist)
|
||||
return bool(normalized_candidate) and any(
|
||||
normalize_text(track_artist) == normalized_candidate for track_artist in track_artists
|
||||
)
|
||||
|
||||
|
||||
def judge_candidate_match(
|
||||
candidate: TrackCandidate,
|
||||
track: Track,
|
||||
title_threshold: float,
|
||||
) -> MatchVerdict:
|
||||
"""Accept only a similar title paired with a near-exact artist."""
|
||||
similarity = title_similarity(candidate.title, track.title)
|
||||
is_artist_match = artist_matches(candidate.artist, track.artists)
|
||||
return MatchVerdict(
|
||||
is_match=similarity >= title_threshold and is_artist_match,
|
||||
title_similarity=similarity,
|
||||
is_artist_match=is_artist_match,
|
||||
)
|
||||
|
||||
|
||||
def candidate_key(candidate: TrackCandidate) -> str:
|
||||
"""Build the normalized cache key for a proposed title and artist."""
|
||||
return f"{normalize_text(candidate.title)}\x00{normalize_text(candidate.artist)}"
|
||||
|
||||
|
||||
def track_key(track: Track) -> str:
|
||||
"""Build the normalized title and primary-artist deduplication key."""
|
||||
primary_artist = track.artists[0] if track.artists else ""
|
||||
return f"{normalize_text(track.title)}\x00{normalize_text(primary_artist)}"
|
||||
|
|
@ -1,6 +1,7 @@
|
|||
"""Pure domain models shared across application boundaries."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from enum import StrEnum
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
|
|
@ -14,3 +15,69 @@ class Track:
|
|||
album_name: str
|
||||
album_art_url: str | None
|
||||
external_url: str | None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TrackCandidate:
|
||||
"""A title and artist pair proposed for Spotify resolution."""
|
||||
|
||||
title: str
|
||||
artist: str
|
||||
|
||||
|
||||
class Familiarity(StrEnum):
|
||||
"""How strongly a request should favor known or unknown music."""
|
||||
|
||||
FAMILIAR = "familiar"
|
||||
MIX = "mix"
|
||||
NEW = "new"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Intent:
|
||||
"""Structured interpretation and candidate set for a discovery query."""
|
||||
|
||||
mood: tuple[str, ...]
|
||||
activity: str | None
|
||||
era: tuple[str, ...]
|
||||
languages: tuple[str, ...]
|
||||
genres: tuple[str, ...]
|
||||
familiarity: Familiarity
|
||||
is_refinement: bool
|
||||
intent_summary: str
|
||||
candidates: tuple[TrackCandidate, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RerankSelection:
|
||||
"""One grounded track selected by the recommender."""
|
||||
|
||||
track_id: str
|
||||
justification: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TasteProfile:
|
||||
"""Bounded Spotify taste signals collected for one session."""
|
||||
|
||||
short_term_artists: tuple[str, ...]
|
||||
long_term_artists: tuple[str, ...]
|
||||
short_term_tracks: tuple[Track, ...]
|
||||
long_term_tracks: tuple[Track, ...]
|
||||
saved_tracks: tuple[Track, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CompressedTasteProfile:
|
||||
"""Prompt-ready taste text plus exact known Spotify track identifiers."""
|
||||
|
||||
text: str
|
||||
known_track_ids: frozenset[str]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CreatedPlaylist:
|
||||
"""The application-owned result of creating a Spotify playlist."""
|
||||
|
||||
id: str
|
||||
url: str
|
||||
|
|
|
|||
28
backend/app/domain/profile.py
Normal file
28
backend/app/domain/profile.py
Normal file
|
|
@ -0,0 +1,28 @@
|
|||
"""Compress Spotify taste signals for prompts and known-track filtering."""
|
||||
|
||||
from app.domain.models import CompressedTasteProfile, TasteProfile, Track
|
||||
|
||||
|
||||
def compress_taste_profile(profile: TasteProfile) -> CompressedTasteProfile:
|
||||
"""Build compact prompt text and the complete known track identifier set."""
|
||||
sections = (
|
||||
_line("Short-term top artists", profile.short_term_artists),
|
||||
_line("Long-term top artists", profile.long_term_artists),
|
||||
_line("Short-term top tracks", _track_labels(profile.short_term_tracks)),
|
||||
_line("Long-term top tracks", _track_labels(profile.long_term_tracks)),
|
||||
_line("Saved-track sample", _track_labels(profile.saved_tracks)),
|
||||
)
|
||||
known_tracks = (*profile.short_term_tracks, *profile.long_term_tracks, *profile.saved_tracks)
|
||||
return CompressedTasteProfile(
|
||||
text="\n".join(sections),
|
||||
known_track_ids=frozenset(track.id for track in known_tracks),
|
||||
)
|
||||
|
||||
|
||||
def _track_labels(tracks: tuple[Track, ...]) -> tuple[str, ...]:
|
||||
return tuple(f"{track.title} by {', '.join(track.artists)}" for track in tracks)
|
||||
|
||||
|
||||
def _line(label: str, values: tuple[str, ...]) -> str:
|
||||
rendered_values = "; ".join(values) if values else "none"
|
||||
return f"{label}: {rendered_values}"
|
||||
Loading…
Add table
Add a link
Reference in a new issue