""" Contains SR audit related sheets. """ from abc import ABC from copy import deepcopy from ...... import logger from ......constants import AuditTypes, DBColumns from ......constants import FlagResolutions as Resolutions from ......constants import Flags, FlagTableColumns from ......constants import RowReportColors as Colors from ......constants import RowReportCols as Cols from ......constants import RowReportColsHuman as ColsHuman from ......constants import RowReportTabs as Tabs from ......constants import RowReportTexts as Texts from ......constants import RowTypes, SnowFlakeColumns from ......typings import Row, RowIndex, Rows from .base import BaseSheet, Column logger = logger.new_logger(__name__) YES = Texts.YES NO = Texts.NO class BaseSheetSR(BaseSheet, ABC): filterAuditTypes = {AuditTypes.SR} class AudioNextSteps(BaseSheetSR): title = Tabs.AUDIO_NEXT_STEPS columns = [ *BaseSheet.columns, Column( key=Cols.ISSUE, label=ColsHuman.ISSUE, color_label_bkg=Colors.FFF5C343, ), Column( key=Cols.ACTION_REQUIRED, label=ColsHuman.ACTION_REQUIRED, color_label_bkg=Colors.FFF5C343, ), Column( key=[Cols.THIRD_PARTY_LABEL, SnowFlakeColumns.CONFLICTING_OWNERS], label=ColsHuman.THIRD_PARTY_LABEL, color_label_bkg=Colors.FFF5C343, ), Column( key=[Cols.LIST_CONFLICTING_TERRITORIES, Cols.CONFLICTING_TERRITORIES], label=ColsHuman.LIST_CONFLICTING_TERRITORIES, handler=lambda x: ", ".join(x) if x else x, color_label_bkg=Colors.FFF5C343, ), Column( key=Cols.LABEL_COMMENTS, label=ColsHuman.LABEL_COMMENTS, color_label_bkg=Colors.FFF5C343, is_dropdown=True, dropdown_default=False, # Empty cell value on default ), Column( key=Cols.LABEL_COMMENTS_TERRITORIES, label=ColsHuman.LABEL_COMMENTS_TERRITORIES, color_label_bkg=Colors.FFF5C343, color_highlight=Colors.FFF9E9B8, ), ] def __init__(self, rows: Rows): rows = self._preprocess_ownership_conflicts(rows) super().__init__(rows) # Dynamically set the highlight_if attribute for the # LABEL_COMMENTS_TERRITORIES column. Cannot set it in the # columns attribute because it would not be possible to encapsulate # the logic in a method. self._set_label_comments_territories_highlight_if() @staticmethod def _preprocess_ownership_conflicts(rows: Rows) -> Rows: """Preprocess rows to handle SR5 and SR8 ownership conflicts. As per Ricky Romano (YT Audits), the following is the requirement: matched_label_name and list_conflicting_territories are from SR5. They pertain to internal conflicts ("Please review and confirm exclusive rights in Column K"). While conflicting_owners and conflicting_territories are from SR8. They pertain to conflicts in YouTube ("Please review and respond to this conflict in Workstation's Conflict Manager") So because conflicts from SR5 and SR8 are different but share the same columns in the report (which are those of SR5), we need to separate them. The report should show at least one row, with either no conflicts or conflicts of either SR5 or SR8; or two rows if both SR5 and SR8 conflicts are present. """ _sr5_label, _sr8_label = "sr5", "sr8" processed_rows = [] conflict_cols = { _sr5_label: (Cols.LIST_CONFLICTING_TERRITORIES, Cols.THIRD_PARTY_LABEL), _sr8_label: ( Cols.CONFLICTING_TERRITORIES, SnowFlakeColumns.CONFLICTING_OWNERS, ), } for row in rows: # Do not use 'pop' here as it would mutate the original list sr5 = (row[col] for col in conflict_cols[_sr5_label]) sr8 = (row[col] for col in conflict_cols[_sr8_label]) both_sr5_and_sr8_have_conflicts = any(sr5) and any(sr8) if not both_sr5_and_sr8_have_conflicts: # Keep only the original row if no conflicts are present or # if only SR5 or SR8 conflicts are present (but not both) processed_rows.append(row) continue # SR5 and SR8 conflicts are present, so we need to split them into two rows, # each with one of the conflicts and the other conflict set to None. for cols in conflict_cols.values(): new_row = deepcopy(row) new_row.update(dict.fromkeys(cols, None)) processed_rows.append(new_row) return processed_rows def _set_label_comments_territories_highlight_if(self) -> None: col_idx, _ = self.find_col(Cols.LABEL_COMMENTS_TERRITORIES) self.columns[col_idx].highlight_if = self._highlight_label_comments_territories def _highlight_label_comments_territories(self, row_idx: RowIndex) -> str: _, col_letter = self.find_col(Cols.LABEL_COMMENTS) formula = ( f"OR(" f"${col_letter}{row_idx}" f'="{Texts.LABEL_COMMENTS_ASSERT_IN_SPECIFIC_TERRITORIES}",' f"${col_letter}{row_idx}" f'="{Texts.LABEL_COMMENTS_RELEASE_IN_SPECIFIC_TERRITORIES}"' f")" ) return formula def handle_rows(self) -> None: label_comments_options = ";".join( [ # Dropdown list Texts.LABEL_COMMENTS_ASSERT_IN_ALL_TERRITORIES, Texts.LABEL_COMMENTS_RELEASE_IN_ALL_TERRITORIES, Texts.LABEL_COMMENTS_ASSERT_IN_SPECIFIC_TERRITORIES, Texts.LABEL_COMMENTS_RELEASE_IN_SPECIFIC_TERRITORIES, ] ) filtered_rows = [] rows = self.rows while rows: row = rows.pop() # Checks whether the row has YT or MRR ownership conflicts and # filters it out if it doesn't. has_yt_ownership_conflict = self._has_yt_ownership_conflict(row) if not has_yt_ownership_conflict: has_mrr_ownership_conflict = self._has_mrr_ownership_conflict(row) if not has_mrr_ownership_conflict: continue row[Cols.ISSUE] = Texts.OWNERSHIP_CONFLICT row[Cols.ACTION_REQUIRED] = ( Texts.OWNERSHIP_CONFLICT_YT_TEXT if has_yt_ownership_conflict else Texts.OWNERSHIP_CONFLICT_MRR_TEXT ) row[Cols.LABEL_COMMENTS] = ( Texts.LABEL_COMMENTS_NONE if has_yt_ownership_conflict else label_comments_options ) filtered_rows.append(row) self.rows = self._deduplicate_rows_by_isrc( filtered_rows, skip_if_ownership_conflict=True ) _3_no_updated_reactivated_corrected = { Cols.IS_OWNERSHIP_CORRECT: NO, Cols.IS_ACTIVE_REFERENCE_FILE: NO, Cols.IS_UGC_MATCH_POLICY_CORRECT: NO, Cols.ACTIONS_TAKEN_BY_ORCHARD: Texts.UPDATED_REACTIVATED_CORRECTED, } _updated_ownership = { Cols.IS_OWNERSHIP_CORRECT: NO, Cols.ACTIONS_TAKEN_BY_ORCHARD: Texts.OWNERSHIP_UPDATED, } class AudioAudit(BaseSheetSR): title = Tabs.AUDIO_AUDIT columns = [ *BaseSheet.columns, Column( key=Cols.IS_DELIVERED_TO_YOUTUBE, label=ColsHuman.IS_DELIVERED_TO_YOUTUBE, ), Column( key=Cols.IS_OWNERSHIP_CORRECT, label=ColsHuman.IS_OWNERSHIP_CORRECT, ), Column( key=Cols.IS_ACTIVE_REFERENCE_FILE, label=ColsHuman.IS_ACTIVE_REFERENCE_FILE, ), Column( key=Cols.IS_UGC_MATCH_POLICY_CORRECT, label=ColsHuman.IS_UGC_MATCH_POLICY_CORRECT, ), Column( key=Cols.ACTIONS_TAKEN_BY_ORCHARD, label=ColsHuman.ACTIONS_TAKEN_BY_ORCHARD, ), Column( key=Cols.ISSUE, label=ColsHuman.ISSUE, ), Column( key=Cols.ADDITIONAL_NOTES, label=ColsHuman.ADDITIONAL_NOTES, ), Column( key=Cols.ACTIONS_REQUIRED, label=ColsHuman.ACTIONS_REQUIRED, color_label_bkg=Colors.FFF5C343, ), ] # These are flag columns that will be set to YES or NO based on the audit flags. flag_columns: list[Cols] = [ Cols.IS_DELIVERED_TO_YOUTUBE, Cols.IS_OWNERSHIP_CORRECT, Cols.IS_ACTIVE_REFERENCE_FILE, Cols.IS_UGC_MATCH_POLICY_CORRECT, ] # Try to reuse code as much as possible to avoid repetition. _all_flag_columns_no: dict[Cols, str] = {col: NO for col in flag_columns} _inactive_reference_file = { Cols.IS_ACTIVE_REFERENCE_FILE: NO, Cols.IS_UGC_MATCH_POLICY_CORRECT: NO, Cols.ISSUE: Texts.INACTIVE_REFERENCE_FILE, } _no_active_references_resolutions = { Resolutions.REACTIVATE_REFERENCE: { None: { Cols.IS_ACTIVE_REFERENCE_FILE: NO, Cols.IS_UGC_MATCH_POLICY_CORRECT: NO, Cols.ACTIONS_TAKEN_BY_ORCHARD: Texts.REACTIVATED_CORRECTED, }, }, Resolutions.INELIGIBLE_REFERENCE: { Resolutions.REFERENCE_INSUFFICIENT_LENGTH: { **_inactive_reference_file, Cols.ADDITIONAL_NOTES: Texts.REFERENCE_INSUFFICIENT_LENGTH, }, Resolutions.REFERENCE_EXCLUDED: { **_inactive_reference_file, Cols.ADDITIONAL_NOTES: Texts.REFERENCE_EXCLUDED, }, Resolutions.REFERENCE_DEACTIVATED_BY_OWNER: { **_inactive_reference_file, Cols.ADDITIONAL_NOTES: Texts.REFERENCE_DEACTIVATED_BY_OWNER, }, Resolutions.REFERENCE_OVERLAP: { **_inactive_reference_file, Cols.ADDITIONAL_NOTES: Texts.REFERENCE_OVERLAP, }, Resolutions.NO_FINGERPRINTING_RIGHTS: { **_inactive_reference_file, Cols.ADDITIONAL_NOTES: Texts.NO_FINGERPRINTING_RIGHTS, }, }, Resolutions.IGNORE: { Resolutions.REFERENCE_FILE_ACTIVE: {}, }, } _ineligible_for_delivery_resolutions = { Resolutions.PUBLIC_DOMAIN: { **_all_flag_columns_no, Cols.ISSUE: Texts.INELIGIBLE_FOR_DELIVERY, Cols.ADDITIONAL_NOTES: Texts.CONTAINS_PUBLIC_DOMAIN, }, Resolutions.NON_EXCLUSIVE: { **_all_flag_columns_no, Cols.ISSUE: Texts.INELIGIBLE_FOR_DELIVERY, Cols.ADDITIONAL_NOTES: Texts.CONTAINS_INELIGIBLE, }, Resolutions.TRACK_OFFER_TYPE_RESTRICTION: { **_all_flag_columns_no, Cols.ISSUE: Texts.INELIGIBLE_FOR_DELIVERY, Cols.ADDITIONAL_NOTES: Texts.FULL_ALBUM_DOWNLOAD_ONLY, }, Resolutions.CONFLICT_RESOLUTION: { **_all_flag_columns_no, Cols.ISSUE: Texts.OWNERSHIP_INCORRECT, Cols.ADDITIONAL_NOTES: Texts.OWNERSHIP_REMOVED_PER_LABEL_REQUEST, }, } _not_delivered = { Cols.IS_DELIVERED_TO_YOUTUBE: NO, Cols.IS_OWNERSHIP_CORRECT: NO, Cols.IS_ACTIVE_REFERENCE_FILE: NO, Cols.IS_UGC_MATCH_POLICY_CORRECT: NO, Cols.ACTIONS_TAKEN_BY_ORCHARD: Texts.TRACK_DELIVERED, } audit_columns = { Flags.NOT_DELIVERED: { Resolutions.DELIVER_TRACK: { None: _not_delivered, }, Resolutions.INELIGIBLE_FOR_DELIVERY: _ineligible_for_delivery_resolutions, }, Flags.NO_ASSET_ID: { Resolutions.DELIVER_TRACK: { None: _not_delivered, }, Resolutions.UPDATE_OWNERSHIP: { col: _3_no_updated_reactivated_corrected for col in (None, Resolutions.MRR_ISSUE, Resolutions.REDELIVER_TRACK) }, Resolutions.INELIGIBLE_FOR_DELIVERY: _ineligible_for_delivery_resolutions, Resolutions.IGNORE: { Resolutions.ASSET_OK: {}, }, }, Flags.TERRITORIES_MISSING: { Resolutions.UPDATE_OWNERSHIP: { col: _updated_ownership for col in (None, Resolutions.BACKFILL_MEAT) }, Resolutions.IGNORE: { Resolutions.TERRITORY_RIGHTS_CORRECT: {}, Resolutions.OWNERSHIP_CORRECT: {}, }, }, Flags.OWNERSHIP_INCOMPLETE: { Resolutions.UPDATE_OWNERSHIP: { None: _updated_ownership, }, Resolutions.IGNORE: { Resolutions.TERRITORY_RIGHTS_CORRECT: {}, }, }, Flags.CAN_REACTIVATE: { Resolutions.REACTIVATE_REFERENCE: { None: { Cols.IS_ACTIVE_REFERENCE_FILE: NO, Cols.IS_UGC_MATCH_POLICY_CORRECT: NO, Cols.ACTIONS_TAKEN_BY_ORCHARD: Texts.REACTIVATED_CORRECTED, } }, Resolutions.IGNORE: { Resolutions.REFERENCE_FILE_ACTIVE: {}, }, }, Flags.NO_ACTIVE_REFERENCES: _no_active_references_resolutions, Flags.NO_ACTIVE_REFERENCES_REASONS_ORCHARD: _no_active_references_resolutions, Flags.NO_ACTIVE_REFERENCES_REASONS_THIRD_PARTY: _no_active_references_resolutions, Flags.BAD_MATCH_POLICY: { Resolutions.CORRECT_MATCH_POLICY: { None: { Cols.IS_UGC_MATCH_POLICY_CORRECT: NO, Cols.ACTIONS_TAKEN_BY_ORCHARD: Texts.MATCH_POLICY_CORRECTED, } }, Resolutions.IGNORE: { Resolutions.MATCH_POLICY_CORRECT: {}, }, }, } def handle_rows(self) -> None: filtered_rows = [] rows = self.rows while rows: row = rows.pop() row_handled = self._handle_row(row) filtered_rows.append(row_handled) self.rows = self._deduplicate_rows_by_isrc(filtered_rows) def _handle_row(self, row: Row, **kwargs) -> Row: row_audit_id = row[DBColumns.ID] audit_flags = self.flags.get(row_audit_id, {}) if self._has_ownership_conflict(row): self._handle_row_with_ownership_conflict(row, audit_flags) # Set all flag columns to YES by default self._set_all_flag_columns_default(row) if audit_flags: self._handle_flag_columns(row, audit_flags) return row @staticmethod def _handle_row_with_ownership_conflict( row: Row, audit_flags: dict[int, list[dict[str, str]]], ) -> None: """Handle a row with ownership conflict.""" row_idx = row[DBColumns.ROW_IDX] row_flags = audit_flags.get(row_idx, []) row_resolutions = {flag[FlagTableColumns.RESOLUTION] for flag in row_flags} add_inactive_ref_file_note = bool( row_resolutions & { # Take only resolutions into account, to apply to all sub-resolutions Resolutions.INELIGIBLE_REFERENCE, Resolutions.NO_ACTIVE_REFERENCES, } ) row[Cols.ISSUE] = ( Texts.OWNERSHIP_CONFLICT_INACTIVE_REFERENCE_FILE if add_inactive_ref_file_note else Texts.OWNERSHIP_CONFLICT ) row[Cols.ACTIONS_REQUIRED] = Texts.REVIEW_AUDIO_NEXT_STEPS def _handle_flag_columns(self, row: Row, audit_flags: dict) -> None: """Handle the flag columns based on the audit flags.""" row_flags = audit_flags.get(row[DBColumns.ROW_IDX], []) for flag in row_flags: flag_type = flag[FlagTableColumns.FLAG] resolution = flag[FlagTableColumns.RESOLUTION] if not resolution: continue # Skip flags not yet resolved resolution_subtype = flag[FlagTableColumns.RESOLUTION_SUBTYPE] try: flag_type_obj = self.audit_columns[flag_type] # Handle possible Python bug affecting # "_no_active_references_resolutions" # which despite being a dict is wrapped in a # tuple, so unwrap it! if isinstance(flag_type_obj, tuple) and len(flag_type_obj) == 1: flag_type_obj = flag_type_obj[0] resolution_obj = flag_type_obj[resolution] flag_column_values = resolution_obj[resolution_subtype] except KeyError as ex: raise KeyError( f"Flag type '{flag_type}', resolution '{resolution}', " f"and resolution subtype '{resolution_subtype}' not found in " f"the audit_columns attribute. Implementation required!" ) from ex self._apply_flag_column_values(row, flag_column_values) def _apply_flag_column_values(self, row: Row, flag_column_values: dict) -> None: for col, value in flag_column_values.items(): if col in self.flag_columns and value is not NO: # NEVER override a NO! If a NO is set, it should NOT be changed. # This could happen if a flag evaluated later sets intends to # set a YES, but a previous flag set a NO. continue row[col] = value def _set_all_flag_columns_default(self, row: Row) -> None: row.update({column: Texts.YES for column in self.flag_columns}) class AudioDeletedCarvedOut(BaseSheetSR): filterRowTypes = {RowTypes.CARVED_OUT, RowTypes.NOT_ACTIVE_RELEASE} title = Tabs.AUDIO_DELETED_CARVED_OUT columns = [ *BaseSheet.columns, Column( key=Cols.STATUS, label=ColsHuman.STATUS, ), ] def handle_rows(self) -> None: filtered_rows = [] rows = self.rows while rows: row = rows.pop() is_carved_out = row[DBColumns.ROW_TYPE] == RowTypes.CARVED_OUT status = ( Texts.CARVED_OUT_RELEASE if is_carved_out else Texts.DELETED_RELEASE ) row[Cols.STATUS] = status filtered_rows.append(row) self.rows = self._deduplicate_rows_by_isrc(filtered_rows)