from datetime import date
from typing import Optional

from pandas import DataFrame, Series, Timestamp, to_datetime


class DfUtils:
    @staticmethod
    def extract_languages(df: DataFrame) -> DataFrame:
        languages_df: DataFrame = df.copy()
        languages_df.loc[:, "languages"] = languages_df["github"].apply(
            lambda g: g.get("languages") if isinstance(g, dict) else None
        )
        return languages_df.drop(columns=["github"], errors="ignore")

    @staticmethod
    def filter_language(df: DataFrame, target_language: str) -> DataFrame:
        target_language_fold: str = target_language.casefold()
        mask: Series = df["languages"].apply(
            lambda languages: (
                any(language is not None and language.casefold() == target_language_fold for language in languages)
                if languages is not None
                else False
            )
        )
        return df[mask]

    @staticmethod
    def extract_tests_count(df: DataFrame) -> DataFrame:
        tests_count_df: DataFrame = df.copy()
        tests_count_df.loc[:, "tests_count"] = tests_count_df["github"].apply(
            lambda g: g.get("tests_count") if isinstance(g, dict) else None
        )
        return tests_count_df.drop(columns=["github"], errors="ignore")

    @staticmethod
    def extract_ai_tooling_markers(df: DataFrame) -> DataFrame:
        ai_tooling_markers_df: DataFrame = df.copy()
        ai_tooling_markers_df.loc[:, "ai_tooling_markers"] = ai_tooling_markers_df["github"].apply(
            lambda g: g.get("ai_tooling_markers") if isinstance(g, dict) else None
        )
        return ai_tooling_markers_df.drop(columns=["github"], errors="ignore")

    @staticmethod
    def extract_ai_declaration_markers(df: DataFrame) -> DataFrame:
        ai_declaration_markers_df: DataFrame = df.copy()
        ai_declaration_markers_df.loc[:, "ai_declaration_markers"] = ai_declaration_markers_df["anki_web"].apply(
            lambda a: a.get("ai_declaration_markers") if isinstance(a, dict) else None
        )
        return ai_declaration_markers_df.drop(columns=["anki_web"], errors="ignore")

    @staticmethod
    def extract_description_language(df: DataFrame) -> DataFrame:
        description_language_df: DataFrame = df.copy()
        description_language_df.loc[:, "description_language"] = description_language_df["anki_web"].apply(
            lambda a: a.get("description_language") if isinstance(a, dict) else None
        )
        return description_language_df.drop(columns=["anki_web"], errors="ignore")

    @staticmethod
    def extract_new_number(df: DataFrame, previous_df: Optional[DataFrame]) -> int:
        if previous_df is None:
            return 0
        previous_ids: set = set(previous_df["id"])
        is_new: Series = ~df["id"].isin(previous_ids)
        return int(is_new.sum())

    @staticmethod
    def extract_updated_number(
        df: DataFrame, day: date, previous_day: Optional[date], previous_df: Optional[DataFrame]
    ) -> int:
        anki_web_df: DataFrame = DataFrame(df["anki_web"].tolist())
        update_dates = to_datetime(anki_web_df["update_date"])
        if previous_day is not None:
            is_updated: Series = (update_dates >= Timestamp(previous_day)) & (update_dates < Timestamp(day))
        else:
            is_updated: Series = (update_dates.dt.year == day.year) & (update_dates.dt.month == day.month)
        if previous_df is not None:
            previous_ids: set = set(previous_df["id"])
            is_existing: Series = df["id"].isin(previous_ids).to_numpy()
            is_updated = is_updated & is_existing
        count: int = int(is_updated.sum())
        return count