Spaces:
Running
Running
File size: 3,709 Bytes
c8d5b78 5c282df c8d5b78 5c282df c8d5b78 5c282df c8d5b78 5c282df c8d5b78 5c282df c946de7 c8d5b78 c946de7 c8d5b78 a74ad81 85d1765 94f494a afa039e c8d5b78 726d972 a7dd548 c8d5b78 726d972 c8d5b78 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | 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
|