"""Logic for Products.""" from oto import response from owsrequest import access from pythonfeatures import pythonfeatures import sentry_sdk from product import config from product.constants import error from product.constants import features from product.constants import header from product.constants import product as product_constants from product.models import country from product.models import ows_account from product.models import ows_carveouts from product.models import ows_carveouts_python from product.models import ows_marketing from product.models import ows_product_digital from product.models import ows_product_physical from product.models import ows_product_review from product.models import ows_timed_release from product.models import ows_video from product.models import product as product_model from product.models import release_artist from product.models import release_country_dates from product.models import release_subgenre from product.models import upc as upc_model from product.utils import feature_control_util def check_product_ownership(product_id, vendor_id=None, subaccount_id=None): """Check if a vendor or subaccount owns a product. Args: product_id (int): Product id of product to verify ownership of. vendor_id (int): Vendor id to verify product ownership for. subaccount_id (int): Subaccount id to verify product ownership for. Returns: Response: Response containing the result of the ownership check. """ product_response = product_model.get_product_by_id(product_id) if not product_response: return product_response product = product_response.message if not subaccount_id and not vendor_id: return response.Response(status=400) authorization = determine_authorization(product, vendor_id, subaccount_id) if not authorization: return authorization return response.Response(status=200) def get_product_by_upc(upc, vendor_id=None, subaccount_id=None): """Return product information from upc. Args: upc (int): the upc associated with a product Returns: response.Response: product information associated with upc """ product_response = product_model.get_product_by_upc(upc) if not product_response: return product_response if not subaccount_id and not vendor_id: return product_response product = product_response.message authorization = determine_authorization(product, vendor_id, subaccount_id) if not authorization: return authorization return product_response def get_product_by_product_id( product_id, vendor_id=None, subaccount_id=None, with_tenant_uuids=False ): """Return product information by product_id. Args: product_id (int): the product_id associated with a product vendor_id (int): Vendor identifier. subaccount_id (int): Subaccount identifier. with_tenant_uuids (bool): Include 4 tenant level UUID fields in result or not. Returns: response.Response: product information associated with product_id """ product_response = product_model.get_product_by_id(product_id) if not product_response: return product_response if with_tenant_uuids: tenant_response = product_model.get_tenant_uuids(product_id) if not tenant_response: return tenant_response product_response.message.update(tenant_response.message) if not subaccount_id and not vendor_id: return product_response product = product_response.message authorization = determine_authorization(product, vendor_id, subaccount_id) if not authorization: return authorization return product_response def get_product_id_by_product_code(product_code, account_type, account_uuid): """Return product_id information by product_code and account. Args: product_code (str): the product_code associated with a product account_type (str): vendor or subaccount account_uuid (UUID): the id if vendor or subaccount Returns: response.Response: product information associated with product_code and the account """ product_id_response = product_model.get_product_id_by_product_code_for_account( product_code, account_type, account_uuid ) if not product_id_response: return product_id_response return response.Response(message={"product_id": product_id_response.message}) def determine_authorization(product, vendor_id, subaccount_id): """Determine if request is authorized. Args: product (dict): Product for which to check ownership. vendor_id (int): Vendor id for which to check ownership. subaccount_id (int): Subaccount id for which to check ownership. returns: response.Response: Status=200 if allowed, status=403 otherwise. """ if vendor_id and not is_vendor_product_owner(product, vendor_id): return response.create_error_response( code=error.ERROR_CODE_AUTHORIZATION, message=error.ERROR_MESSAGE_FORBIDDEN_USER, status=403, ) if subaccount_id and not is_subaccount_product_owner(product, subaccount_id): return response.create_error_response( code=error.ERROR_CODE_AUTHORIZATION, message=error.ERROR_MESSAGE_FORBIDDEN_USER, status=403, ) return response.Response(status=200) def is_vendor_product_owner(product, vendor_id): """Check if a vendor owns a product. Args: product (dict): Product to check ownership of. vendor_id (int): Vendor id to check ownership for. returns: bool: Result of ownership check. """ return str(vendor_id) == str(product.get("vendor_id")) def is_subaccount_product_owner(product, subaccount_id): """Check if a subaccount owns a product. Args: product (dict): Product to check ownership of. subaccount_id (int): Subaccount id to check ownership for. returns: bool: Result of ownership check. """ return str(subaccount_id) == str(product.get("subaccount_id")) def product_code_found_for_account( product_code, account_type, account_id, exclude_product_id=0 ): """Check if product code is found for an account. Args: product_code (str): product code to find. account_type (str): the account type - one of subaccount or vendor. account_id (int): the account_id of the above type. exclude_product_id (int): Exclude this product id Returns: response.Response: no message with status of 200 or 404. """ model_result = product_model.get_first_product_by_product_code_for_account( product_code, account_type, account_id ) status = model_result.status if model_result.status == 200 and exclude_product_id: if model_result.message["product_id"] == exclude_product_id: status = 404 return response.Response(status=status) def product_codes_available_for_account( product_codes, account_type, account_uuid ): """Check if product codes are available for an account. Args: product_codes (list[str]): product codes to find. account_type (str): the account type - one of subaccount or vendor. account_uuid (UUID): the account_id of the above type. Returns: response.Response: a list of available codes with status of 200 """ codes_in_use = product_model.get_products_by_product_code_for_account( product_codes, account_type, account_uuid ).message available_codes = set(product_codes or []) - set(codes_in_use or []) return response.Response(message={"available_product_codes": list(available_codes)}) def delete_product(product_id, account_type=None, account_id=None): """Delete a product by product_id. Args: account_type (str): Account type to verify product ownership for. account_id (int): Account id to verify product ownership for. product_id (int): Product id of product to verify ownership of. returns: response.Response: Status=200 if deleted """ product = get_product_by_product_id(product_id) if not product: return product validation = access.verify_grass_access( account_type, account_id, vendor=product.message.get("vendor_id"), subaccount=product.message.get("subaccount_id"), ) if not validation: return validation if account_id and account_type: ownership_response = check_product_ownership( product_id, **{account_type + "_id": int(account_id)} ) if not ownership_response: return response.Response(status=ownership_response.status) if product.message.get("context_type") == "physical": ows_carveouts.delete_carveout(product.message.get("upc")) ows_carveouts_python.delete_release_dms_master_carveout(product_id) ows_carveouts_python.delete_release_dms_carveout(product_id) ows_marketing.delete_marketing_highlights_by_release(product_id) return ows_product_physical.delete_physical_product(product_id) if ( product.message.get("distribution_format_id") == product_constants.VIDEO_DISTRIBUTION_FORMAT_ID ): ows_carveouts.delete_carveout(product.message.get("upc")) ows_carveouts_python.delete_release_dms_master_carveout(product_id) ows_carveouts_python.delete_release_dms_carveout(product_id) release_subgenre.delete(product_id) release_artist.delete(product_id) return ows_video.delete_product(product_id) if ( product.message.get("distribution_format_id") == product_constants.MUSIC_DISTRIBUTION_FORMAT_ID and product.message.get("context_type") == "digital" ): ows_carveouts.delete_carveout(product.message.get("upc")) ows_carveouts_python.delete_release_dms_master_carveout(product_id) ows_carveouts_python.delete_release_dms_carveout(product_id) ows_marketing.delete_marketing_highlights_by_release(product_id) ows_timed_release.delete_timed_release(product_id) return ows_product_digital.delete_digital_product( product_id, account_type, account_id ) return response.Response(status=400, message="Not Supported") upc_unavailable_response = response.create_error_response( status=403, code=error.ERROR_CODE_UPC_UNAVAILABLE, message=error.ERROR_MESSAGE_UPC_UNAVAILABLE, ) def check_display_upc_availability(account_type, account_id, display_upc, context_type): """Check availability of display_upc for the given vendor or subaccount. Args: account_type (str): Account type to check account_id (int): Vendor or Subaccount id for product display_upc (str): Display UPC value to check context_type (str): Context type of product - `physical` or `digital` Returns: response.Response: object with status based on query result """ # ensure display_upc is not yet used by this account for this context_type context_use_response = product_model.is_display_upc_used_by_account( account_type, account_id, display_upc, context_type ) if context_use_response.status == 500: return context_use_response if context_use_response.message: return upc_unavailable_response # check if display_upc is an orchard owned upc orchard_upc_response = upc_model.get_upcs_by_values([display_upc]) if orchard_upc_response.status == 500: return orchard_upc_response if orchard_upc_response.status == 404: return response.Response() # ensure the orchard owned upc is in use if orchard_upc_response.message[0]["status"] != "used": return upc_unavailable_response # Check feature control 'Physical UPC Generation' enabled for subaccount # to ensure that if display_upc is auto generated then no need to check # further as it should be unique at account level if account_type == header.GRASS_ACCOUNT_TYPE_SUBACCOUNT: is_feature_control_available = ( feature_control_util.check_feature_control_for_subaccount( account_id, features.PHYSICAL_UPC_GENERATION ) ) if is_feature_control_available is True: return response.Response() # ensure display_upc has been used by this and only this account account_use_response = product_model.is_display_upc_used_by_account( account_type, account_id, display_upc ) if account_use_response.status == 500: return account_use_response if not account_use_response.message: return upc_unavailable_response return response.Response() def get_products_for_account( vendor_id=None, subaccount_id=None, status=None, page_offset=None, page_limit=None, start_date=None, end_date=None, sort_order=None, deletions=None, sort_by=None, ): """Get list of products for a vendor or subaccount, filtered by status. Args: vendor_id (int): The vendor's account id subaccount_id (int): The subaccount_id's account id status (str): Status to filter products by page_offset (int): Offset used for pagination page_limit (int): Max number of records to return start_date (date): start date to filter the products by release date. end_date (date): end date to filter the products by release date. sort_order (str): sort by {sort_by} asc or desc deletions (str): Filter by a specific deletion status. sort_by (str): Column name the query result should be sorted by. If none specified or value is invalid, defaults to release_date. Returns: response.Response: Paginated list of products for a given vendor. """ if not subaccount_id and not vendor_id: return response.create_error_response( code=error.ERROR_CODE_BAD_REQUEST, message=error.ERROR_MESSAGE_NO_ACCOUNT_GIVEN, status=400, ) product_response = product_model.get_products_by_account_id( vendor_id=vendor_id, subaccount_id=subaccount_id, status=status, page_offset=page_offset, page_limit=page_limit, start_date=start_date, end_date=end_date, sort_order=sort_order, deletions=deletions, sort_by=sort_by, ) return product_response def get_upcs_by_product_ids(product_ids): """Get list of upcs for particular product ids. Args: product_ids (str): comma separated list of product_ids Returns: response.Response: list of upcs. """ if not product_ids: return response.create_error_response( code=error.ERROR_CODE_NOT_FOUND, message=error.ERROR_MESSAGE_MISSING_PRODUCT_ID, status=404, ) if not product_ids.replace(",", "").isdigit(): return response.create_error_response( code=error.ERROR_CODE_BAD_REQUEST, message=error.ERROR_MESSAGE_INVALID_CHARACTER, status=400, ) # filter will remove extraneous commas. # map will convert all the product ids to int. # set is used to remove duplicate product ids. product_ids_list = set(map(int, filter(None, product_ids.split(",")))) if len(product_ids_list) > int(config.LIMIT): return response.create_error_response( code=error.ERROR_CODE_BAD_REQUEST, message=error.ERROR_MESSAGE_MAX_PRODUCT, status=400, ) products_detail_data = product_model.get_upcs_by_product_ids(product_ids_list) if not products_detail_data: return products_detail_data product_data_items = products_detail_data.message["items"] if len(product_data_items) != len(product_ids_list): found_product_ids_list = [ product["product_id"] for product in product_data_items ] not_found_product_ids_list = product_ids_list.difference(found_product_ids_list) error_message_dict = { "code": error.ERROR_CODE_NOT_FOUND, "message": error.ERROR_MESSAGE_INVALID_PRODUCT_ID, } not_found_product_id_responses = [ {product_id: error_message_dict} for product_id in not_found_product_ids_list ] return response.create_error_response( code=error.ERROR_CODE_NOT_FOUND, message=not_found_product_id_responses, status=404, ) return products_detail_data def get_products_by_upcs(data): """Get list of products for particular upcs. Args: data (dict): list of upcs Returns: response.Response: list of products. """ if "upcs" not in data: return response.create_error_response( code=error.ERROR_CODE_BAD_REQUEST, message=error.ERROR_MESSAGE_MISSING_UPC, status=400, ) upcs = data["upcs"] upcs_list = [int(upc) for upc in upcs] if not upcs_list: return response.create_error_response( code=error.ERROR_CODE_BAD_REQUEST, message=error.ERROR_MESSAGE_MISSING_UPC, status=400, ) if len(upcs_list) > int(config.LIMIT): return response.create_error_response( code=error.ERROR_CODE_BAD_REQUEST, message=error.ERROR_MESSAGE_MAX_UPC, status=400, ) sort_param = data.get("sort_param") sort_dir = data.get("sort_direction") or "ASC" results = product_model.get_products_by_upcs(upcs_list, sort_param, sort_dir) if not results: return results return response.Response({"items": results.message}) def get_product_id_by_upc_dataloaded(data): """Get list of products for particular upcs. Args: data (dict): list of upcs Returns: response.Response: list of products. """ if "upcs" not in data: return response.create_error_response( code=error.ERROR_CODE_BAD_REQUEST, message=error.ERROR_MESSAGE_MISSING_UPC, status=400, ) upcs = data["upcs"] upcs_list = [int(upc) for upc in upcs] if not upcs_list: return response.create_error_response( code=error.ERROR_CODE_BAD_REQUEST, message=error.ERROR_MESSAGE_MISSING_UPC, status=400, ) if len(upcs_list) > int(config.LIMIT): return response.create_error_response( code=error.ERROR_CODE_BAD_REQUEST, message=error.ERROR_MESSAGE_MAX_UPC, status=400, ) results = product_model.get_product_id_by_upc_dataloaded(upcs_list) if not results: return results return response.Response({"items": results.message}) def get_product_document( product_id, with_company_brand=False, with_tenant_uuids=False): """Get the product document (for cloudsearch etl). Args: product_id (int): id of product with_company_brand (bool): Include company brand fields in result or not. with_tenant_uuids (bool): Include 4 tenant level UUID fields in result or not. Returns: response.Response: list of products. """ results = product_model.get_product_document_by_id( product_id, with_company_brand, with_tenant_uuids) if not results: return results product_doc = results.message try: display_status = calculate_display_status(product_doc) product_doc["display_status"] = display_status except IncorrectDisplayStatusError as exception: sentry_sdk.capture_exception(exception) finally: del product_doc["correction_status"] del product_doc["approval_queue_status"] return response.Response(product_doc) def get_products_documents(product_ids): """Get the products documents. Args: product_ids (str): comma separated list of product ids Returns: response.Response: list of products. """ return product_model.get_products_documents_by_ids(product_ids) class IncorrectDisplayStatusError(Exception): """Raised when we have an incorrect combination of statuses.""" def __init__( self, release_status, release_correction_status, release_approval_queue_status ): """Constructor.""" message = ( "Incorrect status combination. release.release_status: {}," "release_correction.status: {}," "release_approval_queue.status: {}".format( release_status, release_correction_status, release_approval_queue_status ) ) super().__init__(message) def calculate_display_status(product_doc): """Calculate display_status for product. Args: product_doc (dict): Product document object Returns: str: calculated display status """ review_status = get_review_status( product_doc.get("release_id"), product_doc["vendor_id"], product_doc["distribution_format_id"], product_doc["approval_queue_status"], product_doc["vendor_owner"], product_doc["assigned_to"], product_doc["assigned_reviewer"], ) return resolve_display_status( product_doc["release_status"], product_doc["correction_status"], review_status, ) def get_review_status( release_id, vendor_id, distribution_format_id, approval_queue_status, owner, assigned_to, assigned_reviewer, ): """Get review status for a product.""" company_brand = ows_account.get_vendor_company_brand(vendor_id) if company_brand.status != 200: raise Exception(company_brand.errors) attributes = { "vendor_id": str(vendor_id), "distribution_format_id": str(distribution_format_id), "company_brand_name": company_brand.message, "owner": owner, "assigned_to": str(assigned_to), "assigned_reviewer": str(assigned_reviewer), } content_review_enabled = ( pythonfeatures.get_single_feature_by_attributes( features.CONTENT_REVIEW_PRODUCT_SPLIT, attributes ).message == "enabled" ) if not content_review_enabled: return approval_queue_status product_review_response = ows_product_review.get_product_review_status(release_id) if product_review_response.status not in [200, 404]: raise Exception(product_review_response.message) if product_review_response.status == 200: return product_review_response.message["release_approval_status"] return approval_queue_status def resolve_display_status(release_status, correction_status, review_status): """Resolve the display status for a product.""" display_status = None if release_status in ( product_constants.LABEL_PROCESSING, product_constants.ORCHARD_PROCESSING, ): display_status = get_processing_status(review_status) elif release_status == product_constants.TRANSFER_TO_CONTENT: display_status = get_transfer_to_content_status(review_status) elif release_status == product_constants.IN_CONTENT: display_status = get_in_content_status(correction_status, review_status) if display_status is None: raise IncorrectDisplayStatusError( release_status, correction_status, review_status ) return display_status def get_processing_status(review_status): """Get the status of an in process product.""" if review_status == product_constants.REJECTED: return product_constants.ACTION_REQUIRED return product_constants.IN_PROGRESS def get_transfer_to_content_status(review_status): """Get the status of a transfer_to_content product.""" if review_status in ( product_constants.CHECKED_IN, product_constants.CHECKED_OUT, None, ): return product_constants.SUBMITTED elif review_status == product_constants.REJECTED: return product_constants.ACTION_REQUIRED def get_in_content_status(correction_status, review_status): """Get the status of an in_content product.""" if correction_status == product_constants.SUBMITTED and review_status in ( product_constants.CHECKED_IN, product_constants.CHECKED_OUT, ): return product_constants.SUBMITTED elif correction_status in (None, product_constants.APPLIED): return product_constants.COMPLETED elif correction_status == product_constants.ACTIVE: if review_status in ( product_constants.CHECKED_IN, product_constants.CHECKED_OUT, product_constants.APPROVED, None, ): return product_constants.ERROR_CORRECTION elif review_status == product_constants.REJECTED: return product_constants.ACTION_REQUIRED def get_product_by_display_upc(vendor_id, display_upc, context_type, show_deletions=False): """Get a product whose vendor, display_upc, and context match the search. Arguments: vendor_id (int): id of vendor display_upc (str): display_upc of the product context_type (str): distribution type of the product (digital|physical) show_deletions (bool): Whether to return deleted products or not Returns: response.Response: list of products """ result = product_model.get_product_by_display_upc( vendor_id, display_upc, context_type, show_deletions ) if not result: return result if not result.message: return response.create_not_found_response() if not show_deletions: if len(result.message) != 1: return response.create_error_response( code=error.ERROR_CODE_INVALID_DATA, message=error.ERROR_MESSAGE_UNIQUE_EXCEPTION, status=500, ) else: # remove duplicate upcs due to deletion only if len(result.message) > 1: not_deleted = [record for record in result.message if record["deletions"] == "N"] # if all display upcs are NOT deleted, assign the filtered list to the message if len(not_deleted) != 0: result.message = not_deleted upc_response = result.message[0] return response.Response(message=upc_response) def check_products_ownership_by_upcs(upcs, vendor_id=None, subaccount_id=None): """Check products ownership by upc(s) and account id. Args: upcs (list): List of upcs. vendor_id (int): Vendor id to verify product ownership for. subaccount_id (int): Subaccount id to verify product ownership for. Returns: Response: Response containing the result of the ownership check. """ if not vendor_id and not subaccount_id: return response.create_error_response( code=error.ERROR_CODE_BAD_REQUEST, message=error.ERROR_MESSAGE_INVALID_ACCOUNT, ) result = product_model.get_products_by_accountid_and_upcs( upcs, vendor_id, subaccount_id ) account_id = vendor_id or subaccount_id if not result.message: return response.create_not_found_response( message=error.ERROR_MESSAGE_UPCS_NOT_FOUND.format( upcs=",".join(upcs), account_id=account_id ) ) products = result.message product_upcs = [str(product["upc"]) for product in products] upc_diff = list(set(upcs) - set(product_upcs)) if upc_diff: return response.create_not_found_response( message=error.ERROR_MESSAGE_UPCS_NOT_FOUND.format( upcs=",".join(upc_diff), account_id=account_id ) ) return response.Response( message=dict(message="All upc(s) belong to label id {}".format(account_id)), status=200, ) def get_products_by_isrc(isrc): """Get the products for a given ISRC. Args: isrc (string): isrc to find products for Returns: response.Response: list of products. """ return product_model.get_products_by_isrc(isrc) def get_product_by_release_artist_id(release_artist_id): """Get the product for a given release_artist_id. Args: release_artist_id (int): release_artist_id to find product for Returns: response.Response: product data. """ return product_model.get_product_by_release_artist_id(release_artist_id) def replace_product_territory_dates(product_id, data): """Deletes all territory dates for given product_id then insert new ones. Args: product_id (int): Product ID data (list): List of territory release dates. Returns: bool """ product = product_model.get_product_by_id(product_id).message country_codes = set() for row in data: country_codes = country_codes | set(row.get("country_codes")) country_data = country.get_country_ids(list(country_codes)) country_map = {} for row in country_data: country_map[row.get("country_code")] = row.get("id") payload = [] for row in data: for country_code in row.get("country_codes"): country_id = country_map.get(country_code) if not country_id: continue if ( not row.get("release_date") and not row.get("sale_start_date") and not row.get("vod_start_date") and not row.get("preorder_date") ): continue record = dict() record["upc"] = product.get("upc") record["release_id"] = product_id record["release_date"] = row.get("release_date") record["sale_start_date"] = row.get("sale_start_date") record["vod_start_date"] = row.get("vod_start_date") record["preorder_date"] = row.get("preorder_date") record["country_id"] = country_map.get(country_code) payload.append(record) release_country_dates.delete_all_by_product_id(product_id) release_country_dates.insert_dates(payload) return True def lookup_product_ownership(product_ids: list[int]) -> response.Response: """Call model to get product ownership.""" result = product_model.get_product_ownership_by_product_id_dataloaded(product_ids) if result: return response.Response(message={"product_ids": result.message}) return response.Response(message={"product_ids": [None] * len(product_ids)})