from collections import defaultdict from collections import namedtuple import csv from functools import partial import io from oto import response from masters_registry import config from masters_registry.connectors import s3 from masters_registry.constant import bulk_tasks_const from masters_registry.constant import error from masters_registry.constant import field_const from masters_registry.models import bulk_tasks from masters_registry.models import ownership from masters_registry.models import ows_carveouts from masters_registry.models import users def create_task(correlation_id, user_id, task_type, count, context): """Create new item in task_status table Args: correlation_id (str): correlation id user_id (int): user id task_type (str): task type count (int): amount of items in batch context (list): task context (list of ISRCs or UPCs) Returns: int: ID of new created task """ user_name = '' user_name_response = users.get_orchard_user_names(list([user_id])) if user_name_response: user_name = user_name_response.message[user_id] task_id = bulk_tasks.create_task( correlation_id, user_id, task_type, count, context, user_name) return task_id def get_tasks(num_records, order_by, order_direction, page_offset): """Get a list of bulk processing tasks Args: num_records (int): number of task statuses to be returned order_by (str): create_datetime or finish_datetime order_direction (str): asc or desc page_offset (int): number of records to skip Returns: response.Response: containing list of tasks dicts """ statues_count = bulk_tasks.get_task_statuses_count() tasks = bulk_tasks.get_task_report( num_records, order_by, order_direction, page_offset ) return response.Response( { field_const.ITEMS: tasks, field_const.TOTAL_COUNT: statues_count } ) def _calculate_status(upc_row, report_type): """Calculate upc status field value. Depends on `success_count`, `error_count` and `status_report` fields Args: upc_row (dict): contains upc data report_type (str): UPC|ISRC report type Returns: str: one of returned statuses of UPC: DONE | PARTIAL_FAIL(only for UPC report type) | FAILED """ success = upc_row[field_const.SUCCESS_COUNT] error = upc_row[field_const.ERROR_COUNT] if success > 0 and error == 0: result_status = bulk_tasks_const.UPDATE_STATUS_SUCCESS elif success > 0 and error > 0: result_status = bulk_tasks_const.UPDATE_STATUS_PARTIAL_FAIL elif _is_claimed_by_another_owner(upc_row): result_status = bulk_tasks_const.UPDATE_STATUS_PARTIAL_FAIL else: result_status = bulk_tasks_const.UPDATE_STATUS_FAIL if (report_type == field_const.ISRC and result_status == bulk_tasks_const.UPDATE_STATUS_PARTIAL_FAIL): result_status = bulk_tasks_const.UPDATE_STATUS_FAIL return result_status def _is_claimed_by_another_owner(upc_row): """ Check if status_report list contains item, which has non-empty claimed_by_another_owner list. Args: upc_row (dict): upc data for report, contains status_report, which is list of dicts with statuses for each ISRC of this UPC Returns: bool """ for isrc_dict in upc_row['status_report']: if isrc_dict.get(field_const.CLAIMED_BY_ANOTHER_OWNER): return True return False def _generate_update_by_upc_report(csv_writer, task_info, correlation_id): """Generating update report by UPC. Args: csv_writer (csv.writer): instance of csv.writer() task_info (TaskStatus): instance of TaskStatus model correlation_id (str): correlation ID for logs """ header_row = bulk_tasks_const.UPC_IMPORT_UPC_REPORT_HEADER csv_writer.writerow(header_row) for upc_row in task_info.result: upc = '="{}"'.format(upc_row[field_const.UPC]) upc_status = _calculate_upc_status(upc_row) successful_isrc_count = upc_row[field_const.SUCCESS_COUNT] failed_isrc_count = upc_row[field_const.ERROR_COUNT] vendor_id = '' reports = upc_row[field_const.STATUS_REPORT] if reports: # All ISRCs of given UPC are supposed to have same # vendor_id values. vendor_id = reports[0].get(field_const.VENDOR_ID, '') row = [vendor_id, upc, upc_status, successful_isrc_count, failed_isrc_count] csv_writer.writerow(row) def _calculate_upc_status(upc_row): """Calculates UPC status that will be placed in the UPC report. Args: upc_row (dict): contains upc data Returns: str: UPC status """ statuses = [] for isrc in upc_row[field_const.STATUS_REPORT]: _, status = _get_isrc_status(isrc) statuses.append(status) return _get_upc_status(statuses) def _get_failed_reason_when_claimed_by_another_owner( claimed_by_another_owner): """Make error message when isrc territory claimed by another owner(s). Gets track info from art_relations by tuid(s). If multiple owners claimed different territories, error_template will be repeated for each owner. Args: claimed_by_another_owner (list): [(country_code, tuid)] Returns: str: error message, if multiple owners provided, .e.g Territories Already Claimed by Label ID 11: UK, US. """ error_template = bulk_tasks_const.INTERNAL_CONFLICT_CREATED_WARNING errors = [] territories = _group_territories_by_tuid(claimed_by_another_owner) tracks = ownership.get_tracks(list(territories.keys())).message for tuid, data in tracks.items(): label_id = data[field_const.VENDOR_ID] countries = ','.join(territories[tuid]) errors.append(error_template.format( tuid=label_id, countries=countries)) error_message = '.'.join(errors) return error_message def _group_territories_by_tuid(claimed_territories): """Convert list of tuples into dict for easier manipulation. Args: claimed_territories (list): [(country_code, tuid)] [('US', 182), ('UK', 182), ('UA', 183)] Returns: dict: {tuid: [country_codes]} """ result = defaultdict(list) for territory, tuids in claimed_territories: if isinstance(tuids, list): for tuid in tuids: result[tuid].append(territory) else: result[tuids].append(territory) return result def _get_upc_update_message(upc_status): """Returns UPC Update Message. Args: upc_status (str): upc status Returns: str: UPC Update Message """ if upc_status in (bulk_tasks_const.SUCCESS, bulk_tasks_const.WARNING): message = bulk_tasks_const.NA else: message = error.TRACK_LEVEL_FAILURE return message def _get_substore_carveouts_update_message(substore_info, correlation_id): """Returns substore carveouts update message. Args: substore_info (list): substore carveouts information correlation_id (str): correlation ID for logs Returns: str: substore carveouts update message """ # TODO: iterate over a list of carveouts when we add support of # other DMS stores. store_id = substore_info[0] carveouts_response = ows_carveouts.get_store_name( store_id, correlation_id) if carveouts_response: store_name = carveouts_response.message else: store_name = store_id message = error.ERROR_SUBSTORE_CARVEOUTS.format( store_name=store_name, territories=','.join(sorted(substore_info[1]))) return message def _generate_update_by_isrc_report(csv_writer, task_info, correlation_id): """Generating update report by ISRC Args: csv_writer (csv.writer): instance of csv.writer() task_info (TaskStatus): instance of TaskStatus model correlation_id (str): correlation ID for logs """ header_row = bulk_tasks_const.UPC_IMPORT_ISRC_REPORT_HEADER csv_writer.writerow(header_row) rows = _generate_rows(correlation_id, task_info) csv_writer.writerows(rows) def _generate_rows(correlation_id, task_info): """Generates rows for ISRC report. Args: correlation_id (str): correlation ID for logs task_info (TaskStatus): instance of TaskStatus model Returns: list: ISRC report rows """ ResultRow = namedtuple( 'Row', ['upc', 'isrc', 'isrc_status', 'isrc_update_message']) rows = [] for upc_row in task_info.result: upc = '="{}"'.format(upc_row[field_const.UPC]) isrc_rows = [] isrc_upc_statuses = [] for isrc in upc_row[field_const.STATUS_REPORT]: isrc_status, isrc_upc_status = _get_isrc_status(isrc) isrc_upc_statuses.append(isrc_upc_status) isrc_update_message = _get_isrc_update_message( isrc, correlation_id) isrc_string = isrc[field_const.ISRC] isrc_rows.append( ResultRow(upc, isrc_string, isrc_status, isrc_update_message)) upc_status = _get_upc_status(isrc_upc_statuses) for i, isrc in enumerate(upc_row[field_const.STATUS_REPORT]): upc_update_message = _get_upc_update_message(isrc_upc_status) vendor_id = isrc.get(field_const.VENDOR_ID, '') tuid = isrc.get(field_const.TUID, '') isrc_row = isrc_rows[i] rows.append([ vendor_id, isrc_row.upc, upc_status, upc_update_message, tuid, isrc_row.isrc, isrc_row.isrc_status, isrc_row.isrc_update_message ]) return rows def _get_isrc_update_message(isrc, correlation_id): """Returns ISRC Update Message. Args: isrc (dict): result of processing ISRC correlation_id (str): correlation ID for logs Returns: str: ISRC Update Message """ error_message = isrc[field_const.ERROR_MESSAGE] update_messages = [] claimed_by_another_owner = isrc.get(field_const.CLAIMED_BY_ANOTHER_OWNER) claimed_by_the_same_label = isrc.get(field_const.CLAIMED_BY_THE_SAME_LABEL) resolved_conflict = isrc.get(field_const.RESOLVED_CONFLICT) substore_carveouts = isrc.get(field_const.SUBSTORE_CARVEOUTS) if isrc.get(field_const.LOCKED_TERRITORIES): update_messages.append('Locked Territories: {territories}'.format( territories=','.join(isrc[field_const.LOCKED_TERRITORIES])) ) if resolved_conflict: update_messages.append( bulk_tasks_const.INTERNAL_CONFLICT_RESOLVED_WARNING.format( ','.join(isrc[field_const.REMOVED_TERRITORIES]))) if claimed_by_another_owner: update_messages.append( _get_failed_reason_when_claimed_by_another_owner( claimed_by_another_owner)) if claimed_by_the_same_label: update_messages.append( bulk_tasks_const.CLAIMED_BY_THE_SAME_LABEL_ERROR.format( ','.join(claimed_by_the_same_label))) if error_message.startswith(error.UPC_YOUTUBE_CARVED_OUT): update_messages.append(error.UPC_YOUTUBE_CARVED_OUT_MESSAGE) if substore_carveouts: substore_info = substore_carveouts[0] territories = substore_info[1] if territories: update_messages.append( _get_substore_carveouts_update_message( substore_info, correlation_id)) update_messages.sort(key=len) update_message = ' | '.join(update_messages) if (error_message and not error_message.startswith(error.UPC_YOUTUBE_CARVED_OUT)): update_message = '{0} | {1}'.format(error_message, update_message) return update_message def _get_upc_status(statuses): """Calculates UPC status that will be placed in the report. Args: statuses (list): list of ISRC statuses Returns: str: UPC status """ failed_count = len([s for s in statuses if s == bulk_tasks_const.FAIL]) all_failed = failed_count == len(statuses) some_failed = not all_failed and failed_count > 0 if all_failed: upc_status = bulk_tasks_const.UPDATE_STATUS_FAIL elif some_failed: upc_status = bulk_tasks_const.UPDATE_STATUS_PARTIAL_FAIL else: upc_status = bulk_tasks_const.SUCCESS return upc_status def _get_isrc_status(isrc_result): """Calculates ISRC update status Args: isrc_result (dict): result of processing ISRC Returns: tuple: ISRC status that will be placed in the report, ISRC status that will be used to calculate UPC status """ success = isrc_result.get(field_const.SUCCESS) new_conflicts = isrc_result.get(field_const.CLAIMED_BY_ANOTHER_OWNER) resolved_conflicts = isrc_result.get(field_const.RESOLVED_CONFLICT) sub_carveout = isrc_result.get(field_const.SUBSTORE_CARVEOUTS) claimed = isrc_result.get(field_const.CLAIMED_BY_THE_SAME_LABEL) locked_territories = isrc_result.get(field_const.LOCKED_TERRITORIES) error_message = isrc_result.get(field_const.ERROR_MESSAGE) carveout = error_message == error.UPC_YOUTUBE_CARVED_OUT if any([new_conflicts, resolved_conflicts, locked_territories, sub_carveout, carveout]): isrc_status = bulk_tasks_const.WARNING elif claimed: isrc_status = bulk_tasks_const.FAIL elif success: isrc_status = bulk_tasks_const.SUCCESS else: isrc_status = bulk_tasks_const.FAIL if resolved_conflicts: isrc_upc_status = bulk_tasks_const.SUCCESS else: isrc_upc_status = isrc_status return isrc_status, isrc_upc_status def _generate_bulk_resolve_conflicts_report(csv_writer, task_info): """Generating bulk resolve conflicts report Note that this function assumes that task_info.result is in this format: { 'account_id': '12345', 'successful_isrcs': { 'QA1': {'tuid1': ['AF', 'AD'], 'tuid2': ['CA']}, 'QA2': {'tuid2': ['AF', 'AD'], 'tuid3': {}}, }, 'failed_isrcs': {'QA1': ['tuid1', 'tuid2']}, } Args: csv_writer (csv.writer): instanse of csv.writer() task_info (TaskStatus): task status object """ csv_writer.writerow(bulk_tasks_const.BULK_RESOLVE_CONFLICTS_REPORT_HEADER) ResultRow = namedtuple('Row', ['isrc', 'tuid', 'status', 'territories']) unique_tuids = set() rows = [] result = task_info.result account_id = result[field_const.ACCOUNT_ID] for isrc, isrc_results in result[field_const.SUCCESSFUL_ISRCS].items(): for tuid, territories in isrc_results.items(): if territories: status = bulk_tasks_const.CONFLICT_RESOLVED territories = ', '.join(territories) else: status = bulk_tasks_const.NO_CONFLICT_FOUND territories = bulk_tasks_const.NA rows.append(ResultRow(isrc, tuid, status, territories)) unique_tuids.add(tuid) for isrc, tuids in result[field_const.FAILED_ISRCS].items(): for tuid in tuids: status = bulk_tasks_const.UPDATE_STATUS_FAIL territories = bulk_tasks_const.NA rows.append(ResultRow(isrc, tuid, status, territories)) unique_tuids.add(tuid) tracks_data = ownership.get_tracks(list(unique_tuids)).message for row in rows: upc = '="{}"'.format( tracks_data.get(row.tuid, {}).get(field_const.UPC, '')) csv_writer.writerow([ account_id, upc, row.isrc, row.tuid, row.status, row.territories]) def _generate_lock_report_with_internal_conflict(csv_writer, task_info): """Generate lock report accounting for internal conflict case. Note that this function expect a bit different format of TaskStatus.result. { 'failed_isrcs': ['QA123'], 'reason': 'test lock reason', 'successful_isrcs': ['QA123'], 'territories': ['AD'], 'failed_territories': ['AF'] } Difference from old format is that new has 'failed_territories' key. Which is used in report for failed_isrcs. Args: csv_writer (csv.writer): instanse of csv.writer() task_info (TaskStatus): Instance of TaskStatus model """ csv_writer.writerow(bulk_tasks_const.BULK_LOCK_REPORT_HEADER) result_field = task_info.result reason = result_field[field_const.REASON] for isrc, payload in result_field['isrcs'].items(): failed_territories = payload.get('failed_territories', []) success_territories = payload.get('territories', []) if failed_territories: failed_territories = ','.join(failed_territories) csv_writer.writerow( [isrc, failed_territories, reason, 'Fail', 'Internal Conflict']) if success_territories: success_territories = ','.join(success_territories) csv_writer.writerow( [isrc, success_territories, reason, 'Success', 'N/A']) def _generate_initial_data_report(csv_data, import_report_type): """Generating initial data report Args: csv_data (TaskStatus): data for parsing and storing in csv import_report_type (str): Report type UPC | ISRC Returns: str: generated report content """ output = io.StringIO() csv_writer = csv.writer(output) result_field = csv_data.context if import_report_type == field_const.UPC: csv_writer.writerow([bulk_tasks_const.UPC_COLUMN]) else: csv_writer.writerow([bulk_tasks_const.ISRC_COLUMN]) for item in result_field.split(','): if import_report_type == field_const.UPC: report_row = ['="{}"'.format(item)] else: report_row = [item] csv_writer.writerow(report_row) return output.getvalue() def _create_file_name(task_info, import_report_type, suffix=''): """Generate the report filename Args: task_info (dict): task info from DB import_report_type (str): isrc | upc | None suffix (str): optional report name suffix, ex. '_initial' Returns: str: generated file name for report """ pattern = '{date}_{task_type}_{report_type}_{task_id}{suffix}.csv' data_dict = { 'date': str(task_info.create_datetime.date()), 'task_id': task_info.id, 'suffix': suffix } if import_report_type == field_const.ISRC: data_dict['report_type'] = 'ISRC' else: data_dict['report_type'] = 'UPC' if task_info.type == bulk_tasks_const.BULK_IMPORT: data_dict['task_type'] = 'Bulk_Update_UPCs' elif task_info.type == bulk_tasks_const.BULK_LOCK: data_dict['task_type'] = 'Bulk_Lock_Territories' data_dict['report_type'] = 'ISRC' elif task_info.type == bulk_tasks_const.BULK_RESOLVE_INTERNAL_CONFLICTS: data_dict['task_type'] = 'Bulk_Resolve_Conflicts' data_dict['report_type'] = 'ISRC' else: data_dict['task_type'] = 'Initial_Data' return pattern.format(**data_dict) def upload_report(file_name, generate_report): file_key = '{prefix}/{filename}'.format( prefix=config.REPORTS_FILE_KEY_PREFIX, filename=file_name) if s3.check_if_file_exists(file_key): return s3.get_file_url(file_key) report = generate_report() s3.upload_file_object(file_key, io.BytesIO(report.encode())) return s3.get_file_url(file_key) def _generate_report(task_info, correlation_id, import_report_type=None): """Generate csv report. Args: task_info (TaskStatus): task status object correlation_id (str): correlation ID for logs import_report_type (str): Report type UPC | ISRC Returns: str: generated report content """ output = io.StringIO() csv_writer = csv.writer(output) if task_info.type == bulk_tasks_const.BULK_IMPORT: if import_report_type == field_const.UPC: _generate_update_by_upc_report( csv_writer, task_info, correlation_id) else: _generate_update_by_isrc_report( csv_writer, task_info, correlation_id) if task_info.type == bulk_tasks_const.BULK_LOCK: _generate_lock_report_with_internal_conflict(csv_writer, task_info) if task_info.type == bulk_tasks_const.BULK_RESOLVE_INTERNAL_CONFLICTS: _generate_bulk_resolve_conflicts_report(csv_writer, task_info) return output.getvalue() def generate_report( task_id, correlation_id, import_report_type=None): """Generate csv report. Report of task by given correlationID Args: task_id (int): ID of the task correlation_id (str): correlation ID for logs import_report_type (str): Report type UPC | ISRC Returns: string : S3 URL """ resp = bulk_tasks.get_task(task_id=task_id) if not resp: return resp task_info = resp.message file_name = _create_file_name(task_info, import_report_type) generate_report = partial( _generate_report, task_info, correlation_id, import_report_type) return upload_report(file_name, generate_report) def generate_initial_report( task_id, import_report_type, correlation_id): """Generate initial data csv report. Args: task_id (int): ID of the task import_report_type (str): Report type UPC | ISRC correlation_id (str): correlation ID for logs Returns: string : S3 URL """ resp = bulk_tasks.get_task(task_id=task_id) if not resp: return resp task_info = resp.message if not task_info.context: return response.Response( status=404, message=error.TASK_WITHOUT_CONTEXT) file_name = _create_file_name( task_info, import_report_type, suffix='_initial') generate_report = partial( _generate_initial_data_report, task_info, import_report_type) return upload_report(file_name, generate_report)