Spaces:
Running
Running
| from datetime import date | |
| from typing import Optional | |
| from pandas import DataFrame, Series, Timestamp, to_datetime | |
| class DfUtils: | |
| 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") | |
| 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] | |
| 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") | |
| 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") | |
| 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") | |
| 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") | |
| 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()) | |
| 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 | |