""" Dataclasses for storing data to be inserted into the PPTX template. """ from collections import Counter from dataclasses import dataclass from datetime import date from functools import cached_property from io import BytesIO from itertools import chain from ....constants import Misc from ....typings import ISRC from ....utils import images from ....utils.strings import shorten_number, var_precision from .constants import Actions UniqueISRCs = frozenset[ISRC] NonUniqueISRCs = tuple[ISRC, ...] @dataclass class _PlaceholderDataSRATCalculations: """Dataclass for storing data to be inserted into the PPTX template placeholders. Attributes should be the lower case names of the placeholders in the PPTX template, without the delimiters. Calculated properties can be defined as methods, and they will be accessible as attributes of the dataclass instances. Info about calculation logic: https://docs.google.com/document/d/1nv3Q4iVHqCF5ZuRR5fralb8Sm8IlrVm0GBm5C7xfSqU/edit?usp=sharing """ label_name: str sr_isrcs_unique_total_count: int sr_isrcs_unique_flagged: UniqueISRCs # Before corrections (i.e. initially flagged) sr_isrcs_unique_corrected: UniqueISRCs # After corrections (i.e. number of flagged ISRCs that were corrected) sr_isrcs_unique_flagged_yt_ownership_conflict: UniqueISRCs sr_isrcs_unique_flagged_mrr_ownership_conflict: UniqueISRCs sr_isrcs_unique_flagged_ineligible_track: UniqueISRCs sr_isrcs_unique_corrected_yt_ownership_updated: UniqueISRCs sr_isrcs_unique_corrected_reference_reactivated: UniqueISRCs sr_isrcs_unique_corrected_match_policy_updated: UniqueISRCs srugc_avg_daily_views: int srugc_match_count: int srugc_tracks_with_ugc_match_pct: float | None # None means not applicable sr_added_territory_rights_count: int # Number of flag resolutions involving # added territory rights # After an audit is completed, the CMS Conflict, Registry Conflict, and Ineligible # Tracks total unique ISRCs and their percentage of the entire catalog should # equal the After Not Monetizing (SR) track count and percentage. sr_actionable_conflict_count: int # Conflicting owner count in SR8 sr_actionable_attached_conflict_count: int # Locked rows count in SR4 at_isrcs_total_count: int at_isrcs_flagged: NonUniqueISRCs # Before corrections (i.e. initially flagged) at_isrcs_corrected: NonUniqueISRCs # After corrections (i.e. number of flagged ISRCs that were corrected) at_isrcs_flagged_yt_incorrect_channel: NonUniqueISRCs at_isrcs_flagged_yt_ownership_conflict: NonUniqueISRCs at_isrcs_corrected_incorrect_channel_fixed: NonUniqueISRCs at_isrcs_corrected_yt_ownership_updated: NonUniqueISRCs at_redelivered_count: int # Number of flag resolutions involving track redelivery at_remapped_count: int # Number of flag resolutions involving track remapping def __post_init__(self): self.label_name = self.label_name.upper() self.date = date.today().strftime("%B %Y").upper() # e.g. "JANUARY 2022" self.srugc_tracks_with_ugc_match_pct = ( # Convert to string representation Misc.NA if self.srugc_tracks_with_ugc_match_pct is None else _pct(self.srugc_tracks_with_ugc_match_pct * 100, True) ) @cached_property def sr_before_monetizing_track_count(self) -> int: """The number of Sound Recordings that were monetizing on YouTube before the audit. """ return self.sr_isrcs_unique_total_count - len(self.sr_isrcs_unique_flagged) @cached_property def sr_before_monetizing_track_pct(self) -> float: """The number of Sound Recordings that were monetizing on YouTube before the audit, as a percentage of the total number of Sound Recordings. """ return self._sr_pct_of_total(self.sr_before_monetizing_track_count) @cached_property def sr_before_not_monetizing_track_count(self) -> int: """The number of Sound Recordings that were not monetizing on YouTube before the audit, i.e. the number of flagged Sound Recordings. """ return len(self.sr_isrcs_unique_flagged) @cached_property def sr_before_not_monetizing_track_pct(self) -> float: """The number of Sound Recordings that were not monetizing on YouTube before the audit, as a percentage of the total number of Sound Recordings. """ return self._sr_pct_of_total(self.sr_before_not_monetizing_track_count) @cached_property def sr_after_monetizing_track_count(self) -> int: """The number of Sound Recordings that are monetizing on YouTube after the audit. """ return self.sr_before_monetizing_track_count + len( self.sr_isrcs_unique_corrected ) @cached_property def sr_after_monetizing_track_pct(self) -> float: """The number of Sound Recordings that are monetizing on YouTube after the audit, as a percentage of the total number of tracks. """ return self._sr_pct_of_total(self.sr_after_monetizing_track_count) @cached_property def sr_after_not_monetizing_track_count(self) -> int: """The number of Sound Recordings that are not monetizing on YouTube after the audit. """ return self.sr_isrcs_unique_total_count - self.sr_after_monetizing_track_count @cached_property def sr_after_not_monetizing_track_pct(self) -> float: """The number of Sound Recordings that are not monetizing on YouTube after the audit, as a percentage of the total number of Sound Recordings. """ return self._sr_pct_of_total(self.sr_after_not_monetizing_track_count) @cached_property def sr_monetization_variation_pct(self) -> float: """The percentage variation in monetization after the audit.""" return self.sr_after_monetizing_track_pct - self.sr_before_monetizing_track_pct @cached_property def sr_yt_ownership_conflict_count(self) -> int: """The number of Sound Recordings with a YouTube ownership conflict.""" return len(self.sr_isrcs_unique_flagged_yt_ownership_conflict) @cached_property def sr_yt_ownership_conflict_pct(self) -> float: """The number of Sound Recordings with a YouTube ownership conflict, as a percentage of the total number of Sound Recordings. """ return self._sr_pct_of_total(self.sr_yt_ownership_conflict_count) @cached_property def sr_mrr_ownership_conflict_count(self) -> int: """The number of Sound Recordings with a MRR ownership conflict.""" return len(self.sr_isrcs_unique_flagged_mrr_ownership_conflict) @cached_property def sr_mrr_ownership_conflict_pct(self) -> float: """The number of Sound Recordings with a MRR ownership conflict, as a percentage of the total number of tracks. """ return self._sr_pct_of_total(self.sr_mrr_ownership_conflict_count) @cached_property def sr_ownership_conflict_total_count(self) -> int: """The total number of Sound Recordings with an ownership conflict.""" return ( self.sr_yt_ownership_conflict_count + self.sr_mrr_ownership_conflict_count ) @cached_property def sr_ownership_conflict_total_pct(self) -> float: """The total number of Sound Recordings with an ownership conflict.""" return self._sr_pct_of_total(self.sr_ownership_conflict_total_count) @cached_property def sr_ineligible_track_count(self) -> int: """The number of Sound Recordings that are ineligible for monetization on YouTube. """ return len(self.sr_isrcs_unique_flagged_ineligible_track) @cached_property def sr_ineligible_track_pct(self) -> float: """The number of Sound Recordings that are ineligible for monetization on YouTube, as a percentage of the total number of Sound Recordings. """ return self._sr_pct_of_total(self.sr_ineligible_track_count) @property def sr_agg_added_territory_rights_count_text(self) -> str: """The number of Sound Recordings with added territory rights.""" added_count = self.sr_added_territory_rights_count return f"{added_count} sound recording" f"{'s' if added_count != 1 else ''}" @cached_property def sr_agg_updated_ownership_count(self) -> int: """The number of Sound Recordings with updated YouTube ownership.""" return self._sr_aggregates[Actions.YT_OWNERSHIP_UPDATED] @cached_property def sr_agg_updated_ownership_pct(self) -> float: """The number of Sound Recordings with updated YouTube ownership, as a percentage of the total number of Sound Recordings. """ return self._sr_pct_of_total(self.sr_agg_updated_ownership_count) @cached_property def sr_agg_reactivated_reference_count(self) -> int: """The number of Sound Recordings with reactivated reference.""" return self._sr_aggregates[Actions.REFERENCE_REACTIVATED] @property def sr_agg_reactivated_reference_count_text(self) -> str: """The number of Sound Recordings with reactivated reference.""" reactivated_count = self.sr_agg_reactivated_reference_count return ( f"{reactivated_count} sound recording" f"{'s' if reactivated_count != 1 else ''}" ) @cached_property def sr_agg_reactivated_reference_pct(self) -> float: """The number of Sound Recordings with reactivated reference, as a percentage of the total number of Sound Recordings. """ return self._sr_pct_of_total(self.sr_agg_reactivated_reference_count) @cached_property def sr_agg_updated_match_policy_count(self) -> int: """The number of Sound Recordings with updated match policy.""" return self._sr_aggregates[Actions.MATCH_POLICY_UPDATED] @property def sr_agg_updated_match_policy_count_text(self) -> str: """The number of Sound Recordings with updated match policy.""" updated_count = self.sr_agg_updated_match_policy_count return f"{updated_count} sound recording" f"{'s' if updated_count != 1 else ''}" @cached_property def sr_agg_updated_match_policy_pct(self) -> float: """The number of Sound Recordings with updated match policy, as a percentage of the total number of Sound Recordings. """ return self._sr_pct_of_total(self.sr_agg_updated_match_policy_count) @cached_property def _sr_aggregates(self) -> dict[str, int]: """Perform Sound Recording aggregates and bucket them into a number of categories, using different priority levels. """ # Collect all unique ISRCs that were flagged for any ownership conflict type isrcs_with_ownership_conflict = ( self.sr_isrcs_unique_flagged_yt_ownership_conflict ) | self.sr_isrcs_unique_flagged_mrr_ownership_conflict corrected_yt_ownership_updated_count = ( isrcs_with_ownership_conflict & self.sr_isrcs_unique_corrected_yt_ownership_updated ) isrcs_with_ownership_conflict = ( isrcs_with_ownership_conflict - corrected_yt_ownership_updated_count ) corrected_reference_reactivated_count = ( isrcs_with_ownership_conflict & self.sr_isrcs_unique_corrected_reference_reactivated ) isrcs_with_ownership_conflict = ( isrcs_with_ownership_conflict - corrected_reference_reactivated_count ) corrected_match_policy_updated_count = ( isrcs_with_ownership_conflict & self.sr_isrcs_unique_corrected_match_policy_updated ) isrcs_with_ownership_conflict = ( isrcs_with_ownership_conflict - corrected_match_policy_updated_count ) assert not isrcs_with_ownership_conflict, ( f"All ISRCs with ownership conflicts should have been bucketed, but the " f"following were not: {', '.join(isrcs_with_ownership_conflict)}" ) return { Actions.YT_OWNERSHIP_UPDATED: len(corrected_yt_ownership_updated_count), Actions.REFERENCE_REACTIVATED: len(corrected_reference_reactivated_count), Actions.MATCH_POLICY_UPDATED: len(corrected_match_policy_updated_count), } @cached_property def at_before_monetizing_track_count(self) -> int: """The number of Art Tracks that were monetizing on YouTube before the audit.""" return self.at_isrcs_total_count - len(self.at_isrcs_flagged) @cached_property def at_before_monetizing_track_pct(self) -> float: """The number of Art Tracks that were monetizing on YouTube before the audit, as a percentage of the total number of Art Tracks. """ return self._at_pct_of_total(self.at_before_monetizing_track_count) @cached_property def at_before_not_monetizing_track_count(self) -> int: """The number of Art Tracks that were not monetizing on YouTube before the audit. """ return len(self.at_isrcs_flagged) @cached_property def at_before_not_monetizing_track_pct(self) -> float: """The number of Art Tracks that were not monetizing on YouTube before the audit, as a percentage of the total number of Art Tracks. """ return self._at_pct_of_total(self.at_before_not_monetizing_track_count) @cached_property def at_after_monetizing_track_count(self) -> int: """The number of Art Tracks that are monetizing on YouTube after the audit.""" return self.at_before_monetizing_track_count + len(self.at_isrcs_corrected) @cached_property def at_after_monetizing_track_pct(self) -> float: """The number of Art Tracks that are monetizing on YouTube after the audit, as a percentage of the total number of Art Tracks. """ return self._at_pct_of_total(self.at_after_monetizing_track_count) @cached_property def at_after_not_monetizing_track_count(self) -> int: """The number of Art Tracks that are not monetizing on YouTube after the audit.""" return self.at_isrcs_total_count - self.at_after_monetizing_track_count @cached_property def at_after_not_monetizing_track_pct(self) -> float: """The number of Art Tracks that are not monetizing on YouTube after the audit, as a percentage of the total number of Art Tracks. """ return self._at_pct_of_total(self.at_after_not_monetizing_track_count) @cached_property def at_monetization_variation_pct(self) -> float: """The percentage variation in monetization after the audit.""" return self.at_after_monetizing_track_pct - self.at_before_monetizing_track_pct @cached_property def at_yt_ownership_conflict_count(self) -> int: """The number of Art Tracks with a YouTube ownership conflict.""" return len(self.at_isrcs_flagged_yt_ownership_conflict) @cached_property def at_yt_ownership_conflict_pct(self) -> float: """The number of Art Tracks with a YouTube ownership conflict, as a percentage of the total number of Art Tracks. """ return self._at_pct_of_total(self.at_yt_ownership_conflict_count) @property def at_agg_redelivered_count_text(self) -> str: """The number of Art Tracks which have been redelivered.""" redelivered_count = self.at_redelivered_count return ( f"{redelivered_count} Art Track" f"{'s' if redelivered_count != 1 else ''}" ) @property def at_agg_remapped_count_text(self) -> str: """The number of Art Tracks which have been remapped.""" remapped_count = self.at_remapped_count return f"{remapped_count} Art Track" f"{'s' if remapped_count != 1 else ''}" @cached_property def _at_aggregates(self) -> dict[str, int]: """Perform Art Track aggregates and bucket them into a number of categories, using different priority levels. """ # Collect all ISRCs that were either mapped to the correct topic # channel, had their ownership updated, or both. Do NOT make them unique # yet, as some calculations are based on non-unique counts. In order to # save memory and be able to work with set-like operations but without # losing the original counts, we use a Counter. relevant_isrcs = Counter( self.at_isrcs_corrected_incorrect_channel_fixed + self.at_isrcs_corrected_yt_ownership_updated ) def tuple_func(_relevant, _seq): segment = _relevant.keys() & _seq result = tuple( chain.from_iterable( (isrc for _ in range(count)) for isrc, count in _relevant.items() if isrc in segment ) ) for isrc in segment: del _relevant[isrc] return result corrected_incorrect_channel_fixed = tuple_func( relevant_isrcs, self.at_isrcs_corrected_incorrect_channel_fixed ) corrected_yt_ownership_updated = tuple_func( relevant_isrcs, self.at_isrcs_corrected_yt_ownership_updated ) assert not relevant_isrcs, "All ISRCs should have been bucketed." return { Actions.INCORRECT_CHANNEL_FIXED: len(corrected_incorrect_channel_fixed), Actions.YT_OWNERSHIP_UPDATED: len(corrected_yt_ownership_updated), } def _pct_of_total(self, count: int, total_attr: str, _round: int = 1) -> float: """Calculate the percentage of the total number of unique ISRCs (either Sound Recordings or Art Tracks) relative to the given count. Parameters: - count: The number to calculate the percentage for. - total_attr: The attribute name that holds the total count. - _round: The number of decimal places to round the result to. """ total_count = getattr(self, total_attr, 0) return round((count / total_count) * 100, _round) if total_count else 0 def _sr_pct_of_total(self, count: int, _round: int = 1) -> float: """Calculate the percentage of the total number of unique Sound Recordings ISRCs relative to the given count. """ return round((count / self.sr_isrcs_unique_total_count) * 100, _round) def _at_pct_of_total(self, count: int, _round: int = 1) -> float: """Calculate the percentage of the total number of unique Art Tracks ISRCs relative to the given count. """ return round((count / self.at_isrcs_total_count) * 100, _round) @dataclass class PlaceholderDataSRAT(_PlaceholderDataSRATCalculations): @property def sr_bar_chart_overview(self) -> str: """Text for the bar chart overview slide.""" monetization_increase_pct = self.sr_monetization_variation_pct assert ( monetization_increase_pct >= 0 ), "Monetization percentage should not decrease!" text = [] if monetization_increase_pct > 0: text.append( f"At the conclusion of this audit, we’ve increased the " f"monetization potential of your audio catalog by " f"{_pct(monetization_increase_pct)}, from " f"{_pct(self.sr_before_monetizing_track_pct)} to " f"{_pct(self.sr_after_monetizing_track_pct)}. This was " f"accomplished by correcting issues in a few key areas: " f"reactivating references to claim User-Generated Content " f"(UGC), updating ownership, and correcting UGC policies. " f"As a bonus, we also audited your audio catalog made " f"available for streaming on YouTube via Art Tracks; " f"check out slide 10 for more information." ) else: text.append( f"At the conclusion of this audit, your audio catalog monetization " f"remained at {_pct(self.sr_after_monetizing_track_pct)}." ) if self.sr_ownership_conflict_total_count > 0: text.append( "There are more opportunities to increase your revenue-earning " "potential and reach on YouTube, but we need your input. Refer to " "slide 6 on how you can help us resolve any outstanding issues." ) return "\n\n".join(text) @property def sr_audit_process_overview_ownership(self) -> str: """Text for the audit process overview slide [Ownership].""" if self.sr_agg_updated_ownership_count > 0: return ( f"{_pct(self.sr_agg_updated_ownership_pct)}% of your audio catalog was " f"updated to reflect your rights." ) return "All territory rights were reflected correctly for your audio catalog." @property def sr_audit_process_overview_references(self) -> str: """Text for the audit process overview slide [References].""" if self.sr_agg_reactivated_reference_count > 0: return ( f"We reactivated references on " f"{self.sr_agg_reactivated_reference_count}% " f"of your audio catalog." ) return "All reference files were enabled for your audio catalog." @property def sr_audit_process_overview_match_policy(self) -> str: """Text for the audit process overview slide [Match Policy].""" if self.sr_agg_updated_match_policy_count > 0: return ( f"We updated the UGC match policy for " f"{self.sr_agg_updated_match_policy_count}% of your audio catalog." ) return "All match policies were set up correctly." @property def sr_audit_results_after_monetizing(self) -> str: """Text for the audit process results slide [After Monetizing].""" return ( f"{_pct(self.sr_after_monetizing_track_pct)}% of your audio catalog " f"was enabled for monetization after the audit." ) @property def sr_mt(self) -> str: """MONETIZING (no % symbol).""" return _pct(self.sr_before_monetizing_track_pct, False) @property def sr_ou(self) -> str: """OWNERSHIP UPDATED (no % symbol).""" return _pct(self.sr_agg_updated_ownership_pct, False) @property def sr_rr(self) -> str: """REFERENCES REACTIVATED (no % symbol).""" return _pct(self.sr_agg_reactivated_reference_pct, False) @property def sr_mp(self) -> str: """MATCH POLICIES CORRECTED (no % symbol).""" return _pct(self.sr_agg_updated_match_policy_pct, False) @property def sr_nm(self) -> str: """NOT MONETIZING (no % symbol).""" return _pct(self.sr_after_not_monetizing_track_pct, False) @property def sr_audit_next_steps_intro(self) -> str: """Text for the audit process next steps slide.""" return ( f"At the conclusion of the audit, we’ve identified that " f"{_pct(self.sr_ownership_conflict_total_pct)} of your audio " f"fingerprints are in conflict, preventing you from monetizing." ) @property def sr_audit_next_steps(self) -> str: """Text for the audit process next steps slide.""" text = [] if self.sr_yt_ownership_conflict_count > 0: text.append( f"There are {self.sr_yt_ownership_conflict_count} conflicts ready " f"for your review in the Conflict Manager section of Workstation." ) if self.sr_mrr_ownership_conflict_count > 0: an_additional = ( " an additional" if (self.sr_yt_ownership_conflict_count > 0) else "" ) text.append( f"There are{an_additional} " f"{self.sr_mrr_ownership_conflict_count} conflicts outlined " f"in the “Audio Next Steps” tab of the attached " f"Excel document." ) return "\n\n".join(text) if text else "" @property def sr_monetization_summary_first(self) -> str: """Text for the monetization summary slide [First].""" return ( f"After auditing the {self.sr_isrcs_unique_total_count} unique sound " f"recording{'s' if self.sr_isrcs_unique_total_count != 1 else ''} in your " f"audio catalog, {_pct(self.sr_after_monetizing_track_pct)} " f"of your audio catalog is enabled for monetization." ) @property def sr_monetization_summary_second(self) -> str: """Text for the monetization summary slide [Second].""" # Evaluate if monetization has increased using total counts, more # reliable than using percentage, which might have been rounded. if self.sr_monetization_variation_pct > 0: return ( f"This is a {_pct(self.sr_monetization_variation_pct)} increase in " f"monetization for your audio catalog." ) return "" @property def sr_ami(self) -> str: """Text for the SR audio monetization increase bubble.""" text = "+" if self.sr_monetization_variation_pct > 0 else "" return text + _pct(self.sr_monetization_variation_pct) @property def srugc_adv(self) -> str: """SRUGC Average Daily Views (formatted for output).""" return shorten_number(self.srugc_avg_daily_views) @property def srugc_mc(self) -> str: """SRUGC Match Count (formatted for output).""" return shorten_number(self.srugc_match_count) @property def at_monetization_summary(self) -> str: """Text for the art tracks summary slide [Second].""" # Evaluate if monetization has increased using total counts, more # reliable than using percentage, which might have been rounded. if self.at_monetization_variation_pct > 0: return ( f"This is a {_pct(self.at_monetization_variation_pct)} increase in " f"monetization for your Art Track catalog." ) return "" @dataclass class ImageData: """Dataclass for storing data to be inserted into the PPTX image placeholders. Attributes should be the lower case names of the placeholders in the PPTX template, without the delimiters. To define an image placeholder in the PPTX template, use the following format: `<>`. The placeholder must be present in the text of a shape in the PPTX template, and it must be the only text in the shape. The image will replace the shape containing the placeholder. """ top_artist_image: BytesIO def __post_init__(self): """Automatically convert WebP images to PNG, as they're not supported in PPTX files (or at least with the `python-pptx` library). """ if not self.top_artist_image: return is_webp = images.is_webp(self.top_artist_image) if is_webp: self.top_artist_image = BytesIO( images.convert_to_png(self.top_artist_image) ) def _pct(number: int | float, include_symbol: bool = True) -> str: """Return a string representation of a percent with variable precision. This is intended for use with percentages, so it will always return a string with one decimal places, except when the number has no decimals or all decimals are 0, in which case it will return an integer. Args: number: The number to convert to a percentage. include_symbol: Whether to include the percent symbol. """ return f"{var_precision(number, 1)}{'%' if include_symbol else ''}"