"""Flag checks for MV audits.""" import asyncio import itertools from typing import Callable, Iterator, Self import numpy as np from pandas import DataFrame as Df from pandas import Series, isna from .... import logger from ....connectors.youtube_apis.content_id.models import Reference from ....constants import FlagDetails, FlagResolutions, Flags, MiscLower from ....logic.audit.logic.common import OutputColumns, SFCols from ....typings import AssetID from ....utils.html import bold from ....utils.pandas_helpers import add_col_if_not_exists from ..logic import common from . import conditions from .base import _FlagChecks from .youtube_checks import YTChecks logger = logger.new_logger(__name__) _runtime_err_msg = "Flag checks have not been run yet." class FlagChecksMV(_FlagChecks): """Class which contains all the flag checks for the MV audit.""" def __init__(self, df: Df, flagger: Callable) -> None: if df.empty: logger.info("No data provided for MV audit. Some checks will crash.") super().__init__(df, flagger) self._condition_privacy_status_public = ( conditions.has_privacy_status_public(self.df) if SFCols.VIDEO_PRIVACY_STATUS in self.df.columns else None ) self._yt_checker = YTChecks() # Keep as instance attribute to leverage cache async def ownership_incomplete(self) -> Self: """Flag in place rows with incomplete ownership / ownership not in the list of valid ownership types. """ mv = self.df async def ownership_invalid(row): if isna(row): return True try: check = await common.is_ownership_valid(row) except NotImplementedError: return True is_valid = check[0] return not is_valid mv.loc[ self._condition_has_asset_id, OutputColumns.AUDIT_FLAG_OWNERSHIP_INCOMPLETE ] = await asyncio.gather( # Async for performance *[ ownership_invalid(ownership) for ownership in mv.loc[self._condition_has_asset_id, SFCols.OWNERSHIP] ] ) flag_rows = mv[conditions.has_ownership_incomplete(mv)].copy(deep=False) if len(flag_rows): # Double check the ownership of the flagged rows in YouTube CMS, # remove the false positives. await self._double_check_yt_cid_ownerships(flag_rows) def _details(row) -> str | None: ownership_territories = row[SFCols.OWNERSHIP] if ( not isinstance(ownership_territories, set) or not ownership_territories ): # Catches NaNs, None, empty sets, etc. return "No ownership data available." return None common.add_flag_db( flag_rows, self.flagger, Flags.OWNERSHIP_INCOMPLETE, details=_details ) return self async def no_active_references(self, auto_resolve_length_lt: int = 30) -> Self: """Flag videos with no active references, regardless of whether they have an asset ID or not (and hence, whether they're in YouTube but aren't active, or they're not in YouTube at all, i.e. flag them even if they don't have an asset ID). Args: auto_resolve_length_lt: Videos with a duration less than this value (in seconds) will be auto-resolved. Any videos with NaN durations will not be auto-resolved. """ mv = self.df mv[OutputColumns.AUDIT_FLAG_NO_ACTIVE_REFERENCES] = np.nan handler = _NoActiveReferences(mv, auto_resolve_length_lt, self._yt_checker) await handler.run() # Auto-resolve the flag if the video duration is less than N seconds! flag_rows_auto_resolved = handler.rows_flagged_auto_resolvable flag_rows_not_auto_resolved = handler.rows_flagged_not_auto_resolvable for rows in [flag_rows_auto_resolved, flag_rows_not_auto_resolved]: if not len(rows): continue auto_resolve: bool = rows is flag_rows_auto_resolved common.add_flag_db( rows, self.flagger, Flags.NO_ACTIVE_REFERENCES, resolution=( FlagResolutions.INELIGIBLE_REFERENCE if auto_resolve else None ), resolution_subtype=( FlagResolutions.REFERENCE_INSUFFICIENT_LENGTH if auto_resolve else None ), ) return self def third_party_claim(self) -> Self: """Flag in place rows with no ISRC/Custom ID and no Asset ID, but do have a 3rd party claim. """ mv = self.df def details(row): other_owners_claiming = row[SFCols.OTHER_OWNERS_CLAIMING].split(",") other_owners_claiming = ( bold(owner) for owner in sorted(other_owners_claiming) ) other_owners_claiming = ", ".join(other_owners_claiming) return f"Claimed by: {other_owners_claiming}" flag_condition = conditions.has_third_party_claim(mv, other_claimants=True) mv.loc[flag_condition, OutputColumns.AUDIT_FLAG_THIRD_PARTY_CLAIM] = True if flag_condition.any(): flag_rows = mv[flag_condition] common.add_flag_db( flag_rows, self.flagger, Flags.THIRD_PARTY_CLAIM, details=details, ) return self def must_claim(self) -> Self: """Flag in place rows with no ISRC/Custom ID and no Asset ID, and without a 3rd party claim. """ mv = self.df flag_condition = conditions.has_third_party_claim(mv, other_claimants=False) mv.loc[flag_condition, OutputColumns.AUDIT_FLAG_MUST_CLAIM] = True if flag_condition.any(): flag_rows = mv[flag_condition] common.add_flag_db(flag_rows, self.flagger, Flags.MUST_CLAIM) return self def asset_missing_isrc(self) -> Self: """Flag in place public videos with no ISRC/Custom ID but have an Asset ID.""" mv = self.df mask = ( ~self._condition_has_isrc & self._condition_has_asset_id & self._condition_privacy_status_public ) mv.loc[mask, OutputColumns.AUDIT_FLAG_ASSET_MISSING_ISRC] = True flag_rows = mv[conditions.has_asset_missing_isrc(mv)] if len(flag_rows): common.add_flag_db( flag_rows, self.flagger, Flags.ASSET_MISSING_ISRC, ) return self async def bad_match_policy(self) -> Self: """Flag in place rows the match policies of which are not in the list of valid policies. Flag them even if there's no asset ID. """ mv = self.df handler = _BadMatchPolicy(mv, self._yt_checker) await handler.run() flag_rows = mv[mv[OutputColumns.AUDIT_FLAG_BAD_MATCH_POLICY]].copy() def _details(row): match_policy = row[SFCols.MATCH_POLICY] if not isinstance(match_policy, str) or not match_policy.strip(): # Catches NaNs, None, empty strings, etc. return bold(FlagDetails.NO_ACTIVE_MATCH_POLICY) return bold(row[SFCols.MATCH_POLICY]) if len(flag_rows): # Double check the match policies of the flagged rows in YouTube CMS, # remove the false positives. await self._double_check_yt_cid_match_policies(flag_rows) common.add_flag_db( flag_rows, self.flagger, Flags.BAD_MATCH_POLICY, details=_details, ) return self async def _double_check_yt_cid_ownerships(self, flag_rows: Df) -> None: """Double check the ownership of the flagged rows in YouTube CID. It will drop the rows that have been flagged and have the same ownership in YouTube CMS, as they're false positives. This will be done in place. Args: flag_rows: The rows that have been flagged. """ if flag_rows.empty: return # Asset IDs and ownerships are required to check the ownership in YouTube CMS, # so skip the rows that don't have them. rows_to_check = flag_rows[ self._condition_has_asset_id.reindex(flag_rows.index, fill_value=False) ] if rows_to_check.empty: return asset_ids = rows_to_check[SFCols.ASSET_ID].dropna().unique().tolist() found_ownerships = await self._yt_checker.get_asset_ownerships_ours(asset_ids) found_ownerships_ours = { k: v for k, v in found_ownerships.items() if v is not None # Will be None if no ownership of ours found } if not found_ownerships_ours: # No ownership of ours found in YT CID return length_before = len(flag_rows) yt_asset_owned_territories = { o.asset_id: o.territories for o in found_ownerships_ours.values() } indices_to_drop = [] for idx, row in rows_to_check.iterrows(): asset_id = row[SFCols.ASSET_ID.value] ownership = row[SFCols.OWNERSHIP.value] row_territories = ownership if isinstance(ownership, set) else set() yt_territories = set(yt_asset_owned_territories.get(asset_id, [])) if row_territories == yt_territories: # False positive: the ownership currently set in YT CID matches the # ownership in the row. indices_to_drop.append(idx) flag_rows.drop( indices_to_drop, inplace=True, ) logger.debug( "Found and unflagged {} ownership incomplete false positives!", length_before - len(flag_rows), ) async def _double_check_yt_cid_match_policies(self, flag_rows: Df) -> None: """Double check the match policies of the flagged rows in YouTube CMS. It will drop the rows that have been flagged and have the same match policy in YouTube CMS, as they're false positives. This will be done in place. Args: flag_rows: The rows that have been flagged. """ if flag_rows.empty: return # Asset IDs and match policies are required to check the match policy in # YouTube CMS, so skip the rows that don't have them. rows_to_check = flag_rows[ self._condition_has_asset_id.reindex(flag_rows.index, fill_value=False) & conditions.has_match_policy(flag_rows).reindex( flag_rows.index, fill_value=False ) ] if rows_to_check.empty: return asset_ids = rows_to_check[SFCols.ASSET_ID].dropna().unique().tolist() found_match_policies = await self._yt_checker.get_asset_match_policies( asset_ids ) if not found_match_policies: return false_positives = set() for asset_id, match_policy_rules in found_match_policies.items(): for rule in match_policy_rules: # Usually there's only one rule if rule.is_valid(): # If valid, it's a false positive false_positives.add(asset_id) break # No need to check any other rules for this asset flag_rows.drop( flag_rows[flag_rows[SFCols.ASSET_ID].isin(false_positives)].index, inplace=True, ) logger.debug( "Found and dropped {} bad match policy false positives!", len(false_positives), ) class _BadMatchPolicy: """Class to encapsulate the bad match policy flag checks.""" _flag_col = OutputColumns.AUDIT_FLAG_BAD_MATCH_POLICY def __init__(self, mv: Df, yt_checker: YTChecks = None) -> None: self.mv = mv self._yt_checker = yt_checker or YTChecks() add_col_if_not_exists(self.mv, self._flag_col) add_col_if_not_exists(self.mv, SFCols.MATCH_POLICY) async def run(self) -> None: """Run the bad match policy flag checks.""" # To enhance flagging, first fill in missing match policies by querying the # YouTube Content ID API. await self._populate_missing_match_policies() self._flag_rows() async def _populate_missing_match_policies(self) -> None: """For rows with missing match policies, query the YouTube Content ID API to check if the video is in YouTube and has a match policy. This is because Snowflake is sometimes missing this information. All rows with missing match policies will be updated in place. """ asset_ids_to_fetch = self._get_asset_ids_missing_match_policies() results = await self._yt_checker.get_asset_match_policies(asset_ids_to_fetch) match_policy_map = { asset_id: list({mp.action for mp in match_policies})[0] for asset_id, match_policies in results.items() if match_policies } # Use map and only update rows where a match is found (do not overwrite NaNs) self.mv[SFCols.MATCH_POLICY] = ( self.mv[SFCols.ASSET_ID] .map(match_policy_map) .combine_first(self.mv[SFCols.MATCH_POLICY]) ) def _get_asset_ids_missing_match_policies(self) -> list[AssetID]: """Get the asset IDs of rows with missing match policies.""" mask_no_match_policy = ~conditions.has_match_policy(self.mv) rows_with_no_match_policy = self.mv[mask_no_match_policy] if rows_with_no_match_policy.empty: return [] return rows_with_no_match_policy[SFCols.ASSET_ID].dropna().unique().tolist() def _flag_rows(self) -> None: """Flag rows with bad match policies. Those are not empty strings and not in the set of valid policies. Flags are added in place. Flag only applies to rows with asset IDs. """ invalid_policy = ~self.mv[SFCols.MATCH_POLICY].isin(common.VALID_POLICIES) has_asset_id = conditions.has_asset_id(self.mv) self.mv[self._flag_col] = has_asset_id & invalid_policy class _NoActiveReferences: """Class to encapsulate the no active references flag checks.""" # Mapping of the status to the respective reference columns in the DF _status_cols: dict[str, SFCols] = { MiscLower.ACTIVE: SFCols.ACTIVE_REFERENCE_IDS, MiscLower.INACTIVE: SFCols.INACTIVE_REFERENCE_IDS, } def __init__( self, mv: Df, auto_resolve_length_lt: int = 30, yt_checker: YTChecks = None ) -> None: self.mv = mv self.auto_resolve_length_lt = auto_resolve_length_lt self._yt_checker = yt_checker or YTChecks() self._has_been_run = False add_col_if_not_exists(self.mv, OutputColumns.AUDIT_FLAG_NO_ACTIVE_REFERENCES) @property def rows_flagged(self) -> Df: """Return a subset of flagged rows.""" if not self._has_been_run: raise RuntimeError(_runtime_err_msg) mask = self.mv[OutputColumns.AUDIT_FLAG_NO_ACTIVE_REFERENCES].eq(True) return self.mv[mask] @property def _auto_resolvable_mask(self) -> Series: """Mask for auto-resolved flagged rows.""" video_length = self.rows_flagged[SFCols.VIDEO_LENGTH] has_video_length = video_length.notna() too_short = video_length < self.auto_resolve_length_lt return has_video_length & too_short @property def rows_flagged_auto_resolvable(self) -> Df: """Return a subset of flagged rows eligible for being auto-resolved.""" if not self._has_been_run: raise RuntimeError(_runtime_err_msg) return self.rows_flagged[self._auto_resolvable_mask] @property def rows_flagged_not_auto_resolvable(self) -> Df: """Return a subset of flagged rows not eligible for being auto-resolved.""" if not self._has_been_run: raise RuntimeError(_runtime_err_msg) return self.rows_flagged[~self._auto_resolvable_mask] @property def rows_no_references(self) -> Df: """Return a subset of rows which have neither active nor inactive references (i.e. the active and inactive reference ID columns are both NaN). """ ref_cols = list(self._status_cols.values()) mask_ref_cols_na = self.mv[ref_cols].isna().all(axis=1) return self.mv[mask_ref_cols_na] async def run(self) -> None: """Run the no active references flag checks.""" await self._populate_missing_references() self._flag_rows() self._has_been_run = True async def _populate_missing_references(self) -> None: """For rows with neither active nor inactive references, query the YouTube Content ID API to check if the video is in YouTube and has references. This is because Snowflake is sometimes missing this information. """ mv = self.mv _asset_id, _id, _status = "asset_id", "id", "status" # Create the status columns if they don't exist add_col_if_not_exists(mv, self._status_cols.values()) # Fetch missing references references = await self._fetch_missing_references() # Create a DataFrame from references for better performance df_refs = Df(references, columns=[_asset_id, _id, _status]) subset_agg = df_refs.groupby([_asset_id, _status])[_id].agg(set).reset_index() for status, place_col in self._status_cols.items(): mask = subset_agg[_status].eq(status) matching_subset = ( subset_agg.loc[mask] .drop(columns=[_status]) .rename(columns={_id: place_col}) ) # Set the index to `_asset_id` for proper alignment with `self.mv` matching_subset.set_index(_asset_id, inplace=True) mv.set_index(_asset_id, inplace=True) mv.update(matching_subset) mv.reset_index(inplace=True) async def _fetch_missing_references(self) -> Iterator[Reference]: """Fetch from YouTube missing references for rows where no references are available (neither active nor inactive). Returns: An iterator of reference objects. """ rows_with_no_references = self.rows_no_references if rows_with_no_references.empty: return iter(()) logger.debug( "Fetching references from YouTube for {} rows with no references " "available...", len(rows_with_no_references), ) asset_ids_to_fetch = ( rows_with_no_references[SFCols.ASSET_ID].dropna().unique().tolist() ) results = await self._yt_checker.get_asset_references(asset_ids_to_fetch) return itertools.chain(*results.values()) def _flag_rows(self) -> None: """Flag rows with no active references.""" # The following isn't the most straightforward way to do this; it's quite # contorted. However, it's done this way to avoid certain Pandas futureWarnings # as of October 2024, and it passes the tests! has_asset_id = conditions.has_asset_id(self.mv) active_reference_ids = self.mv[SFCols.ACTIVE_REFERENCE_IDS] flags = has_asset_id & ( active_reference_ids.isna() # Check for NaNs | active_reference_ids.apply(lambda x: False if isna(x) else not bool(x)) | active_reference_ids.astype(str).str.strip().eq("") ) self.mv.loc[:, OutputColumns.AUDIT_FLAG_NO_ACTIVE_REFERENCES] = flags.astype( float ) # Reassign cells of rows without asset id to NaN. See previous comment. self.mv.loc[~has_asset_id, OutputColumns.AUDIT_FLAG_NO_ACTIVE_REFERENCES] = ( np.nan )