""" Sheets for the report spreadsheet. Heavy use of subclassing is made to allow for easy extension of the report and maximum reusability of the code. The base class `_BaseSheet` defines the common logic for all sheets, such as filtering rows based on audit types and row types, handling rows upon instantiation, etc. New sheets can be added by creating a new class that extends `_BaseSheet` and implements all abstract methods. Documentation on the implemented logic of this module: - Linking SR & AT Audit Data To SR & AT Audit Report https://docs.google.com/document/d/1pMceSVuoyUFU-MkDnbCGqG1FXaoSPxvnOEICHwM8K5c/ - Breakdown Of Audit Flags (Audit Flags - Audit Resolution Options) https://docs.google.com/document/d/1VXzV6cSIbgOpgdQNP826A7rYm1rcSWVnuVFVeYHODYQ/ """ import difflib from abc import ABC, abstractmethod from collections import defaultdict from copy import deepcopy from dataclasses import dataclass from typing import Callable, Iterable from ...... import logger from ......constants import AuditTypes, DBColumns, 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 ( AuditID, AuditTypeShort, ColorHex8, ColumnIndex, ColumnName, Row, RowIndex, Rows, XLSXFormula, ) logger = logger.new_logger(__name__) YES = Texts.YES NO = Texts.NO @dataclass(slots=True) class Column: """Column in a report spreadsheet sheet. Attributes: key: Key of the column as it will be added to the row object (e.g. 'label_name'). If an iterable is provided, the keys will be evaluated in order until a truthy value is found, and the first truthy value will be used as the cell value. label: Label of the column as it will appear in the header of the sheet (i.e. the first row, as visible for humans, e.g. 'Label Name') is_dropdown: Whether the column is a dropdown column. If True, the column will be set as dropdown column in the Excel sheet: the cells which have a ';' separator will be set as dropdown cells, otherwise they will be set as text cells. dropdown_default: Default value for the dropdown column. If True, the default value will be set as the first option in the dropdown list. If False, the default value will be empty. This attribute is only relevant if 'is_dropdown' is True. is_hyperlink: Whether the column is a hyperlink column. If True, the column will be scanned for URLs and the cells will be set as hyperlinks in the Excel sheet. color_label_bkg: Background color of the label cell. Must be a hex color with 8 characters (e.g. 'FF52B5C6'). color_highlight: Highlight color of the cell. Must be a hex color with 8 characters (e.g. 'FFFF0000'). handler: Handler for the column values. If provided, the handler will be called with each cell value of the column, and the returned value will be used in the cell. If not provided, the cell value will be used as is. E.g. lambda x: x.upper() would convert all cell values to uppercase. highlight_if: Highlight condition for the cell. If provided, the cell will be highlighted if the condition is met. The condition is a callable that takes the row index as argument and returns a formula to evaluate. E.g. lambda x: f"={x + 1} > 5" would highlight the cell if the row index is greater than 4. The formula must be a valid Excel formula. """ key: str | Iterable[str] label: str is_dropdown: bool = False dropdown_default: bool = False is_hyperlink: bool = False color_label_bkg: ColorHex8 = Colors.FF52B5C6 color_highlight: ColorHex8 = Colors.FFFF0000 handler: Callable = None highlight_if: Callable[[RowIndex], XLSXFormula] = None class BaseSheet(ABC): """Base class for a report spreadsheet sheet.""" # Filter the rows based on the audit types. If None, no filtering is done. filterAuditTypes: set[AuditTypeShort] | None = None # On default, only analyzed rows are included. If None, no filtering is done. filterRowTypes: set[RowTypes] | None = {RowTypes.ANALYZED} # Those are the columns common to all sheets. They can be reused in subclasses, # and additional columns can be added to the list, if needed. columns: list[Column] = [ Column(key=key, label=label) for key, label in [ (Cols.LABEL_NAME, ColsHuman.LABEL_NAME), (Cols.UPC, ColsHuman.UPC), (Cols.ISRC, ColsHuman.ISRC), (Cols.ARTIST, ColsHuman.ARTIST), (Cols.RELEASE_NAME, ColsHuman.RELEASE_NAME), (Cols.TRACK_NAME, ColsHuman.TRACK_NAME), ] ] def __init__(self, rows: Rows, *, flags: Rows = None): """Initialize the sheet with rows. Args: rows: List of rows to include in the sheet. It's a list of dictionaries, where each dictionary represents a row, with keys being column names. The rows are copied to avoid mutation of the original data. flags: Flags to include in the sheet. """ logger.info("Initializing sheet '{}' with {} rows", self.title, len(rows)) if self.filterAuditTypes: rows = [row for row in rows if row[DBColumns.TYPE] in self.filterAuditTypes] if self.filterRowTypes: rows = [ row for row in rows if row[DBColumns.ROW_TYPE] in self.filterRowTypes ] # No mutation of the original data self.rows = deepcopy(rows) if not self.rows: # No rows to process for this audit type. Short-circuit here to prevent # row handling from kicking in, as stuff like missing columns because # of empty rows can lead to crashes. return if self.filterAuditTypes and flags: flags = [ flag for flag in flags if flag[DBColumns.TYPE] in self.filterAuditTypes ] self.flags = self._group_flags(deepcopy(flags) if flags else []) self.handle_rows() @abstractmethod def title(self) -> Tabs: """Title of the sheet, as it will appear in the tab of the Excel file.""" def handle_rows(self) -> None: """Handle rows for the sheet. This method acts as a hook which can be overridden in subclasses, to automatically run operations on the rows upon instantiation of the sheet. """ def find_col(self, key: ColumnName) -> tuple[ColumnIndex, str]: """Find the index and letter of a column as defined in the 'columns' attribute of the sheet, based on the key. Args: key: Key of the column to find. Returns: Tuple with the 0-based index and letter of the column. Raises: NotImplementedError: If the column index is above 26 (i.e. above Z). StopIteration: If the column is not found. """ col_idx = next( idx for idx, column in enumerate(self.columns) if column.key == key ) if col_idx >= 26: raise NotImplementedError("Columns above Z are not implemented!") col_letter = chr(65 + col_idx) return col_idx, col_letter def _has_ownership_conflict(self, row: dict) -> bool: return self._has_yt_ownership_conflict(row) or self._has_mrr_ownership_conflict( row ) @staticmethod def _has_yt_ownership_conflict(row: dict) -> bool: """Check if a row has a YouTube ownership conflict.""" return bool( row[Cols.CONFLICTING_TERRITORIES] or row[SnowFlakeColumns.CONFLICTING_OWNERS] ) @staticmethod def _has_mrr_ownership_conflict(row: dict) -> bool: """Check if a row has an MRR ownership conflict.""" return bool( row[SnowFlakeColumns.MATCHED_LABEL_NAME] or row[SnowFlakeColumns.LIST_CONFLICTING_TERRITORIES] ) def _deduplicate_rows_by_isrc( self, rows: Rows, skip_if_ownership_conflict: bool = False ) -> Rows: """(FOR AUDIO ROWS ONLY) Deduplicate rows based on ISRC. Args: rows: Rows to deduplicate. skip_if_ownership_conflict: Skip deduplication of a group of rows sharing a same ISRC if there's at least one row in the group with a YT ownership conflict AND at least one row with an MRR ownership conflict. Returns: Deduplicated rows. """ rows_grouped_by_isrc = defaultdict(list) for row in rows: if row[DBColumns.TYPE] != AuditTypes.SR: raise ValueError( "Deduplication is only applicable to SR audits, but " f"a row with type '{row[DBColumns.TYPE]}' was found." ) rows_grouped_by_isrc[row[Cols.ISRC]].append(row) deduplicated_rows = [] for row_group in rows_grouped_by_isrc.values(): if len(row_group) == 1: deduplicated_rows.append(row_group[0]) continue if skip_if_ownership_conflict: any_has_yt_ownership_conflict = any( self._has_yt_ownership_conflict(row) for row in row_group ) any_has_mrr_ownership_conflict = any( self._has_mrr_ownership_conflict(row) for row in row_group ) if any_has_yt_ownership_conflict and any_has_mrr_ownership_conflict: deduplicated_rows.extend(row_group) continue selected_row = self._deduplicate_rows_prefer_single_before_album(row_group) deduplicated_rows.append(selected_row) logger.debug( "Deduplicated rows by ISRC: {} -> {} (removed {} duplicates)", len(rows), len(deduplicated_rows), len(rows) - len(deduplicated_rows), ) return deduplicated_rows @staticmethod def _deduplicate_rows_prefer_single_before_album(rows: Rows) -> Row: """(FOR AUDIO ROWS ONLY) Deduplicate rows based on release name and track name similarity. The row with the highest string proximity is selected and returned, with the understanding that they're likely to be a single release instead of an album release. """ results = [] for row in rows: release_name = row[Cols.RELEASE_NAME] track_name = row[Cols.TRACK_NAME] string_proximity = difflib.SequenceMatcher( None, release_name.lower(), track_name.lower() ).ratio() results.append((string_proximity, row)) return max(results, key=lambda x: x[0])[1] @staticmethod def _group_flags(flags: Rows) -> dict[AuditID, dict[RowIndex, Rows]]: """Group flags by audit ID and then by row index.""" grouped_flags = defaultdict(lambda: defaultdict(list)) for flag in flags: grouped_flags[flag[FlagTableColumns.AUDIT]][ flag[FlagTableColumns.ROW_IDX] ].append(flag) return {audit_id: dict(rows) for audit_id, rows in grouped_flags.items()}