"""Logic Legacy Product Mapping.""" import re from functools import lru_cache from flask import g from flask import request from oto import response from pricing import config from pricing.logic import pricing_helper from pricing.logic import product_pricing_override from pricing.logic import territories_helper from pricing.logic import track_pricing_override from pricing.models import legacy_database from pricing.models import orchard_pricing_tier from pricing.models import ows_track from pricing.models import product_migration_status from pricing.models import product_orchard_pricing_tier from pricing.models import product_pricing_override as \ product_pricing_override_model from pricing.models import store_pricing_scheme from pricing.models import track_pricing_override as \ track_pricing_override_model import requests ITUNES_STORE_ID = 1 AMAZON_STORE_ID = 187 GOOGLE_STORE_ID = 496 COUNTRY_CUTOFF_COUNT = 20 FILM_PRICING_FAMILY_ID = 1 MUSIC_ALBUM_PRICING_FAMILY_ID = 2 MUSIC_TRACK_PRICING_FAMILY_ID = 3 PHYSICAL_PRICING_FAMILY_ID = 4 PRICING_FAMILY = { 1: { 'product_type_id': 3, 'default_pricing_tier': 'Front' }, 2: { 'product_type_id': 1, 'default_pricing_tier': 'Mid/Front', 'default_orchard_pricing_tier_id': 15 }, 3: { 'product_type_id': 1, 'default_pricing_tier': 'Mid', 'default_orchard_pricing_tier_id': 28 } } EXCLUDE_STORE_IDS = [] DEFAULT_JP_ORCHARD_PRICING_TIER_ID = 27 TERRITORIES_TO_IGNORE = { ITUNES_STORE_ID: [ 'AN', 'AQ', 'AS', 'AW', 'AX', 'BV', 'CC', 'CK', 'CS', 'CX', 'EH', 'FK', 'FO', 'GF', 'GG', 'GI', 'GL', 'GP', 'GS', 'HM', 'IM', 'IO', 'MP', 'MQ', 'NC', 'NF', 'NU', 'PF', 'PM', 'PN', 'PR', 'PS', 'RE', 'SH', 'SJ', 'TF', 'TK', 'UM', 'VI', 'WF', 'YT' ], AMAZON_STORE_ID: [ 'AD', 'AE', 'AF', 'AG', 'AI', 'AL', 'AM', 'AN', 'AO', 'AQ', 'AR', 'AS', 'AU', 'AW', 'AX', 'AZ', 'BA', 'BB', 'BD', 'BE', 'BF', 'BG', 'BH', 'BI', 'BJ', 'BM', 'BN', 'BO', 'BR', 'BS', 'BT', 'BV', 'BW', 'BY', 'BZ', 'CA', 'CC', 'CD', 'CF', 'CG', 'CI', 'CK', 'CL', 'CM', 'CN', 'CO', 'CR', 'CS', 'CU', 'CV', 'CX', 'CY', 'CZ', 'DJ', 'DK', 'DM', 'DO', 'DZ', 'EC', 'EE', 'EG', 'EH', 'ER', 'ET', 'FI', 'FJ', 'FK', 'FM', 'FO', 'GA', 'GD', 'GE', 'GF', 'GG', 'GH', 'GI', 'GL', 'GM', 'GN', 'GP', 'GQ', 'GR', 'GS', 'GT', 'GU', 'GW', 'GY', 'HK', 'HM', 'HN', 'HR', 'HT', 'HU', 'ID', 'IL', 'IM', 'IN', 'IO', 'IQ', 'IR', 'IS', 'JE', 'JM', 'JO', 'KE', 'KG', 'KH', 'KI', 'KM', 'KN', 'KP', 'KR', 'KW', 'KY', 'KZ', 'LA', 'LB', 'LC', 'LI', 'LK', 'LR', 'LS', 'LT', 'LU', 'LV', 'LY', 'MA', 'MC', 'MD', 'MG', 'MH', 'MK', 'ML', 'MM', 'MN', 'MO', 'MP', 'MQ', 'MR', 'MS', 'MT', 'MU', 'MV', 'MW', 'MX', 'MY', 'MZ', 'NA', 'NC', 'NE', 'NF', 'NG', 'NI', 'NL', 'NO', 'NP', 'NR', 'NU', 'NZ', 'OM', 'PA', 'PE', 'PF', 'PG', 'PH', 'PK', 'PL', 'PM', 'PN', 'PR', 'PS', 'PT', 'PW', 'PY', 'QA', 'RE', 'RO', 'RU', 'RW', 'SA', 'SB', 'SC', 'SD', 'SE', 'SG', 'SH', 'SI', 'SJ', 'SK', 'SL', 'SM', 'SN', 'SO', 'SR', 'ST', 'SV', 'SY', 'SZ', 'TC', 'TD', 'TF', 'TG', 'TH', 'TJ', 'TK', 'TL', 'TM', 'TN', 'TO', 'TR', 'TT', 'TV', 'TW', 'TZ', 'UA', 'UG', 'UM', 'UY', 'UZ', 'VA', 'VC', 'VE', 'VG', 'VI', 'VN', 'VU', 'WF', 'WS', 'YE', 'YT', 'ZA', 'ZM', 'ZW' ], GOOGLE_STORE_ID: [] } CUTOFF_PRODUCT_ID = 2089533 def map_legacy_product_pricing_data(pricing_family_id, product_id): """Map legacy pricing tiers to the new structure. Args: pricing_family_id (int): the pricing_family_id product_id (int): the id of the product. Returns: The inserted pricing tiers. """ if isinstance(pricing_family_id, str): pricing_family_id = int(pricing_family_id) if isinstance(product_id, str): product_id = int(product_id) is_product_migrated_response = product_orchard_pricing_tier.\ get_by_product_id_and_pricing_family_id(product_id, pricing_family_id) if is_product_migrated_response.status == 200: return response.Response(message='Product Already exists', status=400) else: orchard_pricing_tiers = orchard_pricing_tier. \ get_orchard_pricing_tier_by_pricing_family_id( pricing_family_id) orchard_pricing_tiers = orchard_pricing_tiers.message if not orchard_pricing_tiers: return response.Response( message='Could not find orchard pricing tiers for family: {0}'. format(pricing_family_id), status=400) legacy_pricing = get_legacy_pricing(pricing_family_id, product_id) pricing_map = create_pricing_map(legacy_pricing['pricing'], orchard_pricing_tiers) orchard_tier = choose_orchard_tier(legacy_pricing['chosen_tiers'], orchard_pricing_tiers, pricing_map) return migrate_to_orchard_tier(orchard_tier, orchard_pricing_tiers, pricing_map, pricing_family_id, product_id) def create_pricing_map(legacy_pricing, orchard_pricing_tiers): """Organise the legacy pricing information by orchard tier. Args: legacy_pricing (dict): the territory to tier dict orchard_pricing_tiers ([dict]): the orchard pricing tiers Returns: the map of legacy tiers to orchard pricing tiers and territories. """ result = {} for territory, pricing_tier in legacy_pricing.items(): matching_orchard_tier = get_matching_orchard_pricing_tier( orchard_pricing_tiers, pricing_tier) if pricing_tier not in result: result[pricing_tier] = { 'tier': matching_orchard_tier, 'territories': set([]) } result[pricing_tier]['territories'].add(territory) return result def choose_orchard_tier(chosen_tiers, orchard_pricing_tiers, pricing_map): """Choose orchard pricing tier. Args: chosen_tiers (dict): the tiers that have been chosen in legacy db orchard_pricing_tiers ([dict]): the orchard pricing tiers pricing_map (dict): all the pricing from the legacy db Returns: the orchard pricing tier to use. """ most_chosen_tier_name = '' most_chosen_tier_count = 0 for tier_name, count in chosen_tiers.items(): if count > most_chosen_tier_count: most_chosen_tier_count = count most_chosen_tier_name = tier_name if most_chosen_tier_name and most_chosen_tier_count > COUNTRY_CUTOFF_COUNT: orchard_tier = get_matching_orchard_pricing_tier( orchard_pricing_tiers, most_chosen_tier_name) if orchard_tier: return orchard_tier largest_item = { 'territories': [] } for item in pricing_map.values(): if len(item['territories']) > len(largest_item['territories']): largest_item = item if 'tier' in largest_item: return largest_item['tier'] for tier in orchard_pricing_tiers['items']: if tier['is_default']: return tier def migrate_to_orchard_tier( orchard_tier, pricing_map, pricing_family_id, product_id): """Migrate this product to the default tier, use overrides for rest. Args: orchard_pricing_tiers ([dict]): the orchard tiers product_id (int): the id of the product. Returns: a dict of country codes to price tiers. """ data = { 'orchard_pricing_tier_id': orchard_tier[ 'orchard_pricing_tier_id'] } tier_response = product_orchard_pricing_tier. \ update_product_with_orchard_pricing_tier( product_id, pricing_family_id, data) if not tier_response: return tier_response override_counter = 1 for tier_name, item in pricing_map.items(): if item['tier'] == orchard_tier: continue data = { 'territories': list(item['territories']), 'orchard_pricing_tier_id': item['tier']['orchard_pricing_tier_id'], 'pricing_family_id': pricing_family_id, 'applies_worldwide': False, 'territory_list_include': True, 'resolution': 'All', 'sort_order': override_counter } create_response = product_pricing_override.\ create_product_pricing_override(product_id, data) if not create_response: return create_response override_counter += 1 return response.Response( status=200, message='Set default tier to: {0} and created {1} overrides'. format(orchard_tier['name'], override_counter - 1)) def get_legacy_pricing(pricing_family_id, product_id): """Get the legacy pricing for a product. Args: pricing_family_id (int): the pricing_family_id product_id (int): the id of the product. Returns: a dict of country codes to price tiers. """ result = { 'pricing': {}, 'chosen_tiers': {} } product_type_id = PRICING_FAMILY[pricing_family_id]['product_type_id'] default_pricing = legacy_database.get_default_pricing_codes( ITUNES_STORE_ID, product_type_id, 'release') for row in default_pricing: result['pricing'][row['country_code']] = row['pricing_tier'] product_pricing = legacy_database.get_all_product_territory_pricing( product_id, ITUNES_STORE_ID, product_type_id) for row in product_pricing: result['pricing'][row['country_code']] = row['pricing_tier'] if row['pricing_tier'] not in result['chosen_tiers']: result['chosen_tiers'][row['pricing_tier']] = 0 result['chosen_tiers'][row['pricing_tier']] += 1 result['priced_territory_count'] = len(product_pricing) return result def map_response_array_of_dicts(db_response, column_list): """Map the legacy db response to an array of pricing tier dictionaries. Args: db_response (List): db response is an array of tuples column_list (List): List of strings with all the column names Returns: The legacy pricing tiers mapped as array of dictionaries """ result = [] for item in db_response: result.append(dict(zip(column_list, item))) return result def check_if_legacy_pricing_tiers_are_the_same(pricing_tiers): """Check if the pricing tiers from the legacy db are the same. Args: pricing_tiers (List): The legacy pricing tiers Returns: a boolean """ first_tier = pricing_tiers[0]['pricing_tier'] for item in pricing_tiers[1:]: if item['pricing_tier'] != first_tier: return False return True def get_matching_orchard_pricing_tier( orchard_pricing_tiers, legacy_pricing_tier): """Find the orchard pricing tier that matches legacy tier. Args: orchard_pricing_tiers (List): The orchard pricing tiers legacy_pricing_tier: (string): The legacy pricing tier as string Returns: """ result = [ orchard_pricing_tier for orchard_pricing_tier in orchard_pricing_tiers['items'] if re.sub(r'[\W_]+', '', legacy_pricing_tier) == re.sub(r'[\W_]+', '', orchard_pricing_tier['name']) # noqa: E501 ] if len(result) > 0: result = result[0] else: result = None return result def unmigrate_product(product_id): """Unmigrate a product by deleting orchard pricing tier and overrides. Args: product_id: (int): The product to unmigrate Returns: response.Response: the response """ product_orchard_pricing_tier.delete_by_product_id_and_pricing_family_id( product_id, 1) product_orchard_pricing_tier.delete_by_product_id_and_pricing_family_id( product_id, 2) product_orchard_pricing_tier.delete_by_product_id_and_pricing_family_id( product_id, 3) product_orchard_pricing_tier.delete_by_product_id_and_pricing_family_id( product_id, 4) product_pricing_override_model.delete_by_product_id(product_id) track_response = ows_track.get_tracks_by_product_id(product_id) if track_response and track_response.message: track_ids = [track['tuid'] for track in track_response.message['items']] for track_id in track_ids: unmigrate_track(track_id) return response.Response() def unmigrate_track(track_id): """Unmigrate a track by deleting its overrides. Args: track_id: (int): The track to unmigrate Returns: response.Response: the response """ return track_pricing_override_model.delete_by_track_id(track_id) def bulk_migrate_film_products(): """Migrate film products to ows-pricing. Returns: a list of dicts with the migration status of each upc """ product_id_rows = legacy_database.get_all_film_product_ids() result = [] all_product_ids = set([]) for product_id in product_id_rows: all_product_ids.add(product_id['release_id']) migrated_product_ids = set( product_orchard_pricing_tier.get_all_migrated_product_ids() ) product_ids_to_migrate = list(all_product_ids - migrated_product_ids) for product_id in product_ids_to_migrate[:20]: url = config.OWS_PRICING_URL + '/legacy/1/product/{0}'.\ format(product_id) migration_response = requests.get(url) item = {'product_id': product_id} if migration_response.status_code != 200: item['result'] = 'Did not migrate' item['status_code'] = migration_response.status_code item['text'] = migration_response.text else: item['result'] = migration_response.text result.append(item) return response.Response(status=200, message=result) def count_bulk_migrate_film_products(): """Count how many products remain to be migrated. Returns: a count of the product IDs left to be migrated """ product_id_rows = legacy_database.get_all_film_product_ids() result = [] all_product_ids = set([]) for row in product_id_rows: all_product_ids.add(row['release_id']) migrated_product_ids = set( product_orchard_pricing_tier.get_all_migrated_product_ids() ) product_ids_to_migrate = list(all_product_ids - migrated_product_ids) result = { 'all_product_ids': len(all_product_ids), 'migrated_product_ids': len(migrated_product_ids), 'product_ids_to_migrate': len(product_ids_to_migrate) } return response.Response(result) def preview_music_migration(product_id): """Preview what the migration script will do for a product. Args: product_id: (int): The ID of the product Returns: a dict describing what migration will do """ product_migration_plan = get_product_migration_plan(product_id) track_migration_plans = get_track_migration_plan(product_id) result = { 'product_migration_plan': product_migration_plan, 'track_migration_plans': track_migration_plans } if config.ENVIRONMENT != config.TEST_ENVIRONMENT and \ request.args.get('profile'): result['performance_profile'] = g.performance_profile return response.Response(result) def execute_music_migration(product_id): """Execute the migration for a music product. Args: product_id: (int): The ID of the product Returns: response.Response: the overrides that have been created. """ product_migration_plan = get_product_migration_plan(product_id) track_migration_plans = get_track_migration_plan(product_id) album_orchard_pricing_tier_response = create_album_orchard_pricing_tier( product_id, product_migration_plan) if not album_orchard_pricing_tier_response: return album_orchard_pricing_tier_response album_orchard_pricing_tier = album_orchard_pricing_tier_response.message tracks_orchard_pricing_tier_response = create_tracks_orchard_pricing_tier( product_id, track_migration_plans) if not tracks_orchard_pricing_tier_response: return tracks_orchard_pricing_tier_response tracks_orchard_pricing_tier = tracks_orchard_pricing_tier_response.message album_overrides_response = create_album_overrides(product_migration_plan) if not album_overrides_response: return album_overrides_response album_overrides = album_overrides_response.message all_track_override_response = create_all_track_override( track_migration_plans) if not all_track_override_response: return all_track_override_response all_track_override = all_track_override_response.message track_overrides_response = create_track_overrides_for_migration( track_migration_plans) if not track_overrides_response: return track_overrides_response track_overrides = track_overrides_response.message result = { 'album_orchard_pricing_tier': album_orchard_pricing_tier, 'tracks_orchard_pricing_tier': tracks_orchard_pricing_tier, 'album_overrides': album_overrides, 'all_track_override': all_track_override, 'track_overrides': track_overrides } if config.ENVIRONMENT != config.TEST_ENVIRONMENT and \ request.args.get('profile'): result['performance_profile'] = g.performance_profile if 'status_before' in request.args and 'status_after' in request.args: status = request.args['status_before'] data = { 'migration_status': request.args['status_after'] } update_response = product_migration_status.\ update_by_status_and_product_id(status, product_id, data) if update_response: result['migration_status'] = update_response.message return response.Response(result) def jit_music_migration(product_id): """Just in time migration for a music product. Args: product_id: (int): The ID of the product Returns: response.Response: the result of the migration. """ film_orchard_pricing_tier_response = product_orchard_pricing_tier.\ get_by_product_id_and_pricing_family_id( product_id, FILM_PRICING_FAMILY_ID) if film_orchard_pricing_tier_response and \ film_orchard_pricing_tier_response.message: return response.Response({ 'status': 'Not migrating because the product is a Film'}) physical_orchard_pricing_tier_response = product_orchard_pricing_tier.\ get_by_product_id_and_pricing_family_id( product_id, PHYSICAL_PRICING_FAMILY_ID) if physical_orchard_pricing_tier_response and \ physical_orchard_pricing_tier_response.message: return response.Response({ 'status': 'Not migrating because the product is Physical'}) album_orchard_pricing_tier_response = product_orchard_pricing_tier.\ get_by_product_id_and_pricing_family_id( product_id, MUSIC_ALBUM_PRICING_FAMILY_ID) if not album_orchard_pricing_tier_response: return album_orchard_pricing_tier_response tracks_orchard_pricing_tier_response = product_orchard_pricing_tier.\ get_by_product_id_and_pricing_family_id( product_id, MUSIC_TRACK_PRICING_FAMILY_ID) if not tracks_orchard_pricing_tier_response: return tracks_orchard_pricing_tier_response if album_orchard_pricing_tier_response.message and \ tracks_orchard_pricing_tier_response.message: return response.Response({'status': 'Already migrated'}) return execute_music_migration(product_id) def create_album_orchard_pricing_tier(product_id, product_migration_plan): """Create the album orchard pricing tier for the migration plan. Args: product_id: (int): The ID of the product product_migration_plan: ([dict]): The migration plan for the product Returns: response.Response: the created product orchard pricing tier """ created = {} if 'orchard_pricing_tier_id' not in product_migration_plan: return response.Response(created) data = { 'orchard_pricing_tier_id': product_migration_plan[ 'orchard_pricing_tier_id'] } create_response = product_orchard_pricing_tier.\ update_product_with_orchard_pricing_tier( product_id, MUSIC_ALBUM_PRICING_FAMILY_ID, data) if not create_response: return create_response created = create_response.message return response.Response(created) def create_tracks_orchard_pricing_tier(product_id, track_migration_plans): """Create the album orchard pricing tier for the migration plan. Args: product_id: (int): The ID of the product track_migration_plans: (dict): The migration plans for the tracks Returns: response.Response: the created product orchard pricing tier """ created = {} if 'All' not in track_migration_plans: return response.Response(created) if 'orchard_pricing_tier_id' not in track_migration_plans['All']: return response.Response(created) data = { 'orchard_pricing_tier_id': track_migration_plans['All'][ 'orchard_pricing_tier_id'] } create_response = product_orchard_pricing_tier.\ update_product_with_orchard_pricing_tier( product_id, MUSIC_TRACK_PRICING_FAMILY_ID, data) if not create_response: return create_response created = create_response.message return response.Response(created) def create_album_overrides(product_migration_plan): """Create the album overrides for the migration plan. Args: product_migration_plan: ([dict]): The migration plan for the product Returns: response.Response: the created overrides """ created = [] if 'product_pricing_overrides_to_create' not in product_migration_plan: return response.Response(created) if len(product_migration_plan['product_pricing_overrides_to_create']) == 0: return response.Response(created) overrides_to_create = product_migration_plan[ 'product_pricing_overrides_to_create'] for override in overrides_to_create: if 'stores' in override: intersect = list( set(override['stores']) & set(EXCLUDE_STORE_IDS)) if len(intersect) > 0: continue create_response = product_pricing_override.\ create_product_pricing_override( override['product_id'], override) if not create_response: return create_response created.append(create_response.message) return response.Response(created) def create_all_track_override(track_migration_plans): """Create the all track override for the migration plan. Args: track_migration_plans: (dict): The migration plans for the tracks Returns: response.Response: the created override """ created = [] if 'All' not in track_migration_plans: return response.Response(created) if 'product_pricing_overrides_to_create' not in \ track_migration_plans['All']: return response.Response(created) overrides_to_create = track_migration_plans['All'][ 'product_pricing_overrides_to_create'] for override in overrides_to_create: if 'stores' in override: intersect = list( set(override['stores']) & set(EXCLUDE_STORE_IDS)) if len(intersect) > 0: continue create_response = product_pricing_override.\ create_product_pricing_override( override['product_id'], override) if not create_response: return create_response created.append(create_response.message) return response.Response(created) def create_track_overrides_for_migration(track_migration_plans): """Create the track overrides for the migration plan. Args: track_migration_plans: (dict): The migration plans for the tracks Returns: response.Response: the created overrides """ created = [] for key, item in track_migration_plans.items(): if key == 'All': continue if 'track_pricing_overrides_to_create' not in item: continue if len(item['track_pricing_overrides_to_create']) == 0: continue overrides_to_create = item['track_pricing_overrides_to_create'] for override in overrides_to_create: if 'stores' in override: intersect = list( set(override['stores']) & set(EXCLUDE_STORE_IDS)) if len(intersect) > 0: continue create_response = track_pricing_override.\ create_track_pricing_override(key, override) if not create_response: return create_response created.append(create_response.message) return response.Response(created) def get_product_migration_plan(product_id): # noqa: C901 """Preview what product pricing overrides the migration script will create. Returns: a dict describing what migration will do """ legacy_product_pricing_rows = legacy_database.\ get_product_pricing_exclude_stores(product_id, 1, EXCLUDE_STORE_IDS) if len(legacy_product_pricing_rows) == 0: account_pricing_tier_response = pricing_helper.\ get_default_account_pricing_tier( product_id, MUSIC_ALBUM_PRICING_FAMILY_ID) if account_pricing_tier_response: orchard_pricing_tier_id = account_pricing_tier_response.\ message['orchard_pricing_tier_id'] return {'orchard_pricing_tier_id': orchard_pricing_tier_id} orchard_pricing_tier_response = get_default_orchard_pricing_tier( MUSIC_ALBUM_PRICING_FAMILY_ID) if not orchard_pricing_tier_response: return orchard_pricing_tier_response orchard_pricing_tier_id = orchard_pricing_tier_response.message[ 'orchard_pricing_tier_id'] return {'orchard_pricing_tier_id': orchard_pricing_tier_id} itunes_pricing_tier = get_itunes_pricing_tier( MUSIC_ALBUM_PRICING_FAMILY_ID, legacy_product_pricing_rows) if itunes_pricing_tier is None: return 'Cannot find itunes pricing tier for this product' orchard_pricing_tier_id = itunes_pricing_tier['orchard_pricing_tier_id'] if int(product_id) >= CUTOFF_PRODUCT_ID: return { 'orchard_pricing_tier_id': orchard_pricing_tier_id, 'product_pricing_overrides_to_create': [], 'exceptions_by_store': {}, 'custom_exceptions_by_store': {}, 'itunes_pricing_tier_name': itunes_pricing_tier['name'] } custom_exceptions_by_store = {} exceptions_by_store = {} for row in legacy_product_pricing_rows: store_id = row['customer_master_master_id'] if store_id in EXCLUDE_STORE_IDS: continue if row['custom_price'] is not None: key = row['currency_code'] + '-' + str(row['custom_price']) if store_id not in custom_exceptions_by_store: custom_exceptions_by_store[store_id] = {} if key not in custom_exceptions_by_store[store_id]: custom_exceptions_by_store[store_id][key] = [] custom_exceptions_by_store[store_id][key].append( row['country_code']) else: if row_needs_override( row, orchard_pricing_tier_id, MUSIC_ALBUM_PRICING_FAMILY_ID): price_code = row['price_code'] if store_id not in exceptions_by_store: exceptions_by_store[store_id] = {} if price_code not in exceptions_by_store[store_id]: exceptions_by_store[store_id][price_code] = [] exceptions_by_store[store_id][price_code].append( row['country_code']) sort_order = 1 product_pricing_overrides_to_create = [] for store_id, price_map in custom_exceptions_by_store.items(): overrides = create_product_custom_overrides( price_map, product_id, MUSIC_ALBUM_PRICING_FAMILY_ID, store_id, sort_order) product_pricing_overrides_to_create += overrides sort_order += len(overrides) for store_id, price_map in exceptions_by_store.items(): overrides = create_product_overrides( price_map, product_id, MUSIC_ALBUM_PRICING_FAMILY_ID, store_id, sort_order) product_pricing_overrides_to_create += overrides sort_order += len(overrides) if itunes_pricing_tier['name'] != \ PRICING_FAMILY[MUSIC_ALBUM_PRICING_FAMILY_ID][ 'default_pricing_tier']: default_orchard_pricing_tier_id = PRICING_FAMILY[ MUSIC_ALBUM_PRICING_FAMILY_ID][ 'default_orchard_pricing_tier_id'] default_orchard_pricing_tier_response = \ get_default_orchard_pricing_tier( MUSIC_ALBUM_PRICING_FAMILY_ID) if default_orchard_pricing_tier_response: default_orchard_pricing_tier_id = \ default_orchard_pricing_tier_response.message[ 'orchard_pricing_tier_id'] itunes_missing_territories = get_missing_territories( legacy_product_pricing_rows, MUSIC_ALBUM_PRICING_FAMILY_ID, ITUNES_STORE_ID) amazon_missing_territories = get_missing_territories( legacy_product_pricing_rows, MUSIC_ALBUM_PRICING_FAMILY_ID, AMAZON_STORE_ID) google_missing_territories = get_missing_territories( legacy_product_pricing_rows, MUSIC_ALBUM_PRICING_FAMILY_ID, GOOGLE_STORE_ID) if len(itunes_missing_territories) > 0: itunes_missing_territories = territories_helper.\ convert_territories_to_iso(str(itunes_missing_territories)) itunes_missing_territories.sort() product_pricing_overrides_to_create.append({ 'product_id': product_id, 'pricing_family_id': 2, 'sort_order': sort_order, 'orchard_pricing_tier_id': default_orchard_pricing_tier_id, 'stores': [ITUNES_STORE_ID], 'territories': itunes_missing_territories, 'applies_worldwide': False, 'territory_list_include': True }) sort_order += 1 if len(amazon_missing_territories) > 0: amazon_missing_territories = territories_helper.\ convert_territories_to_iso(str(amazon_missing_territories)) amazon_missing_territories.sort() product_pricing_overrides_to_create.append({ 'product_id': product_id, 'pricing_family_id': 2, 'sort_order': sort_order, 'orchard_pricing_tier_id': default_orchard_pricing_tier_id, 'stores': [AMAZON_STORE_ID], 'territories': amazon_missing_territories, 'applies_worldwide': False, 'territory_list_include': True }) sort_order += 1 if len(google_missing_territories) > 0: google_missing_territories = territories_helper.\ convert_territories_to_iso(str(google_missing_territories)) google_missing_territories.sort() product_pricing_overrides_to_create.append({ 'product_id': product_id, 'pricing_family_id': 2, 'sort_order': sort_order, 'orchard_pricing_tier_id': default_orchard_pricing_tier_id, 'stores': [GOOGLE_STORE_ID], 'territories': google_missing_territories, 'applies_worldwide': False, 'territory_list_include': True }) sort_order += 1 convert_overrides_to_negative_if_needed( product_pricing_overrides_to_create) return { 'orchard_pricing_tier_id': orchard_pricing_tier_id, 'product_pricing_overrides_to_create': product_pricing_overrides_to_create, 'exceptions_by_store': exceptions_by_store, 'custom_exceptions_by_store': custom_exceptions_by_store, 'itunes_pricing_tier_name': itunes_pricing_tier['name'] } def get_track_migration_plan(product_id): # noqa: C901 """Preview what track pricing overrides the migration script will create. Returns: a dict describing what migration will do """ track_pricing_rows = legacy_database.\ get_track_pricing_exclude_stores(product_id, 1, EXCLUDE_STORE_IDS) if len(track_pricing_rows) == 0: orchard_pricing_tier_id = None account_pricing_tier_response = pricing_helper.\ get_default_account_pricing_tier( product_id, MUSIC_TRACK_PRICING_FAMILY_ID) if account_pricing_tier_response: orchard_pricing_tier_id = account_pricing_tier_response.\ message['orchard_pricing_tier_id'] if not orchard_pricing_tier_id: orchard_pricing_tier_response = get_default_orchard_pricing_tier( MUSIC_TRACK_PRICING_FAMILY_ID) if not orchard_pricing_tier_response: return orchard_pricing_tier_response orchard_pricing_tier_id = orchard_pricing_tier_response.message[ 'orchard_pricing_tier_id'] if int(product_id) >= CUTOFF_PRODUCT_ID: product_pricing_overrides_to_create = [] else: product_pricing_overrides_to_create = [ get_special_jp_product_override(product_id)] return {'All': { 'orchard_pricing_tier_id': orchard_pricing_tier_id, 'product_pricing_overrides_to_create': product_pricing_overrides_to_create }} sort_order = 1 track_migration_plans = {} group_by_track_id = {} for row in track_pricing_rows: if row['id'] not in group_by_track_id: group_by_track_id[row['id']] = [row] else: group_by_track_id[row['id']].append(row) for track_id, rows in group_by_track_id.items(): track_migration_plans[track_id] = get_single_track_migration_plan( track_id, rows, sort_order) sort_order += len( track_migration_plans[track_id][ 'track_pricing_overrides_to_create']) track_rows = legacy_database.get_product_tracks(product_id) if len(track_rows) == len(track_migration_plans.keys()): merge_multiple_identical_track_overrides( track_migration_plans, product_id) merge_orchard_pricing_tiers(track_migration_plans, product_id) if len(track_rows) + 1 != len(track_migration_plans.keys()): create_migration_plans_for_missing_tracks( track_rows, track_migration_plans, sort_order) for track_id, migration_plan in track_migration_plans.items(): if track_id == 'All': if 'product_pricing_overrides_to_create' in migration_plan: convert_overrides_to_negative_if_needed( migration_plan['product_pricing_overrides_to_create']) else: convert_overrides_to_negative_if_needed( migration_plan['track_pricing_overrides_to_create']) if int(product_id) >= CUTOFF_PRODUCT_ID: for track_id, migration_plan in track_migration_plans.items(): if track_id == 'All': if 'product_pricing_overrides_to_create' in migration_plan: migration_plan['product_pricing_overrides_to_create'] = \ purge_legacy_overrides( migration_plan[ 'product_pricing_overrides_to_create']) else: if 'track_pricing_overrides_to_create' in migration_plan: migration_plan['track_pricing_overrides_to_create'] = \ purge_legacy_overrides( migration_plan[ 'track_pricing_overrides_to_create']) return track_migration_plans def merge_orchard_pricing_tiers(track_migration_plans, product_id): """Merge orchard pricing tiers and/or create override. Args: track_migration_plans: ([dict]): The migration plans for all tracks product_id: (int): The product id """ if 'All' not in track_migration_plans: track_migration_plans['All'] = {} orchard_pricing_tiers = {} for track_id, migration_plan in track_migration_plans.items(): if track_id == 'All': continue orchard_pricing_tier = migration_plan['orchard_pricing_tier_id'] if orchard_pricing_tier not in orchard_pricing_tiers: orchard_pricing_tiers[orchard_pricing_tier] = 1 else: orchard_pricing_tiers[orchard_pricing_tier] += 1 all_track_orchard_pricing_tier = max( orchard_pricing_tiers.keys(), key=(lambda key: orchard_pricing_tiers[key])) track_migration_plans['All']['orchard_pricing_tier_id'] = \ all_track_orchard_pricing_tier has_special_jp_override = False if int(product_id) < CUTOFF_PRODUCT_ID: if 'product_pricing_overrides_to_create' in \ track_migration_plans['All']: for override in track_migration_plans['All'][ 'product_pricing_overrides_to_create']: if override['territories'] == ['JP'] and \ override['stores'] == [ITUNES_STORE_ID]: has_special_jp_override = True break for track_id, migration_plan in track_migration_plans.items(): if migration_plan['orchard_pricing_tier_id'] == \ all_track_orchard_pricing_tier: continue sort_order = len(migration_plan['track_pricing_overrides_to_create']) \ + 1 if has_special_jp_override: migration_plan['track_pricing_overrides_to_create'].append({ 'track_id': track_id, 'sort_order': sort_order, 'orchard_pricing_tier_id': migration_plan[ 'orchard_pricing_tier_id'], 'stores': [ITUNES_STORE_ID], 'applies_worldwide': False, 'territory_list_include': False, 'territories': ['JP'] }) sort_order += 1 migration_plan['track_pricing_overrides_to_create'].append({ 'track_id': track_id, 'sort_order': sort_order, 'orchard_pricing_tier_id': migration_plan[ 'orchard_pricing_tier_id'], 'stores': [AMAZON_STORE_ID, GOOGLE_STORE_ID], 'applies_worldwide': True }) else: migration_plan['track_pricing_overrides_to_create'].append({ 'track_id': track_id, 'sort_order': sort_order, 'orchard_pricing_tier_id': migration_plan[ 'orchard_pricing_tier_id'], 'applies_worldwide': True }) def merge_multiple_identical_track_overrides( track_migration_plans, product_id): """Merge track pricing overrides present on every track. Args: track_migration_plans: ([dict]): The migration plans for all tracks product_id: (int): The product id """ number_of_tracks = len(track_migration_plans.keys()) first_track_plan = list(track_migration_plans.values())[0] first_track_overrides = first_track_plan[ 'track_pricing_overrides_to_create'] if len(first_track_overrides) == 0: return product_pricing_overrides_to_create = [] merged_indexes = [] merge_index = 0 sort_order = 1 for first_track_override in first_track_overrides: merge_index += 1 number_of_mergeable_overrides = 0 for track_id in sorted(track_migration_plans.keys()): track_overrides = track_migration_plans[track_id][ 'track_pricing_overrides_to_create'] for track_override in track_overrides: if track_overrides_are_mergeable( first_track_override, track_override): track_override['merge_index'] = merge_index number_of_mergeable_overrides += 1 break if number_of_mergeable_overrides == number_of_tracks: override_to_create = { 'product_id': product_id, 'pricing_family_id': MUSIC_TRACK_PRICING_FAMILY_ID, 'sort_order': sort_order, 'stores': first_track_override['stores'], 'territories': first_track_override['territories'], 'applies_worldwide': False, 'territory_list_include': True } if 'custom_price' in first_track_override: override_to_create['custom_price'] = \ first_track_override['custom_price'] override_to_create['custom_currency_code'] = \ first_track_override['custom_currency_code'] if 'orchard_pricing_tier_id' in first_track_override: override_to_create['orchard_pricing_tier_id'] = \ first_track_override['orchard_pricing_tier_id'] elif 'price_code' in first_track_override: override_to_create['price_code'] = \ first_track_override['price_code'] product_pricing_overrides_to_create.append(override_to_create) merged_indexes.append(merge_index) sort_order += 1 sort_order = 1 for track_id in sorted(track_migration_plans.keys()): track_overrides = track_migration_plans[track_id][ 'track_pricing_overrides_to_create'] track_overrides_without_merged = [] for track_override in track_overrides: if 'merge_index' not in track_override: track_override['sort_order'] = sort_order track_overrides_without_merged.append(track_override) sort_order += 1 elif track_override['merge_index'] not in merged_indexes: del track_override['merge_index'] track_override['sort_order'] = sort_order track_overrides_without_merged.append(track_override) sort_order += 1 track_migration_plans[track_id][ 'track_pricing_overrides_to_create'] = \ track_overrides_without_merged if len(product_pricing_overrides_to_create) > 0: track_migration_plans['All'] = { 'product_pricing_overrides_to_create': product_pricing_overrides_to_create } def track_overrides_are_mergeable(track_override1, track_override2): """Work out if two track pricing overrides are mergeable. Args: track_override1: (dict): The first track pricing override track_override2: (dict): The second track pricing override Returns: bool whether the two overrides can be merged """ if 'custom_price' in track_override1 and 'custom_price' in track_override2: return track_override1['custom_price'] == \ track_override2['custom_price'] and \ track_override1['custom_currency_code'] == \ track_override2['custom_currency_code'] and \ set(track_override1['stores']) == \ set(track_override2['stores']) and \ set(track_override1['territories']) == \ set(track_override2['territories']) if 'price_code' in track_override1 and 'price_code' in track_override2: return track_override1['price_code'] == \ track_override2['price_code'] and \ set(track_override1['stores']) == \ set(track_override2['stores']) and \ set(track_override1['territories']) == \ set(track_override2['territories']) if 'orchard_pricing_tier_id' in track_override1 \ and 'orchard_pricing_tier_id' in track_override2: return track_override1['orchard_pricing_tier_id'] == \ track_override2['orchard_pricing_tier_id'] and \ set(track_override1['stores']) == \ set(track_override2['stores']) and \ set(track_override1['territories']) == \ set(track_override2['territories']) return False def get_single_track_migration_plan(track_id, track_pricing_rows, sort_order): """Work out migration plan for a single track. Args: track_id: (int): The product id track_pricing_rows: ([dict]): The pricing rows for this track sort_order: (int): The sort order Returns: [dict] the track migration plan """ itunes_pricing_tier = get_itunes_pricing_tier( MUSIC_TRACK_PRICING_FAMILY_ID, track_pricing_rows) orchard_pricing_tier_id = itunes_pricing_tier['orchard_pricing_tier_id'] custom_exceptions_by_store = {} exceptions_by_store = {} has_itunes_jp_pricing = False for row in track_pricing_rows: store_id = row['customer_master_master_id'] if store_id in EXCLUDE_STORE_IDS: continue if store_id == ITUNES_STORE_ID and row['country_code'] == 'JP': has_itunes_jp_pricing = True if row['custom_price'] is not None: key = row['currency_code'] + '-' + str(row['custom_price']) if store_id not in custom_exceptions_by_store: custom_exceptions_by_store[store_id] = {} if key not in custom_exceptions_by_store[store_id]: custom_exceptions_by_store[store_id][key] = [] custom_exceptions_by_store[store_id][key].append( row['country_code']) else: if row_needs_override( row, orchard_pricing_tier_id, MUSIC_TRACK_PRICING_FAMILY_ID): price_code = row['price_code'] if store_id not in exceptions_by_store: exceptions_by_store[store_id] = {} if price_code not in exceptions_by_store[store_id]: exceptions_by_store[store_id][price_code] = [] exceptions_by_store[store_id][price_code].append( row['country_code']) track_pricing_overrides_to_create = [] for store_id, price_map in custom_exceptions_by_store.items(): overrides = create_track_custom_overrides( price_map, track_id, MUSIC_TRACK_PRICING_FAMILY_ID, store_id, sort_order) track_pricing_overrides_to_create += overrides sort_order += len(overrides) for store_id, price_map in exceptions_by_store.items(): overrides = create_track_overrides( price_map, track_id, MUSIC_TRACK_PRICING_FAMILY_ID, store_id, sort_order) track_pricing_overrides_to_create += overrides sort_order += len(overrides) if not has_itunes_jp_pricing: track_pricing_overrides_to_create.append( get_special_jp_track_override(track_id, sort_order)) sort_order += 1 if itunes_pricing_tier['name'] != \ PRICING_FAMILY[MUSIC_TRACK_PRICING_FAMILY_ID][ 'default_pricing_tier']: default_orchard_pricing_tier_id = PRICING_FAMILY[ MUSIC_TRACK_PRICING_FAMILY_ID][ 'default_orchard_pricing_tier_id'] default_orchard_pricing_tier_response = \ get_default_orchard_pricing_tier( MUSIC_TRACK_PRICING_FAMILY_ID) if default_orchard_pricing_tier_response: default_orchard_pricing_tier_id = \ default_orchard_pricing_tier_response.message[ 'orchard_pricing_tier_id'] itunes_missing_territories = get_missing_territories( track_pricing_rows, MUSIC_TRACK_PRICING_FAMILY_ID, ITUNES_STORE_ID) amazon_missing_territories = get_missing_territories( track_pricing_rows, MUSIC_TRACK_PRICING_FAMILY_ID, AMAZON_STORE_ID) google_missing_territories = get_missing_territories( track_pricing_rows, MUSIC_TRACK_PRICING_FAMILY_ID, GOOGLE_STORE_ID) if len(itunes_missing_territories) > 0: itunes_missing_territories = territories_helper.\ convert_territories_to_iso(str(itunes_missing_territories)) itunes_missing_territories.sort() track_pricing_overrides_to_create.append({ 'track_id': track_id, 'sort_order': sort_order, 'orchard_pricing_tier_id': default_orchard_pricing_tier_id, 'stores': [ITUNES_STORE_ID], 'territories': itunes_missing_territories, 'applies_worldwide': False, 'territory_list_include': True }) sort_order += 1 if len(amazon_missing_territories) > 0: amazon_missing_territories = territories_helper.\ convert_territories_to_iso(str(amazon_missing_territories)) amazon_missing_territories.sort() track_pricing_overrides_to_create.append({ 'track_id': track_id, 'sort_order': sort_order, 'orchard_pricing_tier_id': default_orchard_pricing_tier_id, 'stores': [AMAZON_STORE_ID], 'territories': amazon_missing_territories, 'applies_worldwide': False, 'territory_list_include': True }) sort_order += 1 if len(google_missing_territories) > 0: google_missing_territories = territories_helper.\ convert_territories_to_iso(str(google_missing_territories)) google_missing_territories.sort() track_pricing_overrides_to_create.append({ 'track_id': track_id, 'sort_order': sort_order, 'orchard_pricing_tier_id': default_orchard_pricing_tier_id, 'stores': [GOOGLE_STORE_ID], 'territories': google_missing_territories, 'applies_worldwide': False, 'territory_list_include': True }) sort_order += 1 return { 'track_pricing_overrides_to_create': track_pricing_overrides_to_create, 'custom_exceptions_by_store': custom_exceptions_by_store, 'exceptions_by_store': exceptions_by_store, 'orchard_pricing_tier_id': orchard_pricing_tier_id, 'itunes_pricing_tier_name': itunes_pricing_tier['name'] } def create_product_overrides( price_map, product_id, pricing_family_id, store_id, sort_order): """Create price overrides from a price map. Args: price_map (dict): the map of price code to territory lists product_id: (int): The product id pricing_family_id: (int): The pricing family id store_id: (int): The store id sort_order: (int): The sort order Returns: [dict] the list of product overrides """ overrides = [] for price_code, country_codes in price_map.items(): territories = territories_helper.convert_territories_to_iso( str(country_codes)) territories.sort() orchard_pricing_tier_id = \ get_orchard_tier_id_for_price_code_and_territories( price_code, pricing_family_id, store_id, territories) override = { 'product_id': product_id, 'pricing_family_id': pricing_family_id, 'sort_order': sort_order, 'stores': [store_id], 'territories': territories, 'applies_worldwide': False, 'territory_list_include': True } if orchard_pricing_tier_id: override['orchard_pricing_tier_id'] = orchard_pricing_tier_id else: override['price_code'] = price_code overrides.append(override) sort_order += 1 return overrides def create_track_overrides( price_map, track_id, pricing_family_id, store_id, sort_order): """Create price overrides from a price map. Args: price_map (dict): the map of price code to territory lists track_id: (int): The track id pricing_family_id: (int): The pricing family id store_id: (int): The store id sort_order: (int): The sort order Returns: [dict] the list of track overrides """ overrides = [] for price_code, country_codes in price_map.items(): territories = territories_helper.convert_territories_to_iso( str(country_codes)) territories.sort() orchard_pricing_tier_id = \ get_orchard_tier_id_for_price_code_and_territories( price_code, pricing_family_id, store_id, territories) override = { 'track_id': track_id, 'sort_order': sort_order, 'stores': [store_id], 'territories': territories, 'applies_worldwide': False, 'territory_list_include': True } if orchard_pricing_tier_id: override['orchard_pricing_tier_id'] = orchard_pricing_tier_id else: override['price_code'] = price_code overrides.append(override) sort_order += 1 return overrides def create_product_custom_overrides( price_map, product_id, pricing_family_id, store_id, sort_order): """Create custom price overrides from a price map. Args: price_map (dict): the map of price code to territory lists product_id: (int): The product id pricing_family_id: (int): The pricing family id store_id: (int): The store id sort_order: (int): The sort order Returns: [dict] the list of product overrides """ overrides = [] for key, country_codes in price_map.items(): territories = territories_helper.convert_territories_to_iso( str(country_codes)) territories.sort() split_key = key.split('-') override = { 'product_id': product_id, 'pricing_family_id': pricing_family_id, 'sort_order': sort_order, 'stores': [store_id], 'custom_price': split_key[1], 'custom_currency_code': split_key[0], 'territories': territories, 'applies_worldwide': False, 'territory_list_include': True } overrides.append(override) sort_order += 1 return overrides def create_track_custom_overrides( price_map, track_id, pricing_family_id, store_id, sort_order): """Create custom price overrides from a price map. Args: price_map (dict): the map of price code to territory lists track_id: (int): The track id pricing_family_id: (int): The pricing family id store_id: (int): The store id sort_order: (int): The sort order Returns: [dict] the list of product overrides """ overrides = [] for key, country_codes in price_map.items(): territories = territories_helper.convert_territories_to_iso( str(country_codes)) territories.sort() split_key = key.split('-') override = { 'track_id': track_id, 'sort_order': sort_order, 'stores': [store_id], 'custom_price': split_key[1], 'custom_currency_code': split_key[0], 'territories': territories, 'applies_worldwide': False, 'territory_list_include': True } overrides.append(override) sort_order += 1 return overrides def row_needs_override(row, orchard_pricing_tier_id, pricing_family_id): """Decide whether this legacy row is an exception. Args: row (dict): the legacy pricing row orchard_price_tier_id: (int): The orchard pricing tier id pricing_family_id: (int): The pricing family id Returns: (bool) whether this row should be considered an exception """ store_id = row['customer_master_master_id'] mapped_tier = map_tier( orchard_pricing_tier_id, pricing_family_id, store_id) if not mapped_tier: return False tier_price_code = get_tier_price_code(mapped_tier, row['country_code']) return tier_price_code != row['price_code'] def map_tier(orchard_price_tier_id, pricing_family_id, store_id): """Get the tier for this store using the orchard_price_tier_id. Args: orchard_price_tier_id (int): the name of the itunes tier pricing_family_id: (int): The pricing family id store_id: (int): The store id Returns: the store_pricing_tier to use for this store """ scheme = get_store_pricing_scheme( pricing_family_id, store_id) if not scheme: return None for tier in scheme['store_pricing_tiers']: if tier['orchard_pricing_tier_id'] == orchard_price_tier_id: return tier def get_orchard_tier_id_for_price_code_and_territories( price_code, pricing_family_id, store_id, territory_codes): """Get the tier for this store and territories using the price_code. Args: price_code (int): The price_code pricing_family_id: (int): The pricing family id store_id: (int): The store id territory_codes: (dict): The territory List Returns: the orchard_pricing_tier_id to use for this store and territories """ scheme = get_store_pricing_scheme( pricing_family_id, store_id) if not scheme: return None for tier in scheme['store_pricing_tiers']: for code in tier['store_pricing_tier_codes']: territories = set() if 'territory_set' in code: territories = code['territory_set'] applies_to_territory = code['applies_worldwide'] \ or (code['territory_list_include'] and set(territory_codes).intersection(territories)) \ or (not code['territory_list_include'] and not set(territory_codes).intersection(territories)) if applies_to_territory and code['price_code'] == price_code: return tier['orchard_pricing_tier_id'] return None def get_tier_price_code(store_pricing_tier, territory_code): """Get the price_code to use for this tier. Args: store_pricing_tier (dict): the store pricing tier territory_code (string): the territory to which the code must apply Returns: the price_code to use for this tier """ for code in store_pricing_tier['store_pricing_tier_codes']: territories = set() if 'territory_set' in code: territories = code['territory_set'] applies_to_territory = code['applies_worldwide'] \ or (code['territory_list_include'] and territory_code in territories) \ or (not code['territory_list_include'] and territory_code not in territories) if applies_to_territory: return code['price_code'] @lru_cache(1000) def get_store_pricing_scheme(pricing_family_id, store_id): """Get the store pricing scheme for this store and pricing family. Args: pricing_family_id: (int): The pricing family id store_id: (int): The store id Returns: the store_pricing_scheme """ result = store_pricing_scheme.get_by_pricing_family_and_store_id( pricing_family_id, store_id) if result: return result.message def get_itunes_pricing_tier(pricing_family_id, legacy_pricing_rows): """Get the orchard pricing tier. Args: legacy_pricing_rows ([dict]): the pricing rows from artist_relations pricing_family_id: (int): The pricing family id Returns: (dict) the orchard pricing tier """ itunes_us_price_code = None for row in legacy_pricing_rows: if row['country_code'] == 'US' and \ row['customer_master_master_id'] == 1: itunes_us_price_code = row['price_code'] break itunes_store_pricing_scheme = get_store_pricing_scheme( pricing_family_id, ITUNES_STORE_ID) for itunes_pricing_tier in \ itunes_store_pricing_scheme['store_pricing_tiers']: tier_price_code = get_tier_price_code(itunes_pricing_tier, 'US') if tier_price_code == itunes_us_price_code: return itunes_pricing_tier for itunes_pricing_tier in \ itunes_store_pricing_scheme['store_pricing_tiers']: if itunes_pricing_tier['name'] == \ PRICING_FAMILY[pricing_family_id]['default_pricing_tier']: return itunes_pricing_tier def get_legacy_music_releases(): """Get legacy music releases. Returns: response.Response: containing the releases """ rows = legacy_database.get_music_releases() if not rows: return response.create_not_found_response() result = [row['release_id'] for row in rows] return response.Response({'items': result}) @lru_cache(1000) def get_default_orchard_pricing_tier(pricing_family_id): """Get the default orchard pricing tier for the pricing family. Args: pricing_family_id (int): the ID of the pricing family Returns: response.Response: containing the default orchard pricing tier """ orchard_pricing_tiers_response = orchard_pricing_tier.\ get_orchard_pricing_tier_by_pricing_family_id(pricing_family_id) if not orchard_pricing_tiers_response: return orchard_pricing_tiers_response orchard_pricing_tiers = orchard_pricing_tiers_response.message['items'] for opt in orchard_pricing_tiers: if 'is_default' in opt and opt['is_default']: return response.Response(opt) return response.Response(orchard_pricing_tiers[0]) def get_special_jp_product_override(product_id, sort_order=1): """Get the special product override for Japan. Args: product_id (int): the ID of the product Returns: dict: containing the override """ return { 'product_id': product_id, 'pricing_family_id': MUSIC_TRACK_PRICING_FAMILY_ID, 'orchard_pricing_tier_id': get_jp_orchard_pricing_tier_id(), 'sort_order': sort_order, 'stores': [ITUNES_STORE_ID], 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True } def get_special_jp_track_override(track_id, sort_order=1): """Get the special track override for Japan. Args: track_id (int): the ID of the track Returns: dict: containing the override """ return { 'track_id': track_id, 'orchard_pricing_tier_id': get_jp_orchard_pricing_tier_id(), 'sort_order': sort_order, 'stores': [ITUNES_STORE_ID], 'territories': ['JP'], 'applies_worldwide': False, 'territory_list_include': True } @lru_cache(1000) def get_jp_orchard_pricing_tier_id(): """Get the orchard pricing tier for Japan (Back). Returns: int: the orchard pricing tier id """ jp_orchard_pricing_tier_id = DEFAULT_JP_ORCHARD_PRICING_TIER_ID orchard_pricing_tiers_response = orchard_pricing_tier.\ get_orchard_pricing_tier_by_pricing_family_id( MUSIC_TRACK_PRICING_FAMILY_ID) if orchard_pricing_tiers_response: for tier in orchard_pricing_tiers_response.message['items']: if tier['name'] == 'Back': jp_orchard_pricing_tier_id = tier['orchard_pricing_tier_id'] break return jp_orchard_pricing_tier_id def get_missing_territories(pricing_rows, pricing_family_id, store_id): """Get the territories not covered by the pricing rows. Args: pricing_rows ([dict]): the pricing rows pricing_family_id (int): the pricing family ID store_id (int): the store ID Returns: list: containing the missing territories """ pricing_territories = [] for row in pricing_rows: if row['customer_master_master_id'] == store_id: pricing_territories.append(row['country_code']) world_territories = territories_helper.get_all_orchard_territories() japan = ['JP'] if pricing_family_id == MUSIC_TRACK_PRICING_FAMILY_ID and \ store_id == ITUNES_STORE_ID: return list( set(world_territories) - set(TERRITORIES_TO_IGNORE[store_id]) - set(japan) - set(pricing_territories) ) return list( set(world_territories) - set(TERRITORIES_TO_IGNORE[store_id]) - set(pricing_territories) ) def convert_overrides_to_negative_if_needed(overrides): """Convert the override territories to negative if more than 100. Args: overrides ([dict]): the overrides """ world_iso_territories = territories_helper.get_all_iso_territories() for override in overrides: if 'territory_list_include' in override and \ override['territory_list_include'] and \ 'territories' in override: applies_worldwide = len(set(world_iso_territories) - set(override['territories'])) == 0 if applies_worldwide: override['applies_worldwide'] = True del override['territories'] elif len(override['territories']) > 100: override['territories'] = territories_helper.\ convert_territories_to_positive_iso( override['territories']) override['territories'].sort() override['territory_list_include'] = False def create_migration_plans_for_missing_tracks( track_rows, track_migration_plans, sort_order): """Create migration plans for the missing tracks. Args: track_rows ([dict]): the track_rows, track_migration_plans ([dict]): the track_migration_plans, sort_order (int): the initial sort order to use for the overrides """ all_track_orchard_pricing_tier_id = track_migration_plans['All'][ 'orchard_pricing_tier_id'] default_orchard_pricing_tier_id = get_default_orchard_pricing_tier( MUSIC_TRACK_PRICING_FAMILY_ID).message['orchard_pricing_tier_id'] track_ids = [row['id'] for row in track_rows] missing_track_ids = list( set(track_ids) - set(track_migration_plans.keys())) for track_id in missing_track_ids: overrides = [get_special_jp_track_override(track_id, sort_order)] sort_order += 1 if all_track_orchard_pricing_tier_id != \ default_orchard_pricing_tier_id: overrides.append({ 'track_id': track_id, 'sort_order': sort_order, 'orchard_pricing_tier_id': default_orchard_pricing_tier_id, 'applies_worldwide': True }) sort_order += 1 track_migration_plans[track_id] = { 'track_pricing_overrides_to_create': overrides } def purge_legacy_overrides(overrides): """Purge overrides that are not worldwide for all stores. Args: overrides ([dict]): a list of overrides to purge Returns: [dict]: the purged list of overrides """ overrides_to_keep = [] for override in overrides: if 'applies_worldwide' in override and \ override['applies_worldwide'] and \ ('stores' not in override or len(override['stores']) == 0): overrides_to_keep.append(override) return overrides_to_keep