"""Product Store Pricing logic.""" import datetime import re from copy import deepcopy from functools import lru_cache from flask import g from flask import request from oto import response from pricing import config from pricing.api import cache from pricing.constants import error, pricing_family from pricing.logic import pricing_helper from pricing.logic import store_pricing_scheme from pricing.logic import territories_helper from pricing.models import legacy_database from pricing.models import orchard_pricing_tier as orchard_pricing_tier_model from pricing.models import product_orchard_pricing_tier from pricing.models import product_pricing_override from pricing.models import track_pricing_override from sentry_sdk import capture_message TRACK_LEVEL_PRICING_FAMILIES = [ pricing_family.AUDIO_TRACK_PRICING_FAMILY_ID, pricing_family.VIDEO_PRICING_FAMILY_ID] STORES_ENABLED_FOR_NEW_PRICING = [1] STORES_WITH_OVERLAPPING_DATES = [1] default_pricing_tiers = { 1: { 'product_type_id': 3, 'default_price_code': '2', 'default_pricing_tier': 'Front', 'fallback-to-legacy': False, 'fallback-to-default': False }, 2: { 'product_type_id': 1, 'default_price_code': '3', 'default_pricing_tier': 'Mid', 'fallback-to-legacy': True, 'fallback-to-default': False }, 3: { 'product_type_id': 1, 'default_price_code': '3', 'default_pricing_tier': 'Mid', 'fallback-to-legacy': True, 'fallback-to-default': False }, 4: { 'product_type_id': 1, 'default_price_code': 'No Price Assigned', 'default_pricing_tier': 'CD Tier 1', 'fallback-to-legacy': False, 'fallback-to-default': True }, 5: { 'product_type_id': 1, 'default_price_code': '3', 'default_pricing_tier': 'Mid', 'fallback-to-legacy': False, 'fallback-to-default': True } } worldwide_territory_code = 'WW' def get_product_store_pricing(pricing_family_id, product_id, store_id): """Get the fully mapped product pricing for a particular store. Args: pricing_family_id (int): the ID of the pricing family. product_id (int): the ID of the product. store_id (int): the ID of the store. Returns: response.Response: containing the product pricing dict. """ if isinstance(pricing_family_id, str): pricing_family_id = int(pricing_family_id) if isinstance(store_id, str): store_id = int(store_id) if isinstance(product_id, str): product_id = int(product_id) if pricing_family_id in TRACK_LEVEL_PRICING_FAMILIES: return get_track_store_pricing( pricing_family_id, product_id, store_id) pricing_result_response = get_product_store_pricing_result( pricing_family_id, product_id, store_id) if not pricing_result_response: return pricing_result_response pricing_result = pricing_result_response.message['pricing_result'] use_interval_pricing = pricing_result_response.message[ 'use_interval_pricing'] if not use_interval_pricing: pricing_result = remove_intervals(pricing_result) result = prep_product_pricing_result(pricing_family_id, pricing_result) return response.Response(result) def get_product_store_pricing_result(pricing_family_id, product_id, store_id): """Get the product pricing result for a particular store. Args: pricing_family_id (int): the ID of the pricing family. product_id (int): the ID of the product. store_id (int): the ID of the store. Returns: response.Response: containing the product pricing dict. """ orchard_pricing_tier = product_orchard_pricing_tier.\ get_by_product_id_and_pricing_family_id(product_id, pricing_family_id) if orchard_pricing_tier: orchard_pricing_tier = orchard_pricing_tier.message pricing_scheme = load_pricing_scheme( pricing_family_id, store_id, orchard_pricing_tier) if pricing_scheme: pricing_scheme = pricing_scheme.message product_pricing_overrides = product_pricing_override.get_by_product_id( product_id ) if not product_pricing_overrides: return response.create_error_response( 400, error.ERROR_MESSAGE_DB_ISSUE) if not orchard_pricing_tier and \ not default_pricing_tiers[pricing_family_id]['fallback-to-legacy']: orchard_pricing_tier = get_default_orchard_pricing_tier( pricing_family_id) if not orchard_pricing_tier: pricing_result = get_legacy_pricing( pricing_family_id, product_id, store_id) if not orchard_pricing_tier and len(pricing_result) > 0: if store_id in STORES_ENABLED_FOR_NEW_PRICING: us_price_code = get_us_price_code( pricing_result, pricing_family_id) need_new_pricing = mark_items_needing_new_pricing( pricing_result, us_price_code) if need_new_pricing: if not pricing_scheme: return response.create_not_found_response() store_pricing_tier = get_store_pricing_tier_with_us_price_code( pricing_scheme, us_price_code) if not store_pricing_tier: return response.create_not_found_response() apply_new_pricing(pricing_result, store_pricing_tier) else: if not orchard_pricing_tier: # if we don't have an orchard and legacy pricing # let's use default orchard pricing from pricing family orchard_pricing_tier = get_default_orchard_pricing_tier( pricing_family_id) if not pricing_scheme: return response.create_not_found_response() pricing_result = get_new_pricing( pricing_family_id, orchard_pricing_tier, pricing_scheme) if not pricing_result: return response.create_not_found_response() product_pricing_overrides = product_pricing_overrides.message for override in product_pricing_overrides['items']: if override['pricing_family_id'] == pricing_family_id \ and override['activated']: pricing_result = apply_product_pricing_override( pricing_family_id, pricing_result, override, store_id, pricing_scheme) use_interval_pricing = False if pricing_scheme and 'use_interval_pricing' in pricing_scheme: use_interval_pricing = pricing_scheme['use_interval_pricing'] return response.Response({ 'pricing_result': pricing_result, 'use_interval_pricing': use_interval_pricing }) @lru_cache(1000) def get_default_orchard_pricing_tier(pricing_family_id): """Get the default orchard pricing tier for a pricing family. Args: pricing_family_id (int): the ID of the pricing family. Returns: dict: containing the orchard_pricing_tier """ orchard_pricing_tiers = orchard_pricing_tier_model.\ get_orchard_pricing_tier_by_pricing_family_id(pricing_family_id) if not orchard_pricing_tiers or not orchard_pricing_tiers.message: return None for orchard_pricing_tier in orchard_pricing_tiers.message['items']: if orchard_pricing_tier['is_default']: return orchard_pricing_tier def prep_product_pricing_result(pricing_family_id, pricing_result): """Prep the product pricing result to be returned. Args: pricing_family_id (int): the ID of the pricing family. pricing_result ([dict]): A list with the pricing results Returns: dict: containing the prepped pricing result """ for item in pricing_result: item['territories'] = \ territories_helper.convert_territories_to_positive_orchard(item) clean_pricing_result(item, pricing_family_id) non_empty_pricing_result = [] for item in pricing_result: if not item: capture_message('pricing result item is null') if 'territories' not in item: capture_message('item has no territories') if not item['territories']: capture_message('item has null territories') if len(item['territories']) > 0: non_empty_pricing_result.append(item) if len(non_empty_pricing_result) == 0: capture_message('empty list of pricing result items') add_ww_territory(pricing_family_id, non_empty_pricing_result) result = { 'items': non_empty_pricing_result } if config.ENVIRONMENT != config.TEST_ENVIRONMENT and \ request.args.get('profile'): result['performance_profile'] = g.performance_profile return result def get_track_store_pricing(pricing_family_id, product_id, store_id): """Get the fully mapped track pricing for a particular store. Args: pricing_family_id (int): the ID of the pricing family. product_id (int): the ID of the product. store_id (int): the ID of the store. Returns: response.Response: containing the track pricing dict. """ pricing_result_response = get_track_store_pricing_result( pricing_family_id, product_id, store_id) if not pricing_result_response: return pricing_result_response pricing_result = pricing_result_response.message['pricing_result'] track_rows = pricing_result_response.message['track_rows'] use_interval_pricing = pricing_result_response.message[ 'use_interval_pricing'] if not use_interval_pricing: pricing_result = remove_intervals(pricing_result) result = prep_track_pricing_result( track_rows, pricing_result, pricing_family_id) return response.Response(result) def get_track_store_pricing_result(pricing_family_id, product_id, store_id): # noqa: C901 """Get the track pricing result for a particular store. Args: pricing_family_id (int): the ID of the pricing family. product_id (int): the ID of the product. store_id (int): the ID of the store. Returns: response.Response: containing the track pricing dict. """ track_rows = legacy_database.get_product_tracks(product_id) if len(track_rows) == 0: return response.create_not_found_response() orchard_pricing_tier = product_orchard_pricing_tier.\ get_by_product_id_and_pricing_family_id(product_id, pricing_family_id) using_default_tier = False if orchard_pricing_tier and orchard_pricing_tier.message: orchard_pricing_tier = orchard_pricing_tier.message else: using_default_tier = True orchard_pricing_tier = get_default_orchard_pricing_tier( pricing_family_id) pricing_scheme = load_pricing_scheme(pricing_family_id, store_id) if pricing_scheme: pricing_scheme = pricing_scheme.message if using_default_tier: # if we don't have an orchard price tier, let's see if we have # legacy pricing pricing_result = get_legacy_track_pricing( pricing_family_id, product_id, store_id, track_rows) if using_default_tier and len(pricing_result) > 0: # if we don't have an orchard price tier and we have legacy pricing # let's use that if store_id in STORES_ENABLED_FOR_NEW_PRICING: for track_row in track_rows: us_price_code = get_us_price_code( pricing_result, pricing_family_id, track_row['id']) need_new_pricing = mark_items_needing_new_pricing( pricing_result, us_price_code, track_row['id']) if need_new_pricing: if not pricing_scheme: return response.create_not_found_response() store_pricing_tier = \ get_store_pricing_tier_with_us_price_code( pricing_scheme, us_price_code) if not store_pricing_tier: return response.create_not_found_response() apply_new_pricing( pricing_result, store_pricing_tier, track_row['id']) else: if not pricing_scheme: return response.create_not_found_response() new_pricing_result = get_new_pricing( pricing_family_id, orchard_pricing_tier, pricing_scheme) if not new_pricing_result: return response.create_not_found_response() pricing_result = [] for track_row in track_rows: for item in new_pricing_result: pricing_result.append({ 'track_ids': [track_row['id']], 'territories': item['territories'], 'territory_list_include': item['territory_list_include'], 'applies_worldwide': item['applies_worldwide'], 'price_code': item['price_code'] }) product_pricing_overrides = product_pricing_override.get_by_product_id( product_id ) if not product_pricing_overrides: return response.create_error_response( 400, error.ERROR_MESSAGE_DB_ISSUE) product_pricing_overrides = product_pricing_overrides.message for override in product_pricing_overrides['items']: if override['pricing_family_id'] == \ pricing_family.AUDIO_TRACK_PRICING_FAMILY_ID \ and override['activated']: pricing_result = apply_product_pricing_override( pricing_family_id, pricing_result, override, store_id, pricing_scheme) for track_row in track_rows: track_pricing_overrides = track_pricing_override.get_by_track_id( track_row['id']) if not track_pricing_overrides: return response.create_error_response( 400, error.ERROR_MESSAGE_DB_ISSUE) track_pricing_overrides = track_pricing_overrides.message for override in track_pricing_overrides['items']: if override['activated']: pricing_result = apply_track_pricing_override( pricing_result, override, store_id, pricing_scheme) use_interval_pricing = False if pricing_scheme and 'use_interval_pricing' in pricing_scheme: use_interval_pricing = pricing_scheme['use_interval_pricing'] return response.Response({ 'pricing_result': pricing_result, 'track_rows': track_rows, 'use_interval_pricing': use_interval_pricing }) def prep_track_pricing_result(track_rows, pricing_result, pricing_family_id): """Prep the track pricing result to be returned. Args: track_rows ([dict]): A list of tracks. pricing_result ([dict]): A list with the pricing results pricing_family_id (int): The pricing family ID. Returns: dict: containing the prepped pricing result """ for item in pricing_result: item['territories'] = \ territories_helper.convert_territories_to_positive_orchard(item) clean_pricing_result(item, pricing_family_id) pricing_result = group_track_pricing_result_by_territory(pricing_result) for item in pricing_result: item['isrcs'] = [] for track_row in track_rows: if track_row['id'] in item['track_ids']: item['isrcs'].append(track_row['isrc']) result = { 'items': pricing_result } if config.ENVIRONMENT != config.TEST_ENVIRONMENT and \ request.args.get('profile'): result['performance_profile'] = g.performance_profile return result def get_legacy_pricing(pricing_family_id, product_id, store_id): """Load the legacy pricing from artist_relations database. Args: pricing_family_id (int): the ID of the pricing family. product_id (int): the ID of the product. store_id (int): the ID of the store. Returns: [dict]: containing the product pricing from art_relations. """ product_type_id = \ default_pricing_tiers[pricing_family_id]['product_type_id'] legacy_default_tier_rows = load_legacy_default_price_codes( store_id, product_type_id, 'release') if not legacy_default_tier_rows: return [] legacy_rows = legacy_database.get_all_product_territory_pricing( product_id, store_id, product_type_id) pricing_result = map_legacy_pricing_rows( legacy_rows, legacy_default_tier_rows, pricing_family_id) for element in pricing_result: if 'legacy_pricing_tier' in element: del element['legacy_pricing_tier'] del element['resolution'] return pricing_result def get_legacy_track_pricing(pricing_family_id, product_id, store_id, track_rows): """Load the legacy track pricing from artist_relations database. Args: pricing_family_id (int): the ID of the pricing family. product_id (int): the ID of the product. store_id (int): the ID of the store. Returns: [dict]: containing the track pricing from art_relations. """ product_type_id = \ default_pricing_tiers[pricing_family_id]['product_type_id'] default_tier_rows = load_legacy_default_price_codes( store_id, product_type_id, 'track') if not default_tier_rows: return [] track_pricing_rows = legacy_database.get_product_track_territory_pricing( product_id, store_id, product_type_id) pricing_result = map_legacy_track_pricing_rows( track_rows, track_pricing_rows, default_tier_rows, pricing_family_id) return pricing_result def find_matching_new_tier_codes(pricing_element, pricing_tiers): """Find the store pricing tier codes that match on tier name and territory. Args: pricing_element (dict): the pricing element to match to. pricing_tiers ([dict]): a list of the tiers to search. Returns: [dict]: the store_pricing_tier_codes that match """ result = [] territory = pricing_element['territories'][0] for tier in pricing_tiers: tier_name = re.sub(r'[\W_]+', '', tier['name']) legacy_tier_name = re.sub( r'[\W_]+', '', pricing_element['legacy_pricing_tier']) if tier_name == legacy_tier_name: for code in tier['store_pricing_tier_codes']: if code_applies_to_territory(code, territory): code['pricing_tier'] = tier['name'] result.append(code) return result def code_applies_to_territory(store_pricing_tier_code, territory): """Check if this code applies to the territory. Args: store_pricing_tier_code (dict): the tier code to check. territory (string): the territory to check for. Returns: bool: returns true if territory is included in the tier_code """ if store_pricing_tier_code['applies_worldwide']: return True if store_pricing_tier_code['territory_list_include']: return territory in store_pricing_tier_code['territory_set'] return territory not in store_pricing_tier_code['territory_set'] def map_legacy_pricing_rows(product_rows, default_rows, pricing_family_id): """Map legacy pricing database rows to [dict]. Args: product_rows ([tuple]): the product pricing rows as a list of tuples. default_rows ([tuple]): the default pricing rows as a list of tuples. pricing_family_id (int): the pricing family ID. Returns: [dict]: the product pricing as a list of dict. """ territory_prices = {} lookup = default_pricing_tiers[pricing_family_id] for territory in territories_helper.get_all_orchard_territories(): territory_prices[territory] = { 'price_code': lookup['default_price_code'], 'legacy_pricing_tier': lookup['default_pricing_tier'], 'source': 'LegacyWorldDefault' } for row in default_rows: territory_prices[row['country_code']] = { 'price_code': row['price_code'], 'legacy_pricing_tier': row['pricing_tier'], 'source': 'LegacyStoreDefault' } for row in product_rows: territory_prices[row['country_code']] = { 'price_code': row['price_code'], 'legacy_pricing_tier': row['pricing_tier'], 'source': 'LegacyProductPrice' } pricing_result = [] for key, value in territory_prices.items(): pricing_result.append({ 'territories': [key], 'territory_list_include': True, 'applies_worldwide': False, 'resolution': 'All', 'price_code': value['price_code'], 'legacy_pricing_tier': value['legacy_pricing_tier'], 'source': value['source'] }) for item in pricing_result: territories_as_string = str(item['territories']) item['territories'] = territories_helper.convert_territories_to_iso( territories_as_string) return pricing_result def map_legacy_track_pricing_rows( track_rows, track_pricing_rows, default_rows, pricing_family_id): """Map legacy track pricing database rows to [dict]. Args: track_rows ([tuple]): the track rows as a list of tuples. track_pricing_rows ([tuple]): the track pricing rows. default_rows ([tuple]): the default pricing rows as a list of tuples. pricing_family_id (int): the pricing family ID. Returns: [dict]: the track pricing as a list of dict. """ track_prices = get_track_prices(track_pricing_rows) default_prices = get_default_track_prices(default_rows, pricing_family_id) pricing_result = [] tracks_to_default_entirely = [] for track_row in track_rows: if track_row['id'] not in track_prices: tracks_to_default_entirely.append(track_row) else: pricing_result += create_track_pricing_elements( track_row['id'], track_prices, default_prices) if tracks_to_default_entirely: first_track = tracks_to_default_entirely[0] track_ids = [track_row['id'] for track_row in tracks_to_default_entirely] pricing_elements = create_track_pricing_elements( first_track['id'], track_prices, default_prices) for element in pricing_elements: element['track_ids'] = track_ids pricing_result.append(element) converted_result = [] for item in pricing_result: territories_as_string = str(item['territories']) item['territories'] = territories_helper.convert_territories_to_iso( territories_as_string) if (len(item['territories']) > 1): for territory in item['territories']: item_with_one_territory = dict(item) item_with_one_territory['territories'] = [territory] converted_result.append(item_with_one_territory) else: converted_result.append(item) return converted_result def create_track_pricing_elements(track_id, track_prices, default_prices): """Create the pricing elements for a track. Args: track_id (int): the ID of the track. track_prices ([dict]): the prices for all the tracks. default_prices ([dict]): the default prices for tracks for the store. Returns: [dict]: the track pricing as a list of dict. """ price_map = {} track_pricing = track_prices.get(track_id) for territory_code, price_code in default_prices.items(): if track_pricing and territory_code in track_pricing: price_code = track_pricing[territory_code] price_map[territory_code] = { 'price_code': price_code, 'track_ids': [track_id], 'territories': [territory_code], 'territory_list_include': True, 'applies_worldwide': False } return price_map.values() def get_default_track_prices(default_rows, pricing_family_id): """Organise the default track prices as dict from territory to price. Args: default_rows ([tuple]): the database rows for the default pricing. pricing_family_id (int): the pricing family ID. Returns: dict: the default track pricing as a dict. """ default_prices = {} lookup = default_pricing_tiers[pricing_family_id] for territory in territories_helper.get_all_orchard_territories(): default_prices[territory] = lookup['default_price_code'] for row in default_rows: default_prices[row['country_code']] = row['price_code'] return default_prices def get_track_prices(track_pricing_rows): """Organise the track prices as dict from track to territory to price. Args: track_pricing_rows ([tuple]): the database rows for the track pricing. Returns: dict: the track pricing as a dict. """ track_prices = {} for row in track_pricing_rows: if row['id'] not in track_prices: track_prices[row['id']] = {} track = track_prices[row['id']] if row['country_code'] not in track: track[row['country_code']] = row['price_code'] return track_prices def get_new_pricing(pricing_family_id, orchard_pricing_tier, pricing_scheme): """Get pricing data for the product from the pricing database. Args: pricing_family_id (int): the ID of the pricing family. orchard_pricing_tier (dict): the orchard tier. pricing_scheme (dict): the store pricing scheme. Returns: [dict]: the product pricing as a list of dict. """ store_pricing_tier = next( ( store_tier for store_tier in pricing_scheme['store_pricing_tiers'] if store_tier['orchard_pricing_tier_id'] == orchard_pricing_tier['orchard_pricing_tier_id'] # noqa: E501 ), None ) if not store_pricing_tier: return None result = [] found_territories = set([]) for code in store_pricing_tier['store_pricing_tier_codes']: if code['applies_worldwide']: territories = territories_helper.get_all_iso_territories() found_territories = set(territories) else: territories = [territory['territory_code'] for territory in code['store_pricing_tier_code_territories']] if len(territories) > 0: if not code['territory_list_include']: positive_territories = territories_helper.\ convert_territories_to_positive_iso(territories) else: positive_territories = territories found_territories = found_territories | set( positive_territories) result.append({ 'territories': territories, 'territory_list_include': code['territory_list_include'], 'applies_worldwide': code['applies_worldwide'], 'resolution': code['resolution'] if 'resolution' in code else None, 'price_code': code['price_code'], 'store_pricing_tier_name': store_pricing_tier['name'] }) territories_to_default = set( territories_helper.get_all_iso_territories()) - found_territories if len(territories_to_default) > 0: pricing_element = { 'territories': list(territories_to_default), 'territory_list_include': True, 'applies_worldwide': False, 'price_code': default_pricing_tiers[pricing_family_id][ 'default_price_code'] } if pricing_family_id == pricing_family.FILM_PRICING_FAMILY_ID: pricing_element['resolution'] = 'SD' result.append(pricing_element) return result def clean_pricing_result(item, pricing_family_id): """Remove unnecessary keys from dict. Args: item (dict): a pricing item. pricing_family_id (int): the pricing family ID. """ if 'territory_list_include' in item: del item['territory_list_include'] if 'applies_worldwide' in item: del item['applies_worldwide'] if 'start_date' in item: item['start_date'] = str(item['start_date']) if 'end_date' in item: item['end_date'] = str(item['end_date']) if 'territories' in item and item['territories'] is not None: item['territories'].sort() if pricing_family_id != pricing_family.FILM_PRICING_FAMILY_ID \ and 'resolution' in item: del item['resolution'] if 'custom_price' in item and item['custom_price'] is None: del item['custom_price'] del item['custom_currency_code'] def add_ww_territory(pricing_family_id, pricing_result): """Add WW territory to pricing list for non-physical. Args: pricing_family_id (string): the pricing family ID pricing_result ([dict]): A list with the pricing results Returns: The pricing result with the ww territory """ if pricing_family_id == pricing_family.PHYSICAL_PRICING_FAMILY_ID: return pricing_element = { 'price_code': default_pricing_tiers[pricing_family_id]['default_price_code'], 'territories': [worldwide_territory_code] } pricing_result.append(pricing_element) def apply_product_pricing_override( pricing_family_id, pricing_result, override, store_id, store_pricing_scheme): """Apply a product pricing override to a store pricing result. Args: pricing_family_id (int): the pricing family id. pricing_result (dict): the pricing for a product on a store. override (dict): the product pricing override. store_id (int): the id of the store. store_pricing_scheme (dict): the store pricing scheme to use. """ if 'stores' in override \ and override['stores'] is not None \ and len(override['stores']) > 0 \ and store_id not in override['stores']: return pricing_result pricing_elements = map_override_to_pricing_elements( pricing_family_id, override, store_pricing_scheme) for pricing_element in pricing_elements: use_overlapping_dates = store_id in STORES_WITH_OVERLAPPING_DATES pricing_result = pricing_helper.apply_pricing_element_to_list( pricing_element, pricing_result, use_overlapping_dates) return pricing_result def apply_track_pricing_override( pricing_result, override, store_id, store_pricing_scheme): """Apply a track pricing override to a store pricing result. Args: pricing_result (dict): the pricing for a product on a store. override (dict): the product pricing override. store_id (int): the id of the store. store_pricing_scheme (dict): the store pricing scheme to use. """ if 'stores' in override \ and override['stores'] is not None \ and len(override['stores']) > 0 \ and store_id not in override['stores']: return pricing_result pricing_elements = map_override_to_pricing_elements( pricing_family.AUDIO_TRACK_PRICING_FAMILY_ID, override, store_pricing_scheme) for pricing_element in pricing_elements: pricing_element['track_ids'] = [override['track_id']] use_overlapping_dates = store_id in STORES_WITH_OVERLAPPING_DATES pricing_result = pricing_helper.apply_pricing_element_to_list( pricing_element, pricing_result, use_overlapping_dates) return pricing_result def map_override_to_pricing_elements( pricing_family_id, override, store_pricing_scheme): """Create a list of pricing elements from the override. Args: pricing_family_id: the pricing family ID. override (dict): the product pricing override. store_pricing_scheme (dict): the store pricing scheme. Returns: [dict]: a list of pricing elements """ pricing_element = {} if 'resolution' in override and override['resolution'] is not None: pricing_element['resolution'] = override['resolution'] if 'resolution' not in pricing_element \ and pricing_family_id == pricing_family.FILM_PRICING_FAMILY_ID: pricing_element['resolution'] = 'All' if 'start_date' in override and override['start_date'] is not None: pricing_element['start_date'] = override['start_date'] if 'end_date' in override and override['end_date'] is not None: pricing_element['end_date'] = override['end_date'] pricing_element['applies_worldwide'] = override['applies_worldwide'] if not override['applies_worldwide']: pricing_element['territories'] = override['territories'] pricing_element['territory_list_include'] = override[ 'territory_list_include'] if 'price_code' in override and override['price_code']: pricing_element['price_code'] = override['price_code'] pricing_element['custom_price'] = None pricing_element['custom_currency_code'] = None return [pricing_element] if 'custom_price' in override and override['custom_price']: pricing_element['custom_price'] = override['custom_price'] pricing_element['custom_currency_code'] = \ override['custom_currency_code'] return [pricing_element] elif 'orchard_pricing_tier_id' in override \ and override['orchard_pricing_tier_id']: if not store_pricing_scheme: return [] store_pricing_tier = None for tier in store_pricing_scheme['store_pricing_tiers']: if tier['orchard_pricing_tier_id'] == \ override['orchard_pricing_tier_id']: store_pricing_tier = tier if store_pricing_tier is None: return [] result = [] for code in store_pricing_tier['store_pricing_tier_codes']: code_element = { 'applies_worldwide': code['applies_worldwide'], 'territory_list_include': code['territory_list_include'], 'territories': code['territory_set'] } if 'resolution' in code and code['resolution'] is not None: code_element['resolution'] = code['resolution'] use_overlapping_dates = store_pricing_scheme['store_id'] in \ STORES_WITH_OVERLAPPING_DATES intersect_result = pricing_helper.intersect( code_element, pricing_element, use_overlapping_dates) if 'intersect' in intersect_result: intersection = intersect_result['intersect'] intersection['price_code'] = code['price_code'] intersection['store_pricing_tier_name'] = \ store_pricing_tier['name'] intersection['custom_price'] = None intersection['custom_currency_code'] = None result.append(intersection) return result else: return [] @cache.memoize(600) def load_legacy_default_price_codes(store_id, product_type_id, price_type): """Get the default legacy price codes. Args: store_id (int): the ID of the store. product_type_id (int): the ID of the product type. price_type (string): 'release' or 'track. Returns: response.Response: containing the legacy default price codes. """ return legacy_database.get_default_pricing_codes( store_id, product_type_id, price_type) @cache.memoize(600) def load_pricing_scheme( pricing_family_id, store_id, orchard_pricing_tier=None): """Get a store pricing scheme and all its tiers by family and store ID. Args: pricing_family_id (int): the ID of the pricing family. store_id (int): the ID of the store. orchard_pricing_tier (dict): the pricing tier of the product. Returns: response.Response: containing the store pricing scheme dict. """ result = store_pricing_scheme.get_by_pricing_family_and_store_id( pricing_family_id, store_id) if not result and orchard_pricing_tier and \ default_pricing_tiers[pricing_family_id]['fallback-to-default']: def_code = \ default_pricing_tiers[pricing_family_id]['default_price_code'] pricing_scheme = { 'store_pricing_tiers': [ { 'orchard_pricing_tier_id': orchard_pricing_tier['orchard_pricing_tier_id'], 'name': 'Default', 'store_pricing_tier_codes': [ { 'price_code': def_code, 'applies_worldwide': True, 'territory_list_include': True } ] } ] } result = response.Response(pricing_scheme) return result def remove_intervals(pricing_result): """Remove interval informations from the pricing results. Args: pricing_result ([dict]): A list with the pricing results Returns: The pricing results without interval informations """ pricing_result_without_interval = [] today = datetime.date.today() for item in pricing_result: if 'end_date' in item and \ isinstance(item['end_date'], datetime.datetime) and \ item['end_date'].date() <= today: continue if 'start_date' in item and \ isinstance(item['start_date'], datetime.datetime) and \ item['start_date'].date() > today: continue if 'start_date' in item: del item['start_date'] if 'end_date' in item: del item['end_date'] pricing_result_without_interval.append(item) return pricing_result_without_interval def get_us_price_code(pricing_result, pricing_family_id, track_id=None): """Get the US price code from the pricing results or return the default. Args: pricing_result ([dict]): A list with the pricing results pricing_family_id (int): the pricing family ID track_id (int): the track ID (optional) Returns: (string): the US price code """ for item in pricing_result: if track_id and track_id not in item['track_ids']: continue if 'US' in item['territories']: return item['price_code'] return default_pricing_tiers[pricing_family_id]['default_price_code'] def mark_items_needing_new_pricing( pricing_result, us_price_code, track_id=None): """Mark items with same price code as the US price code. Args: pricing_result ([dict]): A list with the pricing results us_price_code (int): the price code in the US track_id (int): the track ID (optional) Returns: (boolean): True if at least one item needs new pricing """ need_new_pricing = False for item in pricing_result: if track_id and track_id not in item['track_ids']: continue if item['territories'] != ['US'] and \ item['price_code'] == us_price_code: item['need_new_pricing'] = True need_new_pricing = True return need_new_pricing def get_store_pricing_tier_with_us_price_code(pricing_scheme, us_price_code): """Get the store pricing tier with the same US price code. Args: pricing_scheme (dict): the store pricing scheme us_price_code (int): the price code in the US Returns: (dict): the store pricing tier """ t = 'US' for spt in pricing_scheme['store_pricing_tiers']: for sptc in spt['store_pricing_tier_codes']: territories = set() if 'territory_set' in sptc: territories = sptc['territory_set'] applies_in_the_us = sptc['applies_worldwide'] or \ (sptc['territory_list_include'] and t in territories) or \ (not sptc['territory_list_include'] and t not in territories) if applies_in_the_us and sptc['price_code'] == us_price_code: return spt return None def apply_new_pricing(pricing_result, store_pricing_tier, track_id=None): """Apply new pricing to items that need it. Args: pricing_result ([dict]): A list with the pricing results store_pricing_tier (dict): the store pricing tier track_id (int): the track ID (optional) """ for item in pricing_result: if track_id and track_id not in item['track_ids']: continue if 'need_new_pricing' not in item or not item['need_new_pricing']: continue del item['need_new_pricing'] t = item['territories'][0] for sptc in store_pricing_tier['store_pricing_tier_codes']: territories = set() if 'territory_set' in sptc: territories = sptc['territory_set'] applies_to_territory = sptc['applies_worldwide'] or \ (sptc['territory_list_include'] and t in territories) or \ (not sptc['territory_list_include'] and t not in territories) if applies_to_territory: item['price_code'] = sptc['price_code'] item['store_pricing_tier_name'] = store_pricing_tier['name'] item['source'] = 'NewStorePrice' break def group_track_pricing_result_by_territory(pricing_result): """Group the pricing result by price code. Args: pricing_result ([dict]): the pricing result. Returns: [dict]: the grouped pricing result. """ result = [] result_by_price_code = {} for item in pricing_result: if 'start_date' in item or 'end_date' in item or \ 'custom_price' in item or 'custom_currency_code' in item: result.append(item) continue key = '${0}-${1}'.format(item['price_code'], str(item['track_ids'])) if key not in result_by_price_code: result_by_price_code[key] = deepcopy(item) if 'territories' not in result_by_price_code[key] or \ result_by_price_code[key]['territories'] is None: result_by_price_code[key]['territories'] = [] else: result_by_price_code[key]['territories'] += item['territories'] result += result_by_price_code.values() for item in result: item['territories'] = list(set(item['territories'])) item['territories'].sort() return result