"""Contains the YTChecks class for encapsulating YouTube CMS checks, which can be used in the flagging checks logic. """ from collections import defaultdict from dataclasses import dataclass from .... import logger from ....config import YT_OWNER_ID_ORCHARD from ....connectors.youtube_apis import YtCidClient from ....connectors.youtube_apis.content_id import models from ....constants import Countries, YtCid from ....typings import AssetID, CountryCode2 logger = logger.new_logger(__name__) @dataclass(slots=True) class MatchPolicyRule: """Data class for a match policy rule. This standardizes the quite nested mapping of match policy rules in YouTube CMS, as provided by the API. Mapping example of an instance of this class: { "asset_id": "A671460307259259", "action": "monetize", "content_types": ["audiovisual"], "territories": ["RU"], "type": "exclude" } """ asset_id: AssetID action: str content_types: list[str] | None = None territories: list[CountryCode2] | None = None type: str | None = None def is_valid(self) -> bool: """Return if the match policy rule is valid in accordance to the YouTube Audits Team criteria. """ is_monetize = self.action == YtCid.MONETIZE valid_types = (self.content_types is None) or ( YtCid.AUDIOVISUAL in self.content_types ) valid_territories = (self.territories is None) or self.territories == [ Countries.RU # Ignore RU if present and alone ] valid_type = (self.type is None) or self.type == YtCid.EXCLUDE return is_monetize and valid_types and valid_territories and valid_type class Ownership(models.Ownership): """Data class for an ownership object. This extends the YouTube Content ID ownership object with additional methods and properties. """ pass class YTChecks: """Encapsulates YouTube CMS checks for aiding in flagging checks and uses internal YouTube resource caching for performance and quota efficiency. """ def __init__(self): self._yt_cid_client = YtCidClient() # For performance, we cache the assets, match policies, and references # to avoid fetching them multiple times for different checks during the # lifetime of the object. self._asset_cache: dict[AssetID, models.Asset | None] = {} self._reference_cache: dict[AssetID, list] = {} async def get_asset_match_policies( self, asset_ids: list[AssetID] ) -> dict[AssetID, list[MatchPolicyRule]]: """Get the match policies of each of the provided assets. Args: asset_ids: A list of YouTube asset IDs to get the match policies of. Returns: A dictionary where the keys are the asset IDs and the values are a list of match policy rules of the assets. """ results = await self._get_assets(asset_ids) match_policies = defaultdict(list) for asset in results: try: match_policy_rules = asset.match_policy[YtCid.RULES] except KeyError: logger.debug( "Asset {} has no match policy rules. Skipping...", asset.id ) continue for rule in match_policy_rules: conditions = rule.get(YtCid.CONDITIONS, {}) # Not always present required_territories = conditions.get(YtCid.REQUIRED_TERRITORIES, {}) match_policy = MatchPolicyRule( asset_id=asset.id, action=rule[YtCid.ACTION], content_types=conditions.get(YtCid.CONTENT_MATCH_TYPE, None), territories=required_territories.get(YtCid.TERRITORIES, None), type=required_territories.get(YtCid.TYPE, None), ) match_policies[asset.id].append(match_policy) return dict(match_policies) async def get_asset_ownerships_ours( self, asset_ids: list[AssetID] ) -> dict[AssetID, Ownership | None]: """Get our ownership objects of each of the provided assets. Args: asset_ids: A list of YouTube asset IDs to get the ownerships of. Returns: A dictionary where the keys are the asset IDs and the values are the ownership objects of the assets. If an asset is not owned by us, the value will be None. """ results = await self._get_assets(asset_ids) our_ownerships = {} while results: asset = results.pop() try: ownerships = asset.ownership[YtCid.GENERAL] except KeyError: # Asset has no ownerships. For whatever reason, it can happen # in edge cases. Skip to next. continue ours = next( filter( lambda x: x[YtCid.OWNER] == YT_OWNER_ID_ORCHARD, ownerships, ), None, ) our_ownerships[asset.id] = ( Ownership(**ours, asset_id=asset.id) if ours else None ) return our_ownerships async def get_asset_references( self, asset_ids: list[AssetID] ) -> dict[AssetID, list[models.Reference]]: """Get the references of each of the provided assets. Args: asset_ids: A list of YouTube asset IDs to get the references of. Returns: A dictionary where the keys are the asset IDs and the values are a list of references of the assets, as YouTube reference objects. """ cached_asset_ids = self._reference_cache.keys() fetch_asset_ids = set(asset_ids) - set(cached_asset_ids) # For efficiency, we only fetch the references of the assets that we don't # have in the cache yet. We only cache new references if we make a fresh # fetch. if fetch_asset_ids: fresh_references = await self._yt_cid_client.references_list( fetch_asset_ids ) for reference in fresh_references: self._reference_cache.setdefault(reference.asset_id, []).append( reference ) # Return the references for the requested asset IDs pulling them # from the cache, where both the previously cached and the fresh # references are now stored. references = { asset_id: self._reference_cache.get(asset_id, []) for asset_id in asset_ids } return references async def _get_assets(self, asset_ids: list[AssetID]) -> list[models.Asset]: """Get the assets with the provided asset IDs. Uses a short-term cache (which lives for the duration of the object) to avoid fetching the same asset multiple times for different checks, thus being more performant and consuming less YT CMS API quota. Args: asset_ids: A list of YouTube asset IDs to get the assets of. Returns: A list of YouTube asset objects. Assets that are not found will not be included in the list. """ cached_asset_keys = set(self._asset_cache.keys()) asset_ids_to_fetch = set(asset_ids) - cached_asset_keys if asset_ids_to_fetch: assets = await self._yt_cid_client.assets_list(asset_ids_to_fetch) for asset in assets: # Take into account that assets can be merged assets and thus have # alias IDs. Cache the asset for each of the asset IDs. this_asset_ids = self._get_all_asset_ids_for_asset(asset) for asset_id in this_asset_ids: self._asset_cache[asset_id] = asset try: asset_ids_to_fetch.remove(asset_id) except KeyError: # For merged assets, some of the asset IDs will not be in the # asset_ids_to_fetch set, so we can safely skip them. pass for asset_id in asset_ids_to_fetch: self._asset_cache[asset_id] = None return_assets = [ asset for asset in (self._asset_cache[asset_id] for asset_id in asset_ids) if asset is not None ] return return_assets @staticmethod def _get_all_asset_ids_for_asset(asset: models.Asset) -> list[AssetID]: """Get all the asset IDs for an asset, including the main asset ID and the alias IDs it may have (as a result of being a merged asset). Args: asset: A YouTube asset object. Returns: A list of asset IDs. """ return [asset.id] + (asset.alias_id or [])