"""Script utilities.""" from datetime import timedelta import functools import time from typing import Iterator, List, Tuple import config from logger import logger def handle_errors(custom_handler=None): """Retry on db errors. Args: vendor_index (int): Vendor name argument position. custom_handler: Custom error handling function. """ def inner(f): def wrapped(*args, **kwargs): for i in range(1, config.RETRY_COUNT + 1): try: return f(*args, **kwargs) except Exception as ex: if i != config.RETRY_COUNT: logger.info(f'Retry on error: {ex}') if custom_handler: custom_handler(*args, **kwargs) time.sleep(i * config.RETRY_WAIT_RATE) else: logger.info(ex) raise ex return wrapped return inner def timing(vendor_index: int = 1): """Log execution time. Args: vendor_index (int): Vendor name argument position. """ def inner(f): @functools.wraps(f) def wrapper(*args, **kwargs): start_time = time.time() result = f(*args, **kwargs) elapsed_time = time.time() - start_time logger.debug(f'{args[vendor_index]}: {f.__name__} {timedelta(seconds=elapsed_time)}') return result return wrapper return inner def trim_playlist_id(playlist_id: str) -> str: """Remove pl. from playlist ID. Args: playlist_id (str): Playlist ID. Returns: str: Trimmed playlist ID. """ return playlist_id.replace('pl.', '') def split_chunks(items: List, chunk_size: int) -> Iterator[List]: """Split collection on chunks. Args: items (List): Full collection. chunk_size (int): Chunk size. Returns: Iterator[List]: Set of chunks. """ for i in range(0, len(items), chunk_size): yield items[i: i + chunk_size] def get_max(item1: str or Tuple[str, str], item2: str or Tuple[str, str]) -> str or Tuple[str, str]: """Get max of two items. """ if isinstance(item1, str): return item1 if item1.lower() > item2.lower() else item2 else: return item1 if tuple(i.lower() for i in item1) > tuple(i.lower() for i in item2) else item2 def max_playlist(playlists: List[str] or List[Tuple[str, str]]) -> str or Tuple[str, str]: """Get max playlist from a list case insensitive. Args: playlists: Playlist ID or ID + storefront list. Returns: Max playlist. """ max_item = '' if isinstance(playlists[0], str) else ('', '') for item in playlists: max_item = get_max(max_item, item) return max_item