from logging import Logger from typing import Any, Dict, List, Set, Tuple, Union from urllib.parse import urlencode import requests from apollo_notifications.constants import DSP, PLAYLIST_ENTITY_TYPE from apollo_notifications.user_data.base_client import BaseUserDataClient from apollo_notifications.user_data.config import UserDataConfig class UserDataClient(BaseUserDataClient): def __init__(self, config: UserDataConfig, logger: Logger): super().__init__(config, logger) self._starred_playlist_parser_by_vendor = { DSP.APPLE: self._parse_starred_playlist_item_by_markets, DSP.SPOTIFY: self._parse_starred_playlist_item } def _get_user_settings(self, params=None, headers=None, url=None) -> Dict[str, Any]: url = url or self._get_user_settings_url() headers = headers or {} headers.setdefault("X-App-Slug", self._app_slug) request = requests.Request("GET", url, params=params, headers=headers) return self._send_request(request) def _get_user_settings_url(self, **params) -> str: url = f"{self._base_url}/api/v1/settings/bulk/" if not params: return url return f"{url}?{urlencode([(k, v) for k, v in (params or {}).items()])}" def _parse_settings_item( self, item: Dict[str, Any], vendor: str, user_id_map: Dict[str, int], market_to_users_map: Dict[str, Set[str]] ): """Parse user settings item and add data to the passed structures. Args: item: settings item to parse. vendor: dsp to filter settings for. user_id_map: map of user_id to account_id. market_to_users_map: map of market to set of user interesting in getting notification for this market and passed vendor. """ item_settings = item.get("data", {}) user_id, account_id, markets = item.get("user_id"), item.get("account_id"), item_settings.get("markets", []) if not user_id or not markets: self._logger.warning(f"Settings parser got item without 'user_id' or/and 'markets', skipped: {item}.") return if item_settings.get("notifications") is True and item_settings.get("vendors", {}).get(vendor) is True: user_id_map[user_id] = account_id for market in markets: market_to_users_map.setdefault(market.lower(), set()).add(user_id) def parse_v1_mobile_settings( self, vendor: str, with_account_id: bool = False, **params ) -> Tuple[Dict[str, Set[str]], Dict[str, int] or List[str]]: """Read all user settings (mobile, v1), filter them by notifications activity status for passed vendor. Args: vendor: DSP name. with_account_id: To have in result user to account ID mapping or user ID list. Returns both: - map of market code to set of user ids related to users that want to receive notifications for this market and passed vendor, - list of users ids related to users that want to receive notifications for passed vendor. """ next_page_url = self._get_user_settings_url( type="mobile", version="1", offset=0, limit=self._batch_size, **params ) market_to_users_map = {} user_id_map = {} # needed for starred playlists getting while next_page_url: page = self._get_user_settings(url=next_page_url) _next = page.get('next') next_page_url = _next and f"{self._base_url}/" \ f"{self._inner_api_prefix}{_next.split(self._outer_api_prefix)[1]}" settings = page.get("items", []) for item in settings: self._parse_settings_item(item, vendor, user_id_map, market_to_users_map) return market_to_users_map, user_id_map if with_account_id else list(user_id_map.keys()) def _get_starred_playlists(self, params=None, headers=None, json=None, url=None) -> Dict[str, Any]: url = url or self._get_starred_playlists_url() headers = headers or {} headers.setdefault("X-App-Slug", self._app_slug) request = requests.Request("POST", url, params=params, headers=headers, json=json) return self._send_request(request) def _get_starred_playlists_url(self, **params) -> str: url = f"{self._base_url}/api/v1/favorites/by-params/bulk/" if not params: return url return f"{url}?{urlencode([(k, v) for k, v in (params or {}).items()])}" def _parse_starred_playlist_item( self, playlist: Dict[str, Any], playlist_to_users_map: Dict[str, Set[str]], playlist_ids: Set[str]): playlist_id, user_id = playlist.get("data", {}).get("id"), playlist.get("user_id") if playlist_id and user_id: playlist_to_users_map.setdefault(playlist_id, set()).add(user_id) playlist_ids.add(playlist_id) return self._logger.warning(f"Got starred playlist without id or user_id, skipped: {playlist}") def _parse_starred_playlist_item_by_markets( self, playlist: Dict[str, Any], market_to_playlists_to_users_map: Dict[str, Dict[str, Set[str]]], playlist_ids: Set[str] ): data = playlist.get("data", {}) playlist_id, user_id, market = data.get("id"), playlist.get("user_id"), data.get("country_code") if playlist_id and user_id and market: market_to_playlists_to_users_map.setdefault(market, {}).setdefault(playlist_id, set()).add(user_id) playlist_ids.add(playlist_id) return self._logger.warning(f"Got starred playlist without id or user_id or country_code, skipped: {playlist}") def parse_starred_playlists( self, vendor: str, user_list: List[str] ) -> Tuple[Union[Dict[str, Set[str]], Dict[str, Dict[str, Set[str]]]], Set[str]]: parser = self._starred_playlist_parser_by_vendor[vendor] data_mapping = {} playlist_ids = set() data = { "user_list": user_list, "entity_type_list": [PLAYLIST_ENTITY_TYPE], "vendor_list": [vendor] } next_page_url = self._get_starred_playlists_url(offset=0, limit=self._batch_size) while next_page_url: page = self._get_starred_playlists(url=next_page_url, json=data) _next = page.get('next') next_page_url = _next and f"{self._base_url}/" \ f"{self._inner_api_prefix}{_next.split(self._outer_api_prefix)[1]}" playlists = page.get("items", []) for playlist in playlists: parser(playlist, data_mapping, playlist_ids) return data_mapping, playlist_ids