import decimal import json from collections import defaultdict from datetime import date, datetime, timezone from typing import Dict, List, Tuple from sqlalchemy.engine import Result, Row from playlist.constants.stores import ACTIVE_PLAYLIST_STORE_NAMES from playlist.queries.constants import ( COMPLETION_RATE_FIELDS, DEFAULT_STREAMING_AGGREGATION_VALUES, ) def _clean_nested_arrays(obj: dict) -> dict: """Recursively clean empty arrays from nested objects. Handles nested structures like: {"streams_array": [{}]} -> {"streams_array": []} """ if not isinstance(obj, dict): return obj cleaned = {} for key, value in obj.items(): if isinstance(value, list): # Filter out empty objects and recursively clean nested dicts cleaned[key] = [ _clean_nested_arrays(item) if isinstance(item, dict) else item for item in value if item # Filter out empty/falsy items ] elif isinstance(value, dict): cleaned[key] = _clean_nested_arrays(value) else: cleaned[key] = value return cleaned def format_row(record: Row, array_fields: List[str] = []) -> dict: """array_fields: list of snowflake arrays to parse This function is defensive: some callers or test fixtures may already provide `list`/`dict` objects for array fields, while the DB driver may return JSON-encoded strings. Handle both safely. """ record = dict(record) for key, value in record.items(): if isinstance(value, datetime): record[key] = value.replace(tzinfo=timezone.utc).isoformat() continue if isinstance(value, date): record[key] = value.isoformat() continue if isinstance(value, decimal.Decimal): record[key] = float(value) continue # Convert store_id to int for consistent typing if key == "store_id" and value is not None: try: record[key] = int(value) except (ValueError, TypeError): pass continue if key in array_fields: # Normalize None/empty -> [] if not value: record[key] = [] continue # If already a python list/dict, use as-is if isinstance(value, (list, dict)): # Filter out empty objects and clean nested arrays if isinstance(value, list): cleaned = [item for item in value if item] # Recursively clean nested structures record[key] = [ _clean_nested_arrays(item) if isinstance(item, dict) else item for item in cleaned ] else: record[key] = _clean_nested_arrays(value) continue # Otherwise expect a JSON string and try to parse it try: parsed = json.loads(value) # Filter out empty objects and clean nested arrays if isinstance(parsed, list): cleaned = [item for item in parsed if item] record[key] = [ _clean_nested_arrays(item) if isinstance(item, dict) else item for item in cleaned ] else: record[key] = _clean_nested_arrays(parsed) except (ValueError, TypeError): record[key] = [] return record def _format_placement_row(record: Row) -> dict: # record: a database Row containing aggregated streaming data record = dict(record) placement_record = format_row( record, array_fields=[ "playlist_genres", "streams_array", "listeners_array", "followers_total_array", "followers_daily_change_array", "dimensions", "brand_ids_array", ], ) # Parse available_storefronts separately to preserve NULL for non-AM placements raw = placement_record.get("available_storefronts") placement_record["available_storefronts"] = ( json.loads(raw) if isinstance(raw, str) else raw ) return placement_record def _format_dict_of_placement_rows(d: dict) -> dict: placements = {} for key, value in d.items(): placements[key] = _format_placement_row(value) return placements def format_placements(placement_records: List[dict]) -> List[dict]: return [_format_placement_row(r) for r in placement_records] def format_placements_with_total_count(placement_records) -> Tuple[List[Dict], int]: placements = [] total_count = 0 for record in placement_records: placement_record = _format_placement_row(record) total_count = placement_record.pop("total_count", 0) if "store_id" in placement_record: placement_record["store_name"] = ACTIVE_PLAYLIST_STORE_NAMES[ int(placement_record["store_id"]) ] placements.append(placement_record) return placements, total_count def format_placements_with_gsr_split( placement_records, ) -> Tuple[List[Dict], List[Dict], int]: """Format placement records and split into real GSRs vs placeholders. When ISRCs don't have corresponding GSRs in Neo4j (content removed), we still want to display them using Chartmetric metadata as placeholders. Returns: Tuple of (placements_with_gsr, placeholder_placements, total_count) """ placements_with_gsr = [] placeholder_placements = [] total_count = 0 for record in placement_records: placement_record = _format_placement_row(record) total_count = placement_record.pop("total_count", 0) if "store_id" in placement_record: placement_record["store_name"] = ACTIVE_PLAYLIST_STORE_NAMES[ int(placement_record["store_id"]) ] # Check if this placement has a real GSR in Neo4j gsr_id = placement_record.pop("gsr_id", None) if gsr_id: # Has real GSR - remove Chartmetric fields as they're not needed placement_record.pop("chartmetric_track_name", None) placement_record.pop("chartmetric_artist_name", None) placement_record.pop("chartmetric_artwork_url", None) placements_with_gsr.append(placement_record) else: # No GSR - use Chartmetric data as placeholder # Rename Chartmetric fields to match expected schema placement_record["track_name"] = placement_record.pop( "chartmetric_track_name", None ) placement_record["artist_name"] = placement_record.pop( "chartmetric_artist_name", None ) placement_record["artwork_url"] = placement_record.pop( "chartmetric_artwork_url", None ) placeholder_placements.append(placement_record) return placements_with_gsr, placeholder_placements, total_count def _reduce_stream_record(record: Row, values: Dict[str, int] = None) -> Dict[str, int]: """values: dict of current values for the type being reduced""" record = dict(record) values = values or {**DEFAULT_STREAMING_AGGREGATION_VALUES} for key in DEFAULT_STREAMING_AGGREGATION_VALUES.keys(): # if non existent or with a none value, cast to zero values[key] = values.get(key, 0) + (record.get(key) if record.get(key) else 0) return values def aggregate_breakdown(records: Result) -> dict: """records: list of database records to aggregate""" stores_to_streams = {} playlist_types_to_streams = {} r = [dict(r) for r in records] store_type_counts = defaultdict(int) for placement_streams_record in r: placement_streams_record = dict(placement_streams_record) store_id = placement_streams_record["store_id"] streams_for_store_id = stores_to_streams.get(store_id) stores_to_streams[store_id] = _reduce_stream_record( placement_streams_record, streams_for_store_id ) playlist_type = placement_streams_record["playlist_type"] or "NONE" playlist_types_to_streams[playlist_type] = _reduce_stream_record( placement_streams_record, playlist_types_to_streams.get(playlist_type) ) store_type_counts[store_id] += 1 # since completion rate percents by store id and playlist type and are summed, divide by the total types per # store to get an average. would be cleaner to have this calculation done in the SQL total_count = 0 for store_id, store in stores_to_streams.items(): total_count += store["total_count"] for completion_rate_field in COMPLETION_RATE_FIELDS: if store_type_counts[store_id]: store[completion_rate_field] /= store_type_counts[store_id] return { "stores": _format_dict_of_placement_rows(stores_to_streams), "types": _format_dict_of_placement_rows(playlist_types_to_streams), "total_count": total_count, } def get_max_available_playlist_position_date() -> str: """Get max available date for playlist positions. Returns: str: The max available date in ISO format """ # We do not have a table for this yet so we default to today's date return datetime.today().strftime("%Y-%m-%d")