"""Logic for Blacklist Words. Perform operations Related to blacklist words. """ from functools import cache from cachetools.func import ttl_cache import re from tempfile import NamedTemporaryFile import time from owsresponse import response from blacklist_manager import config from blacklist_manager.connectors import mysql from blacklist_manager.connectors import s3 from blacklist_manager.constants import error from blacklist_manager.constants import error_correction from blacklist_manager.constants import metadata from blacklist_manager.constants.caching import TTL_15_MIN from blacklist_manager.models import blacklist_reasons from blacklist_manager.models import blacklist_words from blacklist_manager.models import ows_product_workflow from blacklist_manager.models import persister from blacklist_manager.models.features import is_enabled_artist_field_blocklist_exact_match from blacklist_manager.models.features import is_enabled_store_blocked_artist_validation from blacklist_manager.validation import validation from blacklist_manager.validation.validation import json_validator POST_BLACKLIST_WORD_VALIDATOR = json_validator( config.POST_BLACKLIST_WORD_SCHEMA) PUT_BLACKLIST_WORD_VALIDATOR = json_validator( config.PUT_BLACKLIST_WORD_SCHEMA) STANDARD_REGEX = r'\b(?=){word}(?!\w)' PRODUCT_REGEX = r'.*(?<=[\W]){word}(?=[\W]).*' # lists of artists always look like "\nNAME1, " for one name or "\nNAME, NAME2, " etc for more than one name" # so this regex looks for the exact name match inside a newline or comma + space terminated by a comma + space # so as to get exact artist name matches ARTIST_REGEX = r'.*(?:(?<=, )|(?<=\n)){word}(?=, ).*' def get_blacklist_words(page_offset=None, page_limit=None, term=None): """Logic handlers. Args: page_offset (int): page offset. page_limit (int): total number of records per page. term (string): Blacklist term to match. Returns: Response: set of all the blacklist words. """ result = blacklist_words.get_blacklist_words( page_offset=page_offset, page_limit=page_limit, term=term) count = blacklist_words.get_blacklist_words_count(term=term) if count: result.message.get('pagination').update(count=count.message) return result def export_blacklist_words(): """Write all blacklist words to a csv on S3 and return download url. Returns: Response: Response with url and count of rows if successful. """ with NamedTemporaryFile( mode='w', delete=False, prefix='blacklist_') as file: s3_key = '{folder}/blacklist_words_{time}.csv'.format( folder=config.EXPORT_FOLDER_NAME, time=int(time.time())) write_result = blacklist_words.write_blacklist_words_to_file(file.name) if not write_result: return write_result upload_result = s3.upload_file_from_disk(s3_key, file.name) if not upload_result: return upload_result url_result = s3.get_signed_url(s3_key) if not url_result: return url_result write_result.message['download_url'] = url_result.message return write_result def delete_blacklist_word(blacklist_id): """Delete blacklist word. Args: blacklist_id (int): unique identifier of blacklist_word. Returns: response.Response: 200 success response if object exists else 404. """ return blacklist_words.delete_blacklist_word(blacklist_id) def create_blacklist_word(blacklist_word_data): """Create a new blacklist word. Args: blacklist_word_data (dict): properties to assign to the new blacklist word. Returns: response.Response: wrapper containing the created blacklist word id or errors. """ validation_response = validation.validate( blacklist_word_data, POST_BLACKLIST_WORD_VALIDATOR) input_validation_response = validation.detect_excel_formula(blacklist_word_data) if not input_validation_response: return input_validation_response if not validation_response: return validation_response sanitized_blacklist_word_data = validation.remove_zero_width_space_characters( blacklist_word_data ) result = blacklist_words.create_blacklist_word(sanitized_blacklist_word_data) return result def update_blacklist_word(blacklist_id, blacklist_word_data): """Update an existing blacklist word. Args: blacklist_id (int): unique identifier of blacklist_word. blacklist_word_data (dict): properties to update of the existing blacklist word. Returns: response.Response: wrapper containing the updated blacklist word id or errors. """ validation_response = validation.validate( blacklist_word_data, PUT_BLACKLIST_WORD_VALIDATOR) if not validation_response: return validation_response result = blacklist_words.update_blacklist_word( blacklist_id, blacklist_word_data) return result def _format_product_for_validation(product, correction): result_dict = {} items = product.items() for key, value in items: corr_val = ', '.join([str(item) for item in correction.get(key, [])]) if value or corr_val: value = value if value else '' new_val = value + ', ' + corr_val keys = result_dict.get(new_val, []) keys.append(key) result_dict[new_val] = keys return result_dict def _format_corrections_dict(corrections): result_dict = {} for correction in corrections: field_dict = error_correction.TRACK_CORRECTION_FIELDS if correction['table_name'] == error_correction.RELEASE_TYPE: field_dict = error_correction.RELEASE_CORRECTION_FIELDS matched_field = field_dict.get(correction['field_name']) if matched_field is not None: value = correction['key_value'] current_value = result_dict.get(matched_field, []) if (type(value) is list): value = [v.get('artist_name', v.get('name', '')) for v in value] current_value += value else: current_value.append(value) result_dict[matched_field] = current_value return result_dict def validate_product_for_blacklisting(product_id): """Validate if a product has blocklisted content.""" with mysql.ar_db_ro_session() as session: return _validate_product_for_blacklisting(product_id, session) def _validate_product_for_blacklisting(product_id, session): """Validate if a product has blacklisted content. Args: product_id (dict): Id of product to be validated for blacklisting. Returns: Response: Blacklisted Word and associated reason's data. """ product = persister.fetch_product_details_by_id(product_id, session) if product.status != 200: return product blacklist_words_data = blacklist_words.get_blacklist_words_for_validation(session) if blacklist_words_data.status != 200: return blacklist_words_data correction = ows_product_workflow.get_error_corrections(product_id) if correction.status != 200: return correction correction_data = {} if correction.message.get('status') != error_correction.APPLIED_STATUS: correction_data = _format_corrections_dict( correction.message.get('items', [])) product_data = _format_product_for_validation(product.message, correction_data) # pad the product data with newlines to make sure there are no regex issues product_text = '\n{}\n'.format('\n'.join(product_data.keys())) blacklisted_in_release = _is_blacklisted_text( product_text, blacklist_words_data.message, PRODUCT_REGEX, session, product_data) if blacklisted_in_release: return response.Response( message={'validation_error': blacklisted_in_release}, status=400) return response.Response(status=200) def validate_artists_for_blacklisting(input_data): """Validate if artists have blocklisted content.""" return _validate_artists_for_blacklisting(input_data) def _validate_artists_for_blacklisting(input_data): """Validate if artists have blocklisted content. Args: input_data (dict array): Artist data to be validated for blacklisting. Returns: Response: Blacklisted Word and associated reason's data. """ lookup_result = _get_blacklist_words_and_reasons( reason_ids=tuple(metadata.SPOTIFY_WATCHLIST_REASON_IDS) ) if lookup_result.status != 200: return lookup_result lookup_data = lookup_result.message track_artist_result = _validate_track_artists( input_data, lookup_data) if 'track_artists' in input_data else None product_artist_result = _validate_product_artists( input_data, lookup_data) if 'product_artists' in input_data else None errors = {} if track_artist_result: errors['track_artists'] = track_artist_result if product_artist_result: errors['product_artists'] = {'matches': product_artist_result } if errors: return response.Response( status=400, message={ 'product_id': input_data.get('product_id', None), 'validation_errors': errors }) return response.Response(status=200) def _validate_product_artists(input_data, blacklist_lookup): """Validate if product artists have blocklisted content. Args: input_data (dict): product artist payload. blacklist_lookup (tuple): blacklist word set, word-to-reason map, and reasons. Returns: list: Matched artist dicts augmented with 'reason'. Empty if no matches. """ blacklist_word_set, word_to_reason_id, reasons = blacklist_lookup artist_group = input_data['product_artists'] artist_match_results = _parse_artists( artist_group, blacklist_word_set, word_to_reason_id, reasons ) if blacklist_word_set else [] return artist_match_results def _validate_track_artists(input_data, blacklist_lookup): """Validate if track artists have blocklisted content. Args: input_data (dict): track artist payload. blacklist_lookup (tuple): blacklist word set, word-to-reason map, and reasons. Returns: list: Matched track groupings with tuid and matches. Empty if none. """ final_result = [] (blacklist_word_set, word_to_reason_id, reasons) = blacklist_lookup if blacklist_word_set: artist_groups = input_data['track_artists'] for grouping in artist_groups: group_result = { 'matches': [], 'tuid': grouping['tuid'] } if 'track_artists' in input_data: artist_match_results = _parse_artists( grouping['artists'], blacklist_word_set, word_to_reason_id, reasons ) group_result['matches'] = artist_match_results if group_result['matches']: final_result.append(group_result) return final_result @cache def _compile_regex(regex, word): return regex.format(word=re.escape(word)) @cache def _find_all_regex(regex, word, text): return re.findall(_compile_regex(regex, word), text, flags=re.IGNORECASE) @cache def _find_all_artist_exact_regex(word, text): return re.findall(_compile_regex(ARTIST_REGEX, word), text) @ttl_cache(maxsize=4, ttl=TTL_15_MIN) def _get_blacklist_words_and_reasons(reason_ids=None): """Get blacklist words and reasons. Args: reason_ids (optional list): reason ids to include in database results for blacklisting. Returns: Response: Concise data on blacklist words and reasons for lookup/ matching. """ with mysql.ar_db_ro_session() as session: if reason_ids: blacklist_words_result = blacklist_words.get_blacklist_words_by_reason_ids( session, reason_ids ) else: blacklist_words_result = blacklist_words.get_blacklist_words_for_validation(session) if blacklist_words_result.status != 200: _get_blacklist_words_and_reasons.cache_clear() return blacklist_words_result blacklist_word_set = set() word_to_reason_id = {} reasons = {} # Filter and build lookup structures in one pass for item in blacklist_words_result.message['items']: word = item['word'].lower() blacklist_word_set.add(word) word_to_reason_id[word] = item['reason_id'] # Fetch all reasons in one call if word_to_reason_id: reasons_result = blacklist_reasons.get_blacklist_reasons_by_id( list(set(word_to_reason_id.values())), session ) if reasons_result.status != 200: _get_blacklist_words_and_reasons.cache_clear() return reasons_result reasons = reasons_result.message # Return results return response.Response( message=(blacklist_word_set, word_to_reason_id, reasons), status=200 ) def _format_store_blocked_product_text(text_fields, text_fields_store_blocked): """Format store-blocked product text.""" if not text_fields: return None for terms, fields in text_fields.items(): filtered_entry = list(filter(lambda x: x in metadata.STORE_BLOCKED_FIELDS, fields)) if filtered_entry: text_fields_store_blocked[terms] = filtered_entry text_store_blocked = '\n{}\n'.format('\n'.join(text_fields_store_blocked.keys())) return text_store_blocked def _parse_artists(artist_group, blacklist_word_set, word_to_reason_id, reasons): """Parse artists using blacklisted words and reason lookup.""" results = [] for artist in artist_group: name = artist['name'].lower() if name in blacklist_word_set: reason_id = word_to_reason_id[name] artist_match = { 'name': artist['name'], 'reason': reasons[reason_id]['reason'], } if 'role' in artist: artist_match['role'] = artist['role'] elif 'type' in artist: artist_match['type'] = artist['type'] results.append(artist_match) return results def _is_blacklisted_text( text, blacklist_words_data, regex, session, text_fields=None): """Check if a text contains any blacklisted word. Args: text (string): text to be validated for blacklisting. blacklist_words_data (dict): Blacklist words data. regex: Regex to validated text_fields (dict): option fields to match against text Returns: Response: Blacklisted Words and associated reason's data. """ list_of_words_data = [] matched_blacklist_items = [] text_fields_store_blocked = {} text_store_blocked = _format_store_blocked_product_text( text_fields, text_fields_store_blocked) is_enabled_artist_field_blocklist_exact_match_result = is_enabled_artist_field_blocklist_exact_match() for data in blacklist_words_data['items']: word = data['word'].lower() # Condition that checks for 'store-blocked artist' reasons is_exclusive_store_blocked = data['reason_id'] in metadata.STORE_BLOCKED_REASON_IDS # Match on 'store-blocked artist' only eligible text if is_exclusive_store_blocked: if is_enabled_artist_field_blocklist_exact_match_result: word = data['word'] if text_store_blocked is not None and word in text_store_blocked: values_matched = _find_all_artist_exact_regex(data['word'], text_store_blocked) if text_fields_store_blocked: data['matched_fields'] = [text_fields_store_blocked[val][0] for val in values_matched] matched_blacklist_items.append(data) else: if text_store_blocked is not None and word in text_store_blocked.lower(): values_matched = _find_all_regex(regex, data['word'], text_store_blocked) if values_matched: if text_fields_store_blocked: data['matched_fields'] = [text_fields_store_blocked[val][0] for val in values_matched] matched_blacklist_items.append(data) # do a simple check # it saves a full second elif word not in text.lower(): continue else: # Match on the full text values_matched = _find_all_regex(regex, data['word'], text) if values_matched: if text_fields: data['matched_fields'] = [text_fields[val][0] for val in values_matched] matched_blacklist_items.append(data) if not matched_blacklist_items: return False # Fetch all reasons based on results from match operations reasons = blacklist_reasons.get_blacklist_reasons_by_id( [_x['reason_id'] for _x in matched_blacklist_items], session).message matched_blacklist_words = [_x['word'] for _x in matched_blacklist_items] is_store_blocked_artist_validation_enabled = \ is_enabled_store_blocked_artist_validation() store_blocked_artist_items = [] for item in matched_blacklist_items: if item.get('matched_fields') is not None: for field in item.get('matched_fields'): formatted_item = { 'word': item['word'], 'reason': reasons[item['reason_id']]['reason'], 'alert': reasons[item['reason_id']]['alert'], 'contact': reasons[item['reason_id']]['contact'], 'matched_field': field } # Isolate 'Store-Blocked Artist' (reason ID = 1) matches # into their own result so callers can act on them independently # of the other blacklist reasons. if (is_store_blocked_artist_validation_enabled and item['reason_id'] == metadata.STORE_BLOCKED_ARTIST_REASON_ID): store_blocked_artist_items.append(formatted_item) else: list_of_words_data.append(formatted_item) result = { 'matched_blacklist_words': matched_blacklist_words, 'items': list_of_words_data } if store_blocked_artist_items: result['store_blocked_artist'] = store_blocked_artist_items return result def validate_text_for_blacklisting(text_data): """Validate if text has blacklisted content. Args: text_data (dict): dict containing the text to check. Example: {"text": "Beyonce"} Returns: Response: Blacklisted Word and associated reason's data. """ if not text_data or not text_data.get('text'): return response.create_error_response( error.TEXT_MISSING_ERROR, error.ERROR_MESSAGE_MISSING_INPUT_TEXT) with mysql.ar_db_ro_session() as session: blacklist_words_data = blacklist_words.get_blacklist_words_for_validation( session) if not blacklist_words_data: return blacklist_words_data blacklisted = _is_blacklisted_text( text_data['text'], blacklist_words_data.message, STANDARD_REGEX, session, None) if blacklisted: return response.Response( message={'validation_error': blacklisted}, status=400) return response.Response(status=200)