"""This module contains functions that return boolean Series indicating whether certain conditions are met for assets in the provided DataFrame. The functions in this module are used in the flag checks. """ from enum import StrEnum from typing import Any from pandas import DataFrame from ....constants import MiscLower, OutputColumns from ....constants import SnowFlakeColumns as SFCols def has_non_empty_str(df: DataFrame, col: str) -> DataFrame: """Return a boolean Series indicating whether the values in the specified column are non-empty strings. Args: df (DataFrame): DataFrame containing the assets. col (str): The column to check. Returns: A boolean Series indicating whether the values in the specified column are non-empty strings (or non-empty StrEnums). """ non_nas = df[col].fillna("") all_strings = non_nas.apply(_is_string_or_str_enum) no_empty_strings = non_nas.astype(str).str.strip().ne("") return all_strings & no_empty_strings def _has_non_empty_collection(df: DataFrame, col: str) -> DataFrame: """Return a boolean Series indicating whether the values in the specified column are non-empty collections. Non-empty collections are lists, sets, or strings with at least one non-empty element. See tests for examples. Args: df (DataFrame): DataFrame containing the assets. col (str): The column to check. Returns: A boolean Series indicating whether the values in the specified column are non-empty collections. """ return df[col].apply( lambda x: ( len([e for e in x if e.strip()]) > 0 if isinstance(x, (list, set, str)) else False ) ) def has_isrc(df: DataFrame) -> DataFrame: """Return a boolean Series indicating whether the assets have an ISRC. Returns: A boolean Series indicating whether the videos have an ISRC. """ return has_non_empty_str(df, SFCols.ISRC) def has_asset_id(df: DataFrame) -> DataFrame: """Return a boolean Series indicating whether the assets have an asset ID. Returns: A boolean Series indicating whether the videos have an asset ID. """ return has_non_empty_str(df, SFCols.ASSET_ID) def has_other_owners_claiming(df: DataFrame) -> DataFrame: """Return a boolean Series indicating whether the assets have other owners claiming. Returns: A boolean Series indicating whether the videos have other owners claiming. """ return _has_non_empty_collection(df, SFCols.OTHER_OWNERS_CLAIMING) def has_third_party_claim(df: DataFrame, other_claimants: bool = True) -> DataFrame: """Return a boolean Series indicating whether the assets have a third party claim. Args: df (DataFrame): DataFrame containing the assets. other_claimants (bool): Whether to consider other claimants. Defaults to True. If False, the result will signal which assets can be claimed due to the absence of other claimants. Returns: A boolean Series indicating whether the videos have a third party claim or can be claimed due to the absence of other claimants. """ is_public = has_privacy_status_public(df) has_other_claims = has_other_owners_claiming(df) if other_claimants: return is_public & has_other_claims return is_public & ~has_other_claims & ~has_asset_id(df) def has_ownership(df: DataFrame) -> DataFrame: """Return a boolean Series indicating whether the assets have ownership. Returns: A boolean Series indicating whether the videos have ownership. """ return has_non_empty_str(df, SFCols.OWNERSHIP) def has_match_policy(df: DataFrame) -> DataFrame: """Return a boolean Series indicating whether the assets have a match policy. Returns: A boolean Series indicating whether the videos have a match policy. """ return has_non_empty_str(df, SFCols.MATCH_POLICY) def has_ownership_incomplete(df: DataFrame) -> DataFrame: """Return a boolean Series indicating whether the assets have incomplete ownership. Returns: A boolean Series indicating whether the videos have incomplete ownership. """ return df[OutputColumns.AUDIT_FLAG_OWNERSHIP_INCOMPLETE].eq(True) def has_asset_missing_isrc(df: DataFrame) -> DataFrame: """Return a boolean Series indicating whether the assets have a missing ISRC. Returns: A boolean Series indicating whether the videos have a missing ISRC. """ return df[OutputColumns.AUDIT_FLAG_ASSET_MISSING_ISRC].eq(True) def _has_privacy_status(df: DataFrame, status: str) -> DataFrame: """Return a boolean Series indicating whether the privacy status of the assets is the provided status. Args: df (DataFrame): DataFrame containing the assets. status (str): The privacy status to check for, e.g. "public". Returns: A boolean Series indicating whether the privacy status of the videos is public. """ return df[SFCols.VIDEO_PRIVACY_STATUS].fillna("").str.lower().eq(status) def has_privacy_status_public(df: DataFrame) -> DataFrame: """Return a boolean Series indicating whether the privacy status of the assets is public. Returns: A boolean Series indicating whether the privacy status of the videos is public. """ return _has_privacy_status(df, MiscLower.PUBLIC) def has_topic_channel(df: DataFrame) -> DataFrame: """Return a boolean Series indicating whether the assets have a topic channel. Returns: A boolean Series indicating whether the videos have a topic channel. """ col = SFCols.CHANNEL_DISPLAY_NAME return df[col].apply(type).eq(str) & has_non_empty_str(df, col) def _is_string_or_str_enum(value: Any) -> bool: """Return whether the value is a string or a StrEnum.""" return isinstance(value, (str, StrEnum))