"""Flag checks for SR audits."""

from typing import Self, Sequence

from pandas import DataFrame as Df
from pandas import Series

from ....constants import FlagResolutions, Flags, YTAssetNotActiveReasons
from ....logic.audit.logic.common import OutputColumns, SFCols, YTMatchPolicies
from ....models import NewAuditFlag
from ....typings import AssetID
from ....utils.html import bold
from ..logic import common
from .base import _FlagChecks, _FlagCleanUp
from .youtube_checks import YTChecks


class FlagChecksSR(_FlagChecks):
    """Class which contains all the flag checks for the SR audit."""

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)

        self._yt_checker = YTChecks()  # Keep as instance attribute to leverage cache

    def territories_missing(self, sr_missing_territories: Df) -> Self:
        """Flag in place rows with territories in registry but missing in YouTube."""
        sr = self.df
        self.add_flag(
            OutputColumns.AUDIT_FLAG_TERRITORIES_MISSING,
            right_df=sr_missing_territories,
            right_col=OutputColumns.TERRITORIES_MISSING,
        )
        flag_rows = sr[
            self._condition_has_asset_id
            & sr[OutputColumns.AUDIT_FLAG_TERRITORIES_MISSING]
            & sr[OutputColumns.AUDIT_FLAG_TERRITORIES_MISSING]
            .astype(str)
            .str.strip()
            .ne("")
        ]
        if len(flag_rows):
            common.add_flag_db(
                flag_rows,
                self.flagger,
                Flags.TERRITORIES_MISSING,
                OutputColumns.AUDIT_FLAG_TERRITORIES_MISSING,
            )
        return self

    def _no_active_references(
        self,
        sr_no_active_references: Df,
        *,
        right_col: str,
        flag_col: str,
        flag: str,
        **kwargs,
    ) -> Self:
        """Base function to flag rows with no active references. Do
        not call this function directly, use the specific methods
        leveraging it instead.

        Args:
            right_col: The column in the right DataFrame which contains the
                reasons why the references are not active.
            flag_col: The column where the flag will be stored.
            flag: The flag to add to the flagged rows.

        Kwargs:
            Any additional kwargs to pass to the `add_flag_db` function.
        """
        sr = self.df
        self.add_flag(flag_col, right_df=sr_no_active_references, right_col=right_col)
        flag_rows = sr[
            # Do not flag rows with missing asset ID (they will be flagged in
            # a more general way, as NO_ASSET_ID).
            self._condition_has_asset_id
            & sr[flag_col].notna()
            & sr[flag_col].ne(set())
            & sr[flag_col].astype(str).str.strip().ne("")
        ]
        if len(flag_rows):
            common.add_flag_db(
                flag_rows, self.flagger, flag, details=flag_col, **kwargs
            )
        return self

    def no_active_references_reasons_orchard(self, sr_no_active_references: Df) -> Self:
        """Flag in place rows with reasons why they're not active for Orchard."""

        # This is an ordered hierarchy of reasons which, if present,
        # will trigger an autoresolution of the flag with the corresponding
        # resolution subtype. Generally, not more than one reason can be
        # present, but the hierarchy is in place to handle the edge case
        # where multiple of them are present and to future proof the code.
        autoresolve_reasons_resolution_subtypes: Sequence[
            tuple[YTAssetNotActiveReasons, FlagResolutions]
        ] = [
            (
                YTAssetNotActiveReasons.INSUFFICIENT_LENGTH,
                FlagResolutions.REFERENCE_INSUFFICIENT_LENGTH,
            ),
            (
                YTAssetNotActiveReasons.ALMOST_ENTIRELY_EXCLUDED,
                FlagResolutions.REFERENCE_EXCLUDED,
            ),
            (
                YTAssetNotActiveReasons.CLOSED_BY_OWNER,
                FlagResolutions.REFERENCE_DEACTIVATED_BY_OWNER,
            ),
            (
                YTAssetNotActiveReasons.BULK_UPDATER,
                FlagResolutions.REFERENCE_DEACTIVATED_BY_OWNER,
            ),
        ]

        # For performance, create a set with the reasons which can be autoresolved,
        # to be able to check if a reason is in the set in O(1) time.
        autoresolve_reasons = {
            item[0] for item in autoresolve_reasons_resolution_subtypes
        }

        def autoresolution_handler(
            row: Series,
        ) -> tuple[FlagResolutions, FlagResolutions]:
            reasons = row[OutputColumns.AUDIT_FLAG_NO_ACTIVE_REFERENCES_REASONS_ORCHARD]
            resolution, resolution_subtype = None, None  # Default values

            autoresolvable_reasons = reasons & autoresolve_reasons
            if autoresolvable_reasons:
                for reason, subtype in autoresolve_reasons_resolution_subtypes:
                    if reason in reasons:
                        resolution = FlagResolutions.INELIGIBLE_REFERENCE
                        resolution_subtype = subtype
                        break

            return resolution, resolution_subtype

        return self._no_active_references(
            sr_no_active_references,
            right_col=OutputColumns.REASONS_ORCHARD,
            flag_col=OutputColumns.AUDIT_FLAG_NO_ACTIVE_REFERENCES_REASONS_ORCHARD,
            flag=Flags.NO_ACTIVE_REFERENCES_REASONS_ORCHARD,
            resolution_handler=autoresolution_handler,
        )

    def no_active_references_reasons_third_party(
        self, sr_no_active_references: Df
    ) -> Self:
        """Flag in place rows with reasons why they're not active for third party."""
        # This is an ordered hierarchy of reasons which, if present,
        # will trigger an autoresolution of the flag with the corresponding
        # resolution subtype. Generally, not more than one reason can be
        # present, but the hierarchy is in place to handle the edge case
        # where multiple of them are present and to future proof the code.
        autoresolve_reasons_resolution_subtypes: Sequence[
            tuple[YTAssetNotActiveReasons, FlagResolutions]
        ] = [
            (
                YTAssetNotActiveReasons.INSUFFICIENT_LENGTH,
                FlagResolutions.REFERENCE_INSUFFICIENT_LENGTH,
            ),
            (
                YTAssetNotActiveReasons.ALMOST_ENTIRELY_EXCLUDED,
                FlagResolutions.REFERENCE_EXCLUDED,
            ),
        ]

        # For performance, create a set with the reasons which can be autoresolved,
        # to be able to check if a reason is in the set in O(1) time.
        autoresolve_reasons = {
            item[0] for item in autoresolve_reasons_resolution_subtypes
        }

        def autoresolution_handler(
            row: Series,
        ) -> tuple[FlagResolutions, FlagResolutions]:
            reasons = row[
                OutputColumns.AUDIT_FLAG_NO_ACTIVE_REFERENCES_REASONS_THIRD_PARTY
            ]
            resolution, resolution_subtype = None, None  # Default values

            autoresolvable_reasons = reasons & autoresolve_reasons
            if autoresolvable_reasons:
                for reason, subtype in autoresolve_reasons_resolution_subtypes:
                    if reason in reasons:
                        resolution = FlagResolutions.INELIGIBLE_REFERENCE
                        resolution_subtype = subtype
                        break

            return resolution, resolution_subtype

        return self._no_active_references(
            sr_no_active_references,
            right_col=OutputColumns.REASONS_THIRD_PARTY,
            flag_col=OutputColumns.AUDIT_FLAG_NO_ACTIVE_REFERENCES_REASONS_THIRD_PARTY,
            flag=Flags.NO_ACTIVE_REFERENCES_REASONS_THIRD_PARTY,
            resolution_handler=autoresolution_handler,
        )

    def can_reactivate(self, sr_no_active_references: Df) -> Self:
        """Flag in place rows which can be reactivated."""
        sr = self.df
        self.add_flag(
            OutputColumns.AUDIT_FLAG_CAN_REACTIVATE,
            right_df=sr_no_active_references,
            right_col=OutputColumns.CAN_REACTIVATE,
        )

        falsy_values = {"", "0", "false"}

        _condition_flag_can_reactivate_is_truthy = (
            sr[OutputColumns.AUDIT_FLAG_CAN_REACTIVATE].notna()
            & sr[OutputColumns.AUDIT_FLAG_CAN_REACTIVATE].ne(set())
            & ~sr[OutputColumns.AUDIT_FLAG_CAN_REACTIVATE]
            .astype(str)
            .str.lower()
            .str.strip()
            .isin(falsy_values)
        )

        # Do not flag rows which were closed by the owner, as they can't be
        # reactivated.
        _not_closed_by_owner = sr[
            OutputColumns.AUDIT_FLAG_NO_ACTIVE_REFERENCES_REASONS_ORCHARD
        ].apply(
            lambda x: (
                not isinstance(x, set)
                or YTAssetNotActiveReasons.CLOSED_BY_OWNER not in x
            )
        )

        flag_rows = sr[
            # Do not flag rows with missing asset ID (they will be flagged in
            # a more general way, as NO_ASSET_ID).
            self._condition_has_asset_id
            & _condition_flag_can_reactivate_is_truthy
            & _not_closed_by_owner
        ]

        if len(flag_rows):
            common.add_flag_db(flag_rows, self.flagger, Flags.CAN_REACTIVATE)

        return self

    async def bad_match_policy(self) -> None:
        """Flag in place rows with bad match policy."""
        sr = self.df
        whitelisted_match_policies = {
            YTMatchPolicies.MONETIZE_IN_ALL_COUNTRIES,
            YTMatchPolicies.BLOCK_IN_ALL_COUNTRIES,
        }

        sr[OutputColumns.AUDIT_FLAG_BAD_MATCH_POLICY] = ~sr[SFCols.MATCH_POLICY].isin(
            whitelisted_match_policies
        )

        flag_rows = sr[
            # Do not flag rows with missing asset ID (they will be flagged in
            # a more general way, as NO_ASSET_ID).
            self._condition_has_asset_id
            & sr[OutputColumns.AUDIT_FLAG_BAD_MATCH_POLICY]
        ]

        # Check the match policies of the flagged assets in YouTube, and add
        # the match policies to the flag details, as context for the resolving
        # team.
        match_policies = await self._get_match_policies_as_string(
            flag_rows[SFCols.ASSET_ID].unique()
        )

        def details(row) -> str:
            policies = match_policies.get(row[SFCols.ASSET_ID], bold("?"))
            return f"Current match policies: {policies}"

        if len(flag_rows):
            common.add_flag_db(flag_rows, self.flagger, Flags.BAD_MATCH_POLICY, details)

    async def _get_match_policies_as_string(
        self, asset_ids: list[AssetID]
    ) -> dict[AssetID, str]:
        """Get the match policies of the provided asset IDs as a string. The
        match policies are formatted in HTML and they are in bold.

        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 the
            match policies of the assets as a string, e.g. "monetize (1), block (2)".
            The number in parentheses is the number of territories the match policy
            applies to. If an asset has no match policies, the value will be "None".
        """
        yt_data = await self._yt_checker.get_asset_match_policies(asset_ids)
        match_policies = {}

        for k, v in yt_data.items():
            strings = []
            for match_policy in v:
                action = match_policy.action
                territories = match_policy.territories
                brackets = "-"
                if territories:
                    brackets = f"{len(territories)}"
                    if policy_type := match_policy.type:
                        brackets = f"{policy_type} {brackets}"
                strings.append(f"{bold(action)} ({brackets})")
            match_policies[k] = ", ".join(strings) if strings else bold("None")

        return match_policies


class FlagCleanup(_FlagCleanUp):
    """Perform cleanup operations on the flag rows for the SR audit."""

    def __init__(self, flags: list[NewAuditFlag]) -> None:
        super().__init__(flags)

    def rule_can_reactivate(self) -> Self:
        """Rule to cleanup the CAN_REACTIVATE flag."""
        self._only_one_flag(
            SFCols.ASSET_ID,
            flag=Flags.CAN_REACTIVATE,
            discard_flags=[
                Flags.NO_ACTIVE_REFERENCES_REASONS_ORCHARD,
                Flags.NO_ACTIVE_REFERENCES_REASONS_THIRD_PARTY,
            ],
        )
        return self