from collections import defaultdict from dataclasses import asdict, dataclass, field from typing import List, Dict, Any from smelog.factory import BoundLogger from charts_messages.client.utils import get_active_users_by_isrc from charts_messages.config import EVENT from charts_messages.constants import NO_SETTINGS_HANDLER from charts_messages.utils import get_user_settings_handler def handle_user_devices( devices: List[Dict[str, Any]] ) -> List[Dict[str, Any]]: """Get information that we are interested in from user devices Args: devices: List of Dict containing information of user devices ex: [{ "id": int "is_active": bool, "os": str = "ios" | "android", "expo_token": str },] Returns: List of Dicts with "expo_token and device "id" """ return [ { "expo_token": device["expo_token"], "id": device["id"] } for device in devices ] @dataclass class AccountToNotify: account_id: int devices: list = field(default_factory=list) def get_account_to_notify( account_id: str, devices: List[Dict[str, Any]] ) -> Dict[str, Any]: """Get dict with account_id and account devices information as dict Args: account_id: str representation of a user account_id devices: List[Dict] with user devices information ex: [{ "id": int "expo_token": str },] Returns: """ account_devices = handle_user_devices(devices) return asdict( AccountToNotify( account_id=int(account_id), devices=account_devices ) ) def filter_users_to_notify_by_settings( logger: BoundLogger, chart: Dict[str, str], users: List[Dict[str, Any]] ) -> Dict[str, List[Dict[str, Any]]]: """Filter users by notification settings if they are interested in notifications of a chart Can handle different versions of user settings Args: logger: chart: chart information that is processing got from event meta ex: { "dsp": "spotify", "country_code": "us", "type": "viral", "breakdown": "daily" } users: Returns: Dict with isrc keys mapped to list of user(s) device(s) that want to get a notification about this track """ result = defaultdict(list) for item in users: isrc_key = item["entity_id"] accounts = item["accounts"] for account in accounts: settings = account["settings"] settings_version = settings["version"] user_id = account["user_id"] logger.info(f"User id: {user_id}" f"Settings version: {settings_version}") handler = get_user_settings_handler(settings_version) logger.info(f"User id: {user_id}" f"User settings handler by settings version: {handler.__name__}") if handler == NO_SETTINGS_HANDLER: logger.info(f"No handler for user with id: {user_id} and settings version {settings_version}") continue if not handler(chart=chart, user_settings_data=settings["data"]): logger.info(f"User with id {user_id} does not want to get chart {EVENT} notifications" f"User settings data: {settings['data']}") continue result[isrc_key].append( get_account_to_notify( account_id=account["account_id"], devices=account["devices"] ) ) return result def get_users_to_notify( logger: BoundLogger, chart: Dict[str, str], isrc_list: List[str] ) -> Dict[str, List[Dict[str, Any]]]: """Get all active users that are interested in tracks by isrc and return only ones that wish to get such a type of notification Args: logger: chart: Dict chart information such as ["dsp", "country_code", "type", "breakdown"] isrc_list: List tracks isrc got from event Returns: Dict with isrc keys and List of Lists with user devices information that wants to get notification about isrc related track in a particular events in a chart we process events of """ users = get_active_users_by_isrc(isrc_list) return filter_users_to_notify_by_settings(logger, chart, users)