"""Logic layer to handle interactions with the Stream model.""" from collections import defaultdict from datetime import datetime from typing import Any, cast from ddtrace import tracer from flask import g from marshmallow import fields from neo4j.time import Date from owsresponse import response, status as ows_status from segment import analytics from notifications import config from notifications.constants import subscriptions as subscriptions_constants from notifications.constants.stream import USER_EMAIL_FEED from notifications.models import ( graphql_router, label, ows_account, ows_users, product, stream, subscriptions, ) from notifications.types import FollowConfigs from notifications.utils.stream import get_profile_feed_id from notifications.validation.relationship import ( RELATIONSHIP_FROM_PROFILE, ParticipantFollow, SoundRecordingFollow, SubAccountFollow, VendorFollow, create_custom_follow_schema_for_ws, create_has_auto_follow_schema_for_ws, internal_to_external_entity, ) analytics.write_key = config.SEGMENT_WRITE_KEY @tracer.wrap() def _get_fanout_events( follow_configs: FollowConfigs, event_time: datetime, event: dict[str, Any], ) -> list[dict[str, Any]]: """Fan-out events according to current follows. Args: follow_configs (list): tuples of (Schema, id) to pull relationships for event_time (datetime): datetime of event occurance event (dict): custom data about event Response: list: dicts reprsenting fanned out events - metadata (dict): same as event param - time (datetime): same as event_time param - sources (list): entity types whose follows cause fan out - profile_id (int): identifier of profile to get notified - profile_type (str): type of profile to get notified """ profile_map = defaultdict(set) for schema_cls, identifier in follow_configs: if not identifier: continue schema = schema_cls() node_type = schema.declared_fields['entity_node_type'].load_default event_source = internal_to_external_entity(node_type) profile_types = schema.declared_fields['profile_type'].validate.choices # ty: ignore[unresolved-attribute] relationship = schema.declared_fields['relationship'].validate.comparable # ty: ignore[unresolved-attribute] relationship_from = ( cast(fields.Constant, schema.declared_fields.get('relationship_from')).load_default if schema.declared_fields.get('relationship_from') else RELATIONSHIP_FROM_PROFILE ) subscription_name = ( cast(fields.Constant, schema.declared_fields.get('subscription_name')).load_default if schema.declared_fields.get('subscription_name') else None ) profiles = subscriptions.get_subscribed_profiles( profile_types, node_type, identifier, relationship, relationship_from, subscription_name, ) if not profiles.message: continue # group all event sources for each profile for profile in profiles.message: key = (profile['profile_type'], profile['profile_id']) profile_map[key].add(event_source) # merge events by profile and sources and return complete event result = [] for (profile_type, profile_id), sources in profile_map.items(): result.append( { 'metadata': event, 'time': event_time, 'sources': sorted(list(sources)), 'profile_id': int(profile_id), 'profile_type': profile_type, 'profile_string': f'{profile_type}_{profile_id}', } ) return result def add_product_rejection_activity( product_id: int, rejection_reasons: list[str] ) -> response.Response: """Add an rejection activity to a feed. No need to send it to fanout lambda as each profile won't get more than 1 message. Params: product_id (int): Product id. rejection_reasons (list): Reason for rejection Returns: flask.Response: A 201 response if the activity has been added. """ subscription_feed_type = 'productRejection' product_result = product.get_product_details_by_id(product_id) if not product_result or not product_result.message: return product_result product_obj = product_result.message if product_obj['project'].get('subaccountId'): feed_type = 'Subaccount' feed_id = product_obj['project']['subaccountId'] else: feed_type = 'Vendor' feed_id = product_obj['project']['vendorId'] activity_feed_name = 'email_notification_rejection' activity_feed_id = f'{feed_type}_{feed_id}'.lower() g.add_log_tags(feed_name=activity_feed_name, feed_id=activity_feed_id, product_id=product_id) if tracer.enabled: span = tracer.current_root_span() if span: span.set_tag('activity.feed_name', activity_feed_name) span.set_tag('activity.feed_id', activity_feed_id) default_brand = label.get_default_brand_for_label(feed_type, feed_id) if default_brand is None or default_brand.message == 'theorchard': default_brand = 'orchard' else: default_brand = default_brand.message assigned_to_response = ows_account.get_vendor_assigned_to(product_obj['project']['vendorId']) if not assigned_to_response or not assigned_to_response.message: return assigned_to_response vendor_assigned_to = assigned_to_response.message assigned_to_id = vendor_assigned_to.get('id') orchadmin_user_response = ows_users.get_user(f'oa:{assigned_to_id}') if not orchadmin_user_response or not orchadmin_user_response.message: return orchadmin_user_response orchadmin_user = orchadmin_user_response.message assigned_to_email = orchadmin_user.get('email') label_name = label.get_name_for_label(feed_type, feed_id) if not label_name: g.log.error( f'Function get_name_for_label cannot resolve label_name for feed_type:' f' {feed_type} and feed_id: {feed_id}...' f' Exiting from add_product_rejection_activity.' ) return label_name else: label_name = label_name.message payload = { 'actor': 'Product', 'verb': 'Rejected', 'object': 'Audio Product', # send them as custom fields to avoid character limit in 'object' key. 'artist_name': product_obj['artist']['name'], 'product_name': product_obj['name'], 'upc': product_obj['upc'], 'project_id': product_obj['project']['id'], 'product_id': product_id, 'assigned_to_email': assigned_to_email, 'context_type': product_obj['contextType'], 'rejection_reasons': rejection_reasons, # this is used by daemon-notification to know the type of activity, sincec we directly # write to user's getStream. 'template_name': 'digital_rejection', 'original_feed': f'email_notification_rejection:{feed_type}_{feed_id}', 'default_brand': default_brand, 'label_name': label_name, 'label_id': feed_id, } user_feed_results = add_label_activities_to_user( feed_type, int(feed_id), subscription_feed_type, payload ) return user_feed_results def add_product_approval_activity(product_id: int) -> response.Response: """Add an approval activity to a feed. No need to send it to fanout lambda as each profile won't get more than 1 message. Params: product_id (int): Product id. Returns: flask.Response: A 201 response if the activity has been added. """ subscription_feed_type = 'productApproval' product_result = product.get_product_details_by_id(product_id) if not product_result or not product_result.message: return product_result product_obj = product_result.message if product_obj['project'].get('subaccountId'): feed_type = 'Subaccount' feed_id = product_obj['project']['subaccountId'] else: feed_type = 'Vendor' feed_id = product_obj['project']['vendorId'] activity_feed_name = 'email_notification_rejection' activity_feed_id = f'{feed_type}_{feed_id}'.lower() g.add_log_tags(feed_name=activity_feed_name, feed_id=activity_feed_id, product_id=product_id) if tracer.enabled: span = tracer.current_root_span() if span: span.set_tag('activity.feed_name', activity_feed_name) span.set_tag('activity.feed_id', activity_feed_id) default_brand = label.get_default_brand_for_label(feed_type, feed_id) if default_brand is None or default_brand.message == 'theorchard': default_brand = 'orchard' else: default_brand = default_brand.message label_name = label.get_name_for_label(feed_type, feed_id) if not label_name: g.log.error( f'Function get_name_for_label cannot resolve label_name for feed_type:' f' {feed_type} and feed_id: {feed_id}...' f' Exiting from add_product_approval_activity.' ) return label_name else: label_name = label_name.message payload = { 'actor': 'Product', 'verb': 'Approved', 'object': 'Audio Product', 'artist_name': product_obj['artist']['name'], 'product_name': product_obj['name'], 'product_id': product_id, 'project_id': product_obj['project']['id'], 'default_brand': default_brand, 'label_name': label_name, 'label_id': feed_id, 'upc': product_obj['upc'], 'sale_start_date': Date.iso_format(product_obj['saleStartDate']), # this is used by daemon-notification to know the type of activity, sincec we directly # write to user's getStream. 'template_name': 'digital_approval', 'original_feed': f'label_release_approval:{feed_type}_{feed_id}', 'approved_releases': [ { 'release_name': product_obj['name'], 'artist_name': product_obj['artist']['name'], 'product_id': product_id, 'project_id': product_obj['project']['id'], 'display_upc': product_obj['upc'], 'context_type': product_obj['contextType'], } ], } user_feed_results = add_label_activities_to_user( feed_type, int(feed_id), subscription_feed_type, payload ) return user_feed_results def add_label_activities_to_user( activity_type: str, activity_id: int, subscription_feed_type: str | None, payload: dict[str, Any], ) -> response.Response: """Add vendor or subaccount level activities to users following it. This only check if we follow that vendor or subaccount, not release or track or GParticipant. Keeping it same as legacy email system. It then writes to user's getstream feed directly. Getstream will write to sqs. Args: activity_type (str): Vendor or subaccount activity_id (int): vendor or subaccount id subscription_feed_type (str): FeedType on Subscription node. payload (dict): custom data to send to getstream. """ g.log.info( 'Adding label activities to user', resources=dict( activity_type=activity_type, activity_id=activity_id, subscription_feed_type=subscription_feed_type, **g.log_tags, ), ) subscription = subscriptions.get_subscription_by_param( feed_type=subscription_feed_type, followed_entity=activity_type ) if not subscription or not subscription.message: g.log.error(f'No subscription found for {subscription_feed_type}') return subscription subscription_name = subscription.message['name'] follow_configs: FollowConfigs = [ # create dynamic schema instead of doing if-else for each HAS_FOLLOWED_{subscription_name} (create_custom_follow_schema_for_ws(subscription_name, activity_type), activity_id), # also select profiles that has HAS_AUTO_FOLLOWED on identity. (create_has_auto_follow_schema_for_ws(activity_type, subscription_name), activity_id), ] default_brand = label.get_default_brand_for_label(activity_type, activity_id) if default_brand is None or default_brand.message == 'theorchard': default_brand = 'orchard' else: default_brand = default_brand.message label_name = label.get_name_for_label(activity_type, activity_id) if not label_name: g.log.error( f'Function get_name_for_label cannot resolve label_name for feed_type:' f' {activity_type} and feed_id: {activity_id}...' f' Exiting from add_label_activities_to_user.' ) return label_name else: label_name = label_name.message payload['default_brand'] = default_brand payload['label_name'] = label_name event_time = datetime.now() fanout_events = _get_fanout_events(follow_configs, event_time, payload) for x in range(len(fanout_events)): g.log.info(f'Fanout_event:{fanout_events[x]}') g.log.info(f'Number of all fanout_events: {len(fanout_events)}') profiles_affected = [] # add activity to user stream. for event in fanout_events: if event['profile_type'] != 'LabelProfile': continue add_activity( USER_EMAIL_FEED, get_profile_feed_id('alw', event['profile_id']), event['metadata'], retries=2, ) g.log.info( f'Sending {USER_EMAIL_FEED} activity', resources=dict(userId=f'{event["profile_type"]}:{event["profile_id"]}', **g.log_tags), ) profiles_affected.append(event['profile_string']) return response.Response(message=profiles_affected, status=ows_status.CREATED) def add_social_spike_activity( date: datetime, network: str, new_followers: int, chartmetric_id: int ) -> response.Response: """Resolve global participant id and add to GetStream feed. Args: date (datetime): date of event at midnight network (string): social network that was event origin new_followers (int): number of follower increase chartmetric_id (int): chartmetric artist identifier Returns: Response: 201 on added """ (participant_id, participant_name) = graphql_router.get_participant_by_chartmetric_id( chartmetric_id ) if participant_id: follow_configs: FollowConfigs = [(ParticipantFollow, participant_id)] event = { 'id': participant_id, 'name': participant_name, 'new_followers': new_followers, 'network': network, } fanout_events = _get_fanout_events(follow_configs, date, event) for x in range(len(fanout_events)): g.log.info(f'Fanout_event:{fanout_events[x]}') g.log.info(f'Number of all fanout_events: {len(fanout_events)}') return stream.add_social_spike_activity(fanout_events) return response.Response(status=ows_status.NOT_FOUND) def add_playlist_placement_activity( datetime: datetime, playlist: dict[str, Any], sound_recording: dict[str, Any] ) -> response.Response: """Resolve sound recoding by isrc, get profiles, add to GetStream feeds. Args: datetime (datetime): timestamp event happened playlist (dict): see sound_recording_activity schema sound_recording (dict): see sound_recording_activity schema Returns: response.Response: 200 nothing added, no subscribers 201 if some or all added 404 couldn't find sound recording by isrc """ (sound_recording_id, sound_recording_name, participant_ids) = ( graphql_router.get_sound_recording_details(sound_recording['isrc']) ) if sound_recording_id: vendor_ids = list(set(x['vendor_id'] for x in sound_recording['tracks'])) vendor_ids.sort() subaccount_ids = list( set( x['subaccount_id'] for x in sound_recording['tracks'] if x['subaccount_id'] is not None ) ) subaccount_ids.sort() follow_configs: FollowConfigs = [(SoundRecordingFollow, sound_recording_id)] follow_configs += [(VendorFollow, x) for x in vendor_ids] follow_configs += [(SubAccountFollow, x) for x in subaccount_ids] follow_configs += [(ParticipantFollow, x) for x in participant_ids] event = { 'sound_recording': { 'id': sound_recording_id, 'name': sound_recording_name, 'isrc': sound_recording['isrc'], }, 'playlist': playlist, } fanout_events = _get_fanout_events(follow_configs, datetime, event) for x in range(len(fanout_events)): g.log.info(f'Fanout_event:{fanout_events[x]}') g.log.info(f'Number of all fanout_events: {len(fanout_events)}') return stream.add_playlist_placement_activity(fanout_events) return response.Response(status=ows_status.NOT_FOUND) def add_trending_track_activity( # noqa: PLR0913 date: datetime, dsp: str, region: str, percent_diff: int, day_streams: int, track: dict[str, Any], ) -> response.Response: """Resolve track by isrc, get profiles, add to GetStream feeds. Args: date (datetime): date of event at midnight dsp (str): streaming platform spike occured in region (str): region spike occured in percent_diff (int): percentage diff in streams from previous day day_streams (int): streams during spike date track (dict): details about spiking track Returns: response.Response: 200 nothing added, no subscribers 201 if some or all added """ (sound_recording_id, sound_recording_name, participant_ids) = ( graphql_router.get_sound_recording_details(track['isrc']) ) if sound_recording_id: follow_configs: FollowConfigs = [ (SoundRecordingFollow, sound_recording_id), (VendorFollow, track['vendor_id']), (SubAccountFollow, track['subaccount_id']), ] follow_configs += [(ParticipantFollow, x) for x in participant_ids] track['name'] = sound_recording_name event = { 'track': track, 'day_streams': day_streams, 'percent_diff': percent_diff, 'dsp': dsp, 'region': region, } fanout_events = _get_fanout_events(follow_configs, date, event) for x in range(len(fanout_events)): g.log.info(f'Fanout_event:{fanout_events[x]}') g.log.info(f'Number of all fanout_events: {len(fanout_events)}') return stream.add_trending_track_activity(fanout_events) return response.Response(status=ows_status.NOT_FOUND) def add_streams_updated_activity( timestamp: str, store_id: int, available_date: str ) -> response.Response: """Add 'streams_updated' activity to GetStream feed. Args: timestamp (str): timestamp of event in ISO format store_id (int): id of store (DSP) available_date (str): date of stream data availability in ISO format """ event_ids = (subscriptions_constants.STREAMS_UPDATED_FEED_TYPE, str(store_id)) feed_id = f'{subscriptions_constants.STREAMS_UPDATED_FEED_TYPE}_all' activity = { 'payload': {'store_id': store_id}, 'actor': 'store', 'verb': subscriptions_constants.STREAMS_UPDATED_FEED_TYPE, 'foreign_id': ':'.join(event_ids).replace(',', ''), 'object': { 'store_id': store_id, 'available_date': available_date, 'activity_sources': ['store'], }, 'time': timestamp, } g.log.info('Adding streams_updated activity', resources={'activity': activity}) return add_activity(subscriptions_constants.STREAMS_UPDATED_FEED_TYPE, feed_id, activity) def add_activity( feed_name: str, feed_id: str, payload: dict[str, Any], retries: int = 5 ) -> response.Response: """Add an activity to a feed. Args: feed_name (str): the feed name to get the feed feed_id (str): the feed id to get the feed payload (dict): the data from which to create the activity retries (int): the number of retries Returns: Response: A 201 response if the activity has been added. """ stream_response = stream.add_activity(feed_name, feed_id, payload) if not stream_response and retries > 0: return add_activity(feed_name, feed_id, payload, retries=retries - 1) return stream_response def subscribe( user_feed_name: str, user_id: str, feed_name: str, feed_id: str, user_feed_id: str | None = None ) -> response.Response: """Subscribe a user to a feed. Args: user_feed_name (str): the user feed name user_id (str): the orchard user id feed_name (str): the feed name to get the feed feed_id (str): the feed id to get the feed user_feed_id (str): the user feed id to use instead of the user id Returns: Response: A 200 response if the user has been subscribed. """ user_response = ows_users.get_user(user_id) if not user_response: return user_response if config.SEGMENT_WRITE_KEY: analytics.identify(user_id) analytics.track( user_id, 'Subscribed', {'feed_name': feed_name, 'feed_id': feed_id, 'user_id': user_id} ) return stream.subscribe(user_feed_name, user_id, feed_name, feed_id, user_feed_id) def unsubscribe( user_feed_name: str, user_id: str, feed_name: str, feed_id: str, user_feed_id: str | None = None ) -> response.Response: """Unsubscribe a user from a feed. Args: user_feed_name (str): the user feed name user_id (str): the orchard user id feed_name (str): the feed name to get the feed feed_id (str): the feed id to get the feed user_feed_id (str): the user feed id to use instead of the user id Returns: Response: A 200 response if the user has been unsubscribed. """ if config.SEGMENT_WRITE_KEY: analytics.identify(user_id) analytics.track( user_id, 'Unsubscribed', {'feed_name': feed_name, 'feed_id': feed_id, 'user_id': user_id}, ) return stream.unsubscribe(user_feed_name, user_id, feed_name, feed_id, user_feed_id) def get_user_notifications( user_id: str, user_feed_name: str, user_feed_id: str | None = None ) -> response.Response: """Get a user's notifications. Args: user_id (str): the orchard user id user_feed_name (str): the user feed name user_feed_id (str): the user feed id to use instead of the user id Returns: Response: containing the user's notifications """ return stream.get_user_notifications(user_id, user_feed_name, user_feed_id) def get_user_subscriptions( user_id: str, user_feed_name: str, user_feed_id: str | None = None ) -> response.Response: """Get a user's subscriptions. Args: user_id (str): the orchard user id user_feed_name (str): the user feed name user_feed_id (str): the user feed id to use instead of the user id Returns: Response: containing the user's subscriptions """ return stream.get_user_subscriptions(user_id, user_feed_name, user_feed_id) def get_feed_subscribers(feed_name: str, feed_id: str) -> response.Response: """Get a feed's subscribers. Args: feed_name (str): the feed name feed_id (str): the feed id Returns: Response: containing the feed's subscribers """ return stream.get_feed_subscribers(feed_name, feed_id)