import asyncio import copy import inspect from apollo_utils.core.utils.result_types import DictResult, ListResult from apollo_utils.service.exceptions import BadRequest from datetime import date, datetime, time, timedelta, timezone from functools import cmp_to_key from operator import itemgetter from typing import Callable, Dict, Generator, Iterator, List, Optional, TypeVar from apollo_utils.service.utils.amazon import AmazonId from server.legacy.core.constants import ( APPLE, CONCURRENCY_LEVEL, DELPHI_GLOBAL_MARKET, GLOBAL_MARKET, GLOBAL_MARKET_CODE, SPOTIFY, SPOTIFY_URI_PREFIX, AMAZON, ) def get_optional_min(one, another): """Return min for two optional values.""" if another: if one: return min(one, another) return another return one def date_to_timestamp(date_obj: date or datetime, hours: int = 0) -> int: if isinstance(date_obj, date): date_obj = datetime.combine(date_obj, time(hours, 0), tzinfo=timezone.utc) return int(datetime.timestamp(date_obj)) def dates_range( start_date: date or datetime, end_date: date or datetime, as_timestamp: bool = False ) -> Generator[date or datetime or int, None, None]: """Generator yield dates between two dates.""" days_delta = end_date - start_date for i in range(days_delta.days + 1): day = start_date + timedelta(days=i) yield date_to_timestamp(day) if as_timestamp else day def iter_chunk(iterable: list, chunk_size: int = 1) -> Iterator: """Split collection of items on chunks. :param iterable: Collection of items. :param chunk_size: Size of chunks. :return: Iterator of chunks. """ count = len(iterable) for i in range(0, count, chunk_size): yield iterable[i : min(i + chunk_size, count)] # noqa def deep_sum(original_dict: dict, update_dict: dict, sum_list_items: bool = False) -> dict: """Sum up two items. Args: original_dict: Original dict to update. update_dict: Dict with new content to sum into original. Returns: Updated original dict. """ for key, value in update_dict.items(): original_value = original_dict.get(key) if isinstance(original_value, list): original_dict[key] = ( [None if x is None and y is None else (x or 0) + (y or 0) for x, y in zip(original_value, value)] if sum_list_items else (original_value + value) ) elif isinstance(original_value, dict): original_dict[key] = deep_sum(original_value, value, sum_list_items) elif isinstance(value, (list, dict)): original_dict[key] = copy.deepcopy(value) elif isinstance(value, (int, float)) or (value is None and isinstance(original_value, (int, float))): original_dict[key] = (original_value or 0) + (value or 0) else: original_dict[key] = value return original_dict ResultType = TypeVar("ResultType", ListResult, DictResult) def request_in_chunks( chunk_size: int = None, chunks_count: int = CONCURRENCY_LEVEL, items_index: int = 1, result_type: ResultType = DictResult, ) -> Callable: """Execute API requests for chunks of items. Executes original request by synchronous chunks with inner asynchronous requests equal to chunk_count each of chunk_size size. :param chunk_size: A size of a chunk. :param chunks_count: A number of chunks to run asynchronously. :param items_index: Args list index. :param result_type: Wrapped function result type class :return: Wrapped function. """ def wrapper(f: Callable): async def wrapped(*args, **kwargs): items = args[items_index] result = result_type.result_cls() sequential_chunk_size = chunks_count * (chunk_size or 1) for sequential_chunk in iter_chunk(items, sequential_chunk_size): if chunk_size: tasks = [] for concurrent_chunk in iter_chunk(sequential_chunk, chunk_size): args_list = list(args) args_list[items_index] = concurrent_chunk tasks.append(f(*tuple(args_list), **kwargs)) chunk_inner_results = await asyncio.gather(*tasks) chunk_result = result_type.result_cls() for inner_result in chunk_inner_results: chunk_result = result_type.update(chunk_result, inner_result) else: args_list = list(args) args_list[items_index] = sequential_chunk chunk_result = await f(*tuple(args_list), **kwargs) result = result_type.update(result, chunk_result) return result wrapped.__signature__ = inspect.signature(f) return wrapped return wrapper def get_id(data: dict) -> str: """Get id from apple playlist/track data.""" return data["id"] def convert_market(market: str, global_market: str = DELPHI_GLOBAL_MARKET) -> str: """Convert global market codes. Args: market: Current market code. global_market: Required global market code. Returns: Correct market code. """ market = market.lower() if market in (DELPHI_GLOBAL_MARKET, GLOBAL_MARKET, GLOBAL_MARKET_CODE): return global_market return market def format_delphi_playlist_id(vendor: str, playlist_id: str) -> str: """Set Delphi format for playlist / station ID. It is called with playlist_id as Delphi returns stations as playlists. Args: vendor: Vendor name. playlist_id: Original playlist / station ID. Returns: Delphi formatted playlist / station ID. """ if playlist_id.startswith(f"{vendor}_"): return playlist_id if vendor == SPOTIFY: if playlist_id.startswith(SPOTIFY_URI_PREFIX): playlist_id = playlist_id[len(SPOTIFY_URI_PREFIX) :] elif vendor == APPLE: if playlist_id.startswith("ra."): playlist_id = playlist_id[3:] if not playlist_id.startswith("pl.") and not playlist_id.isdigit(): playlist_id = f"pl.{playlist_id}" elif vendor == AMAZON: try: amazon_id = AmazonId.parse_from(playlist_id) playlist_id = amazon_id.group_id except ValueError: pass return f"{vendor}_{playlist_id}" def form_chunk_lists(playlists_ids: List[str], chunk_size: int): """Divide list of playlists IDs into chunks. Args: playlists_ids (List[str]): List with playlists IDs. chunk_size (int): Size of each chunk. Returns: List[list]: List with smaller lists of playlists IDs. """ return [playlists_ids[i : i + chunk_size] for i in range(0, len(playlists_ids), chunk_size)] def multikeysort(items: List[Dict], fields: Optional[List[str]]) -> List[dict]: """ Custom solution for sorting list of dicts by multiple fields with support for datetime and None objects. Args: items (List[Dict]): List of items that will be sorted. fields (List[str]): Collection of sorting fields. Returns: List[dict]: Sorted list of items. """ if not fields: return items if items: not_found_fields = [i for i in fields if i.lstrip("-") not in items[0]] if not_found_fields: raise BadRequest(f"Incorrect sort by field(s): {', '.join(not_found_fields)}") comparers = [ ((itemgetter(fld.lstrip("-")), -1) if fld.startswith("-") else (itemgetter(fld.strip()), 1)) for fld in fields ] def cmp(x, y): """ Compare the two objects x and y and return an integer according to the outcome. The return value is negative if x < y, zero if x == y and strictly positive if x > y. """ if x is None and y is None: return 0 elif x is None: return -1 elif y is None: return 1 return (x > y) - (x < y) def comparer(left, right): comparer_iter = (cmp(fn(left), fn(right)) * reverse for fn, reverse in comparers) return next((result for result in comparer_iter if result), 0) primary_field = fields[0].lstrip("-") if fields else None if primary_field: _none_items = filter(lambda itm: itm[primary_field] is None, items) _items = filter(lambda itm: itm[primary_field] is not None, items) return sorted(_items, key=cmp_to_key(comparer)) + sorted(_none_items, key=cmp_to_key(comparer)) return sorted(items, key=cmp_to_key(comparer))