"""Logic for ASGV payload preparation.""" from typing import Any from pydantic import BaseModel, ConfigDict, Field _FALSY_LYRICS: frozenset[str | None] = frozenset([None, "None", ""]) # --- Input models (raw GQL shape) --- # extra="ignore" silently drops all fields not declared here, # which handles all field stripping without an explicit exclusion list. _BASE_CONFIG = ConfigDict(extra="ignore", populate_by_name=True) class _RawPrimaryArtist(BaseModel): model_config = _BASE_CONFIG artist_name: str | None = Field(None, alias="artistName") artist_type: str | None = Field(None, alias="artistType") class _RawWriter(BaseModel): model_config = _BASE_CONFIG track_writer_name: str | None = Field(None, alias="trackWriterName") track_writer_type: str | None = Field(None, alias="trackWriterType") class _RawPerformer(BaseModel): model_config = _BASE_CONFIG # roleId dropped via extra="ignore" name: str | None = None role: str | None = None type: str | None = None class _RawReleaseArtist(BaseModel): model_config = _BASE_CONFIG # artistId dropped via extra="ignore" artist_name: str | None = Field(None, alias="artistName") artist_type: str | None = Field(None, alias="artistType") class _RawSubgenre(BaseModel): model_config = _BASE_CONFIG # id dropped via extra="ignore" name: str | None = None genre: dict[str, Any] | None = None class _RawTrack(BaseModel): model_config = _BASE_CONFIG # Stripped via extra="ignore": isrc, isrcInfo, tuid, offerType, # ownershipRights, pInfo, previewStartTime, instantGrats, focusTrack, # audioAttributes*, rightsAttributes*, validation, labelSoundRecording, # participations track_number: int | None = Field(None, alias="trackNumber") volume_number: int | None = Field(None, alias="volumeNumber") track_name: str | None = Field(None, alias="trackName") duration_minutes: int | None = Field(None, alias="durationMinutes") duration_seconds: int | None = Field(None, alias="durationSeconds") version: str | None = None explicit: str | None = None lyrics: str | None = None language: dict[str, str] | None = None primary_artists: list[_RawPrimaryArtist] = Field( default_factory=list, alias="primaryArtists" ) writers: list[_RawWriter] = Field(default_factory=list) performers: list[_RawPerformer] = Field(default_factory=list) localizations: list[Any] = Field(default_factory=list) class _RawProduct(BaseModel): model_config = _BASE_CONFIG # Stripped via extra="ignore": product_code, delivered_version, # bulk_session, image_location, sale_start_date, # preorder_date, previewable, label, label_participations, # album_pricing_tier, track_pricing_tier, release_schedule, # release_correction, review_history, validation, special_instructions upc: str | None = None product_id: int | None = None product_name: str | None = None format: str | None = None version: str | None = None imprint: str | None = None c_line: str | None = None release_date: str | None = None metadata_language: dict[str, str] | None = None subgenre: _RawSubgenre | None = None artists: list[_RawReleaseArtist] = Field(default_factory=list) product_localizations: list[Any] = Field(default_factory=list) tracks: list[_RawTrack] = Field(default_factory=list) # --- Output models (clean schema for LLM evaluation) --- class ReleaseArtist(BaseModel): """A release-level artist.""" artist_name: str | None = None artist_type: str | None = None class Writer(BaseModel): """A track writer with normalized field names.""" name: str | None = None type: str | None = None class Performer(BaseModel): """A track performer with internal IDs stripped.""" name: str | None = None role: str | None = None class CleanedTrack(BaseModel): """A track with fields cleaned and restructured for LLM evaluation.""" track_number: int | None = None volume_number: int | None = None title: str | None = None version: str | None = None duration_minutes: int | None = None duration_seconds: int | None = None explicit: str | None = None has_lyrics: bool = False language: dict[str, str] | None = None artists_by_type: dict[str, list[str]] = Field(default_factory=dict) writers: list[Writer] = Field(default_factory=list) performers: list[Performer] = Field(default_factory=list) localizations: list[Any] = Field(default_factory=list) class CleanedProduct(BaseModel): """A product payload cleaned and restructured for LLM evaluation.""" upc: str | None = None release_id: int | None = None title: str | None = None format: str | None = None version: str | None = None imprint: str | None = None c_line: str | None = None release_date: str | None = None metadata_language: dict[str, str] | None = None genre: str | None = None subgenre: str | None = None release_artists: list[ReleaseArtist] = Field(default_factory=list) release_localizations: list[Any] = Field(default_factory=list) tracks: list[CleanedTrack] = Field(default_factory=list) # --- Public API --- def process_product(raw: dict[str, Any]) -> dict[str, Any]: """Parse and clean a raw GQL product payload for LLM evaluation. Strips fields irrelevant to Apple Style Guide evaluation and restructures ambiguous fields (e.g. primaryArtists grouped by artistType to prevent rule 2.6 false positives). """ product = _RawProduct.model_validate(raw) return _to_cleaned_product(product).model_dump(mode="json") # --- Private helpers --- def _has_lyrics(lyrics: str | None) -> bool: return lyrics not in _FALSY_LYRICS _ARTIST_TYPE_MAP = {"performer": "primary_artist"} def _group_primary_artists( artists: list[_RawPrimaryArtist], ) -> dict[str, list[str]]: grouped: dict[str, list[str]] = {} for artist in artists: artist_name = artist.artist_name if not artist_name: continue raw_type = artist.artist_type or "unknown" artist_type = _ARTIST_TYPE_MAP.get(raw_type, raw_type) grouped.setdefault(artist_type, []).append(artist_name) return grouped def _to_cleaned_track(track: _RawTrack) -> CleanedTrack: return CleanedTrack( track_number=track.track_number, volume_number=track.volume_number, title=track.track_name, version=track.version, duration_minutes=track.duration_minutes, duration_seconds=track.duration_seconds, explicit=track.explicit, has_lyrics=_has_lyrics(track.lyrics), language=track.language, artists_by_type=_group_primary_artists(track.primary_artists), writers=[ Writer(name=w.track_writer_name, type=w.track_writer_type) for w in track.writers ], performers=[Performer(name=p.name, role=p.role) for p in track.performers], localizations=track.localizations, ) def _to_cleaned_product(product: _RawProduct) -> CleanedProduct: genre: str | None = None subgenre_name: str | None = None if product.subgenre: subgenre_name = product.subgenre.name if product.subgenre.genre: genre = product.subgenre.genre.get("name") return CleanedProduct( upc=product.upc, release_id=product.product_id, title=product.product_name, format=product.format, version=product.version, imprint=product.imprint, c_line=product.c_line, release_date=product.release_date, metadata_language=product.metadata_language, genre=genre, subgenre=subgenre_name, release_artists=[ ReleaseArtist( artist_name=a.artist_name, artist_type=a.artist_type, ) for a in product.artists ], release_localizations=product.product_localizations, tracks=[_to_cleaned_track(t) for t in product.tracks], )