"""Network logic tier.""" from podcast.constants.feature_flag import FEATURE_PODCAST_IA_RESTRUCTURE from podcast.logic import megaphone from podcast.logic import user from podcast.logic import user_v2 from podcast.models import network as network_model from podcast.models import podcast as podcast_model from podcast.models import show_family as show_family_model from podcast.utils import feature_flag_utils def _get_networks(network_ids): networks = network_model.get_networks_by_ids(network_ids)['items'] if feature_flag_utils.get_feature_flag(FEATURE_PODCAST_IA_RESTRUCTURE): podcast_ids = user_v2.podcast_ids_owned_by_current_user() else: podcast_ids = user.podcast_ids_owned_by_current_user() network_id_to_counts = {} if len(podcast_ids): network_id_to_counts = network_model.get_networks_show_and_episode_count( network_ids, podcast_ids) for network in networks: if network['id'] in network_id_to_counts: num_shows, num_episodes = network_id_to_counts[network['id']] else: num_shows, num_episodes = (0, 0) network['num_shows'] = num_shows network['num_episodes'] = num_episodes return networks def _get_podcast_level_networks(): podcast_ids = user.specific_podcast_ids_for_current_user() podcasts = podcast_model.get_podcasts_by_ids(podcast_ids)['items'] network_ids = [podcast['network_id'] for podcast in podcasts] return list(set(network_ids)) def _get_show_family_level_networks(): show_family_ids = user_v2.show_family_ids_for_current_user() show_families = show_family_model.get_show_families(show_family_ids=show_family_ids)['items'] network_ids = [show_family['network_id'] for show_family in show_families] return list(set(network_ids)) def get_networks(): """Get a list of networks. Returns: dict: {'items': (list of networks) } """ podcast_level_networks = _get_podcast_level_networks() accessible_networks = user.network_ids_for_current_user() + podcast_level_networks networks = _get_networks(accessible_networks) return {'items': networks} def get_show_family_networks(): """Get a list of all networks along with show family level networks. Returns: dict: {'items': (list of networks) } """ show_family_level_networks = _get_show_family_level_networks() accessible_networks = user.network_ids_for_current_user() + show_family_level_networks networks = _get_networks(accessible_networks) return {'items': networks} def create_network(name, code, is_sony): """Create a network.""" user.current_user_is_org_admin_or_raise() megaphone_network = megaphone.create_network(name, code) return network_model.create_network(name, megaphone_network['id'], is_sony) def delete_network(megaphone_id): """Create a network.""" user.current_user_is_org_admin_or_raise() megaphone.delete_network(megaphone_id) return network_model.delete_network(megaphone_id) def get_networks_by_ids(network_ids): """Get networks by the ids. Args: network_ids (list of ints): The networks to fetch Returns: dict: {'items': (list of networks) } """ if feature_flag_utils.get_feature_flag(FEATURE_PODCAST_IA_RESTRUCTURE): user_v2.current_user_owns_network_ids_or_raise(network_ids, True) else: user.current_user_owns_network_ids_or_raise(network_ids, True) networks = _get_networks(network_ids) return {'items': networks}