diff --git a/scripts/bigquery_aggraget_fact_selected_layer.py b/scripts/bigquery_aggraget_fact_selected_layer.py index 4db7a9e..5952b15 100644 --- a/scripts/bigquery_aggraget_fact_selected_layer.py +++ b/scripts/bigquery_aggraget_fact_selected_layer.py @@ -435,47 +435,41 @@ def _detect_consistency(df_ind: pd.DataFrame, lower_better: bool) -> tuple: # ============================================================================= -# NARRATIVE BUILDER +# NARRATIVE BUILDER — PER INDICATOR PER YEAR (1 pillar, 1 tahun) +# ============================================================================= +# +# Granularity agg_narrative_indicator SEKARANG per (indicator_id, year), bukan +# lagi per indicator_id gabungan seluruh tahun. Setiap baris narasi menjelaskan +# posisi 1 indikator dalam 1 pillar, pada 1 tahun tertentu: nilai regional, +# skor ternormalisasi, peringkat indikator tsb di antara indikator lain dalam +# pillar yang sama pada tahun yang sama, perubahan YoY, serta negara terbaik/ +# terlemah untuk indikator tsb pada tahun tersebut. # ============================================================================= -def _build_narrative_per_indicator(row: pd.Series, df_full: pd.DataFrame) -> tuple: - ind_id = int(row["indicator_id"]) +def _build_narrative_per_indicator_year(row: pd.Series) -> tuple: + """ + row diharapkan datang dari baris ASEAN (country_id=0) pada agg_indicator_norm, + sudah digabung dengan kolom rank_in_pillar_year, n_indicators_in_pillar_year, + country_best, country_worst (dihitung dari negara asli untuk indicator+year yang sama). + """ ind_name_en = str(row["indicator_name"]).strip() ind_name_id = str(row.get("indicator_name_id", ind_name_en)).strip() unit = str(row["unit"]).strip() if row["unit"] else "" - direction = str(row["direction"]).strip() pillar_en = str(row["pillar_name"]).strip() pillar_id_ = get_pillar_name_id(pillar_en) framework = str(row["framework"]).strip() - year_min = int(row["year_min"]) - year_max = int(row["year_max"]) - lower_better = _is_lower_better(direction) + year = int(row["year"]) - # Gunakan hanya negara asli (bukan ASEAN) untuk analisa tren/gap/konsistensi - df_ind = df_full[ - (df_full["indicator_id"] == ind_id) & - (df_full["country_id"] != ASEAN_COUNTRY_ID) - ].copy() + value = row.get("value", np.nan) + score = row.get("norm_score_1_100", np.nan) + yoy = row.get("yoy_value", np.nan) + rank = row.get("rank_in_pillar_year", np.nan) + n_ind = row.get("n_indicators_in_pillar_year", np.nan) - if df_ind.empty: - na_en = f"{ind_name_en} ({framework}, {pillar_en}): Insufficient data for analysis." - na_id = f"{ind_name_id} ({framework}, {pillar_id_}): Data tidak cukup untuk dianalisis." - return na_en, na_id - - asean_avg_by_year = ( - df_ind.groupby("year")["value"].mean().dropna() - ) - - trend_label = _detect_trend(asean_avg_by_year, lower_better) - gap_label = _detect_gap_trend(df_ind, lower_better) - anomaly_year, anomaly_dir = _detect_anomaly_year(asean_avg_by_year) - best_country_en, worst_country_en, is_consistent = _detect_consistency(df_ind, lower_better) - - best_country_id = get_country_name_id(best_country_en) if best_country_en else None - worst_country_id = get_country_name_id(worst_country_en) if worst_country_en else None - - avg_first = row.get("avg_value_first", np.nan) - avg_last = row.get("avg_value_last", np.nan) + best_country_en = row.get("country_best") + worst_country_en = row.get("country_worst") + best_country_id = row.get("country_best_id") + worst_country_id = row.get("country_worst_id") def fmt(v): if pd.isna(v): @@ -487,65 +481,65 @@ def _build_narrative_per_indicator(row: pd.Series, df_full: pd.DataFrame) -> tup sentences_en = [] sentences_id = [] - s1_en = f"{ind_name_en} ({framework}, {pillar_en}, {year_min}-{year_max}):" - s1_id = f"{ind_name_id} ({framework}, {pillar_id_}, {year_min}-{year_max}):" + perf_word_en = "good" if (pd.notna(score) and score >= _PERFORMANCE_THRESHOLD) else "below target" + perf_word_id = "baik" if (pd.notna(score) and score >= _PERFORMANCE_THRESHOLD) else "di bawah target" + + if pd.notna(rank) and pd.notna(n_ind) and int(n_ind) > 0: + rank_i = int(rank) + n_i = int(n_ind) + suffix = {1: "st", 2: "nd", 3: "rd"}.get(rank_i, "th") + s1_en = ( + f"In {year}, {ind_name_en} ({framework}, {pillar_en}) recorded a regional average of " + f"{fmt(value)}, ranking {rank_i}{suffix} out of {n_i} indicators within the {pillar_en} " + f"pillar for that year, with a normalized score of {fmt(score)} ({perf_word_en})." + ) + s1_id = ( + f"Pada tahun {year}, {ind_name_id} ({framework}, {pillar_id_}) mencatat rata-rata " + f"regional sebesar {fmt(value)}, menempati peringkat {rank_i} dari {n_i} indikator " + f"dalam pilar {pillar_id_} pada tahun tersebut, dengan skor ternormalisasi " + f"{fmt(score)} ({perf_word_id})." + ) + else: + s1_en = ( + f"{ind_name_en} ({framework}, {pillar_en}, {year}): regional average was {fmt(value)}, " + f"with a normalized score of {fmt(score)} ({perf_word_en})." + ) + s1_id = ( + f"{ind_name_id} ({framework}, {pillar_id_}, {year}): rata-rata regional sebesar " + f"{fmt(value)}, dengan skor ternormalisasi {fmt(score)} ({perf_word_id})." + ) sentences_en.append(s1_en) sentences_id.append(s1_id) - trend_map_en = { - "improving_consistent": f"Regional average improved consistently from {fmt(avg_first)} to {fmt(avg_last)}.", - "improving_slowing": f"Regional average improved from {fmt(avg_first)} to {fmt(avg_last)}, though the pace slowed in recent years.", - "deteriorating": f"Regional average worsened from {fmt(avg_first)} to {fmt(avg_last)} over the period.", - "fluctuating": f"Regional average fluctuated between {fmt(avg_first)} and {fmt(avg_last)} with no clear trend.", - "insufficient_data": f"Trend analysis is limited due to sparse data.", - } - trend_map_id = { - "improving_consistent": f"Rata-rata regional membaik secara konsisten dari {fmt(avg_first)} menjadi {fmt(avg_last)}.", - "improving_slowing": f"Rata-rata regional membaik dari {fmt(avg_first)} menjadi {fmt(avg_last)}, namun lajunya melambat dalam beberapa tahun terakhir.", - "deteriorating": f"Rata-rata regional memburuk dari {fmt(avg_first)} menjadi {fmt(avg_last)} sepanjang periode.", - "fluctuating": f"Rata-rata regional berfluktuasi antara {fmt(avg_first)} dan {fmt(avg_last)} tanpa tren yang jelas.", - "insufficient_data": f"Analisis tren terbatas karena data yang tersedia tidak cukup.", - } - sentences_en.append(trend_map_en.get(trend_label, "")) - sentences_id.append(trend_map_id.get(trend_label, "")) - - if gap_label == "widening": - sentences_en.append("Disparity among ASEAN countries has widened over time, indicating unequal progress.") - sentences_id.append("Kesenjangan antar negara ASEAN melebar seiring waktu, menunjukkan kemajuan yang tidak merata.") - elif gap_label == "narrowing": - sentences_en.append("Disparity among ASEAN countries has narrowed, suggesting more balanced regional progress.") - sentences_id.append("Kesenjangan antar negara ASEAN menyempit, mengindikasikan kemajuan regional yang lebih merata.") - elif gap_label == "stable": - sentences_en.append("The gap among ASEAN countries remained relatively stable throughout the period.") - sentences_id.append("Kesenjangan antar negara ASEAN relatif stabil sepanjang periode.") - - if anomaly_year is not None: - if anomaly_dir == "drop": - sentences_en.append(f"A notable decline was recorded in {anomaly_year}, which stood out from the overall pattern.") - sentences_id.append(f"Penurunan signifikan tercatat pada tahun {anomaly_year}, yang menyimpang dari pola keseluruhan.") - elif anomaly_dir == "rise": - sentences_en.append(f"A sharp improvement was observed in {anomaly_year}, standing out from the overall pattern.") - sentences_id.append(f"Peningkatan tajam tercatat pada tahun {anomaly_year}, yang menyimpang dari pola keseluruhan.") - - if best_country_en and worst_country_en: - if is_consistent: - sentences_en.append( - f"{best_country_en} consistently performed above the regional average, " - f"while {worst_country_en} consistently lagged behind." - ) - sentences_id.append( - f"{best_country_id} secara konsisten berada di atas rata-rata regional, " - f"sementara {worst_country_id} secara konsisten tertinggal." - ) + if pd.notna(yoy): + if abs(yoy) < 1e-9: + s2_en = "This value was unchanged compared to the previous year." + s2_id = "Nilai ini tidak berubah dibandingkan tahun sebelumnya." + elif yoy > 0: + s2_en = f"This represents an increase of {fmt(abs(yoy))} from the previous year." + s2_id = f"Ini menunjukkan kenaikan sebesar {fmt(abs(yoy))} dari tahun sebelumnya." else: - sentences_en.append( - f"Overall, {best_country_en} showed the best performance, " - f"while {worst_country_en} had the weakest results across the period." - ) - sentences_id.append( - f"Secara keseluruhan, {best_country_id} menunjukkan performa terbaik, " - f"sementara {worst_country_id} memiliki hasil terlemah sepanjang periode." - ) + s2_en = f"This represents a decrease of {fmt(abs(yoy))} from the previous year." + s2_id = f"Ini menunjukkan penurunan sebesar {fmt(abs(yoy))} dari tahun sebelumnya." + sentences_en.append(s2_en) + sentences_id.append(s2_id) + else: + s2_en = "Year-over-year comparison is not available for this year." + s2_id = "Perbandingan tahun-ke-tahun tidak tersedia untuk tahun ini." + sentences_en.append(s2_en) + sentences_id.append(s2_id) + + if best_country_en and worst_country_en and best_country_en != worst_country_en: + s3_en = ( + f"Among ASEAN countries in {year}, {best_country_en} recorded the best performance " + f"for this indicator, while {worst_country_en} recorded the weakest." + ) + s3_id = ( + f"Di antara negara ASEAN pada tahun {year}, {best_country_id} mencatat performa " + f"terbaik untuk indikator ini, sementara {worst_country_id} mencatat performa terlemah." + ) + sentences_en.append(s3_en) + sentences_id.append(s3_id) narrative_en = " ".join(s for s in sentences_en if s) narrative_id = " ".join(s for s in sentences_id if s) @@ -959,259 +953,138 @@ class IndicatorNormAggregator: return rows_loaded # ========================================================================= - # STEP 11: agg_narrative_indicator (per indicator, ASEAN summary sebagai kolom) + # STEP 11: agg_narrative_indicator (per indicator_id PER YEAR, 1 pillar 1 tahun) # ========================================================================= def _build_narrative_table(self, df_final: pd.DataFrame): self.logger.info("\n" + "=" * 80) self.logger.info("STEP 11: agg_narrative_indicator") - self.logger.info(" Granularity: per indicator_id") - self.logger.info(" ASEAN data: digunakan untuk asean_avg_value_first/last") + self.logger.info(" Granularity: per indicator_id PER YEAR (1 pillar, 1 tahun)") + self.logger.info(" Narasi menjelaskan posisi indikator dalam pillar-nya, per tahun") self.logger.info("=" * 80) - # Negara asli saja untuk analisa statistik - df_real = df_final[df_final["country_id"] != ASEAN_COUNTRY_ID] - df_asean = df_final[df_final["country_id"] == ASEAN_COUNTRY_ID] + # Negara asli saja untuk analisa negara terbaik/terlemah per tahun + df_real = df_final[df_final["country_id"] != ASEAN_COUNTRY_ID].copy() + # Baris ASEAN (regional average) menjadi basis nilai per indicator per year + df_asean = df_final[df_final["country_id"] == ASEAN_COUNTRY_ID].copy() - # ---- Statistik per indikator (negara asli) ---- - df_yr = ( + if df_asean.empty: + self.logger.warning(" [WARNING] Tidak ada baris ASEAN; agg_narrative_indicator kosong.") + df_asean = df_final.copy() + + # ---- Rank indikator dalam pillar yang sama, pada tahun yang sama ---- + # (dihitung dari norm_score_1_100 baris ASEAN/regional; skor sudah + # searah -- semakin tinggi semakin baik -- untuk semua indikator) + df_asean["rank_in_pillar_year"] = ( + df_asean.groupby(["pillar_id", "year"])["norm_score_1_100"] + .rank(method="min", ascending=False) + ) + df_asean["n_indicators_in_pillar_year"] = ( + df_asean.groupby(["pillar_id", "year"])["indicator_id"] + .transform("nunique") + ) + + # ---- Negara terbaik / terlemah per indikator, per tahun (negara asli) ---- + def _best_worst(g: pd.DataFrame) -> pd.Series: + g_valid = g[g["norm_score_1_100"].notna()] + if g_valid.empty: + return pd.Series({"country_best": None, "country_worst": None}) + best_row = g_valid.loc[g_valid["norm_score_1_100"].idxmax()] + worst_row = g_valid.loc[g_valid["norm_score_1_100"].idxmin()] + return pd.Series({ + "country_best" : best_row["country_name"], + "country_worst": worst_row["country_name"], + }) + + country_stats = ( df_real.groupby(["indicator_id", "year"]) - .agg( - avg_value =("value", "mean"), - avg_norm_score =("norm_score_1_100", "mean"), - n_countries_yr =("country_id", "nunique"), - ) + .apply(_best_worst) .reset_index() ) - - df_first = ( - df_yr.sort_values("year").groupby("indicator_id").first().reset_index() - [["indicator_id", "year", "avg_value"]] - .rename(columns={"year": "year_min", "avg_value": "avg_value_first"}) + country_stats["country_best_id"] = country_stats["country_best"].apply( + lambda x: get_country_name_id(x) if pd.notna(x) and x is not None else None ) - df_last = ( - df_yr.sort_values("year").groupby("indicator_id").last().reset_index() - [["indicator_id", "year", "avg_value"]] - .rename(columns={"year": "year_max", "avg_value": "avg_value_last"}) - ) - df_score_avg = ( - df_yr.groupby("indicator_id") - .agg(avg_norm_score_1_100=("avg_norm_score", "mean")) - .reset_index() - ) - df_nc = ( - df_real.groupby("indicator_id")["country_id"] - .nunique().reset_index() - .rename(columns={"country_id": "n_countries"}) + country_stats["country_worst_id"] = country_stats["country_worst"].apply( + lambda x: get_country_name_id(x) if pd.notna(x) and x is not None else None ) - # ASEAN avg per indikator - df_asean_yr = ( - df_asean.groupby(["indicator_id", "year"]) - .agg(asean_avg_value=("value", "mean")) - .reset_index() - ) - df_asean_first = ( - df_asean_yr.sort_values("year").groupby("indicator_id").first().reset_index() - [["indicator_id", "asean_avg_value"]] - .rename(columns={"asean_avg_value": "asean_avg_value_first"}) - ) - df_asean_last = ( - df_asean_yr.sort_values("year").groupby("indicator_id").last().reset_index() - [["indicator_id", "asean_avg_value"]] - .rename(columns={"asean_avg_value": "asean_avg_value_last"}) - ) + # ---- Gabung ---- + df_agg = df_asean.merge(country_stats, on=["indicator_id", "year"], how="left") - # YoY stats (negara asli) - dir_map = ( - df_real[["indicator_id", "direction"]] - .drop_duplicates(subset=["indicator_id"]) - .set_index("indicator_id")["direction"] - .to_dict() - ) - - yoy_parts = [] - for ind_id, grp in df_yr.groupby("indicator_id"): - grp = grp.sort_values("year").copy() - grp["prev_avg"] = grp["avg_value"].shift(1) - grp["yoy"] = np.where( - grp["avg_value"].notna() & grp["prev_avg"].notna(), - grp["avg_value"] - grp["prev_avg"], - np.nan, - ) - grp = grp.drop(columns=["prev_avg"]) - yoy_parts.append(grp) - df_yr = pd.concat(yoy_parts, ignore_index=True) - - def _is_positive_yoy(ind_id, yoy_val): - if pd.isna(yoy_val): - return False - lb = _is_lower_better(dir_map.get(ind_id, "positive")) - return (yoy_val < 0) if lb else (yoy_val > 0) - - yoy_stats = [] - for ind_id, grp in df_yr.groupby("indicator_id"): - grp_yoy = grp[grp["yoy"].notna()].copy() - lb = _is_lower_better(dir_map.get(ind_id, "positive")) - n_total = len(grp_yoy) - n_positive = int(sum(_is_positive_yoy(ind_id, v) for v in grp_yoy["yoy"])) - - if n_total > 0: - idx_best = grp_yoy["yoy"].idxmin() if lb else grp_yoy["yoy"].idxmax() - best_row = grp_yoy.loc[idx_best] - best_yoy_from = best_row["year"] - 1 - best_yoy_to = best_row["year"] - else: - best_yoy_from = np.nan - best_yoy_to = np.nan - - yoy_stats.append({ - "indicator_id" : ind_id, - "n_yoy_total" : n_total, - "n_yoy_positive": n_positive, - "best_yoy_from" : best_yoy_from, - "best_yoy_to" : best_yoy_to, - }) - df_yoy_stats = pd.DataFrame(yoy_stats) - - # Country best/worst - df_country_avg = ( - df_real.groupby(["indicator_id", "country_id", "country_name"]) - .agg(country_avg_value=("value", "mean")) - .reset_index() - ) - country_stats = [] - for ind_id, grp in df_country_avg.groupby("indicator_id"): - lb = _is_lower_better(dir_map.get(ind_id, "positive")) - if lb: - worst_row = grp.loc[grp["country_avg_value"].idxmax()] - best_row = grp.loc[grp["country_avg_value"].idxmin()] - else: - worst_row = grp.loc[grp["country_avg_value"].idxmin()] - best_row = grp.loc[grp["country_avg_value"].idxmax()] - country_stats.append({ - "indicator_id" : ind_id, - "country_worst" : worst_row["country_name"], - "country_best" : best_row["country_name"], - "country_worst_id": get_country_name_id(worst_row["country_name"]), - "country_best_id" : get_country_name_id(best_row["country_name"]), - }) - df_country_stats = pd.DataFrame(country_stats) - - # Dim cols - dim_cols = [ - "indicator_name", "indicator_name_id", - "unit", "direction", - "pillar_name", "pillar_name_id", - "framework", - ] - df_dim = df_real[["indicator_id"] + dim_cols].drop_duplicates(subset=["indicator_id"]) - - # Merge semua - df_agg = ( - df_dim - .merge(df_first, on="indicator_id", how="left") - .merge(df_last, on="indicator_id", how="left") - .merge(df_score_avg, on="indicator_id", how="left") - .merge(df_nc, on="indicator_id", how="left") - .merge(df_yoy_stats, on="indicator_id", how="left") - .merge(df_country_stats, on="indicator_id", how="left") - .merge(df_asean_first, on="indicator_id", how="left") - .merge(df_asean_last, on="indicator_id", how="left") - ) - - # Performance - df_agg["performance"] = pd.NA - has_score = df_agg["avg_norm_score_1_100"].notna() - df_agg.loc[has_score & (df_agg["avg_norm_score_1_100"] >= _PERFORMANCE_THRESHOLD), "performance"] = "Good" - df_agg.loc[has_score & (df_agg["avg_norm_score_1_100"] < _PERFORMANCE_THRESHOLD), "performance"] = "Bad" - - # Build narrative + # ---- Build narrative per baris (per indicator_id per year) ---- narratives_en = [] narratives_id = [] for _, row in df_agg.iterrows(): - n_en, n_id = _build_narrative_per_indicator(row, df_final) + n_en, n_id = _build_narrative_per_indicator_year(row) narratives_en.append(n_en) narratives_id.append(n_id) df_agg["narrative_en"] = narratives_en df_agg["narrative_id"] = narratives_id - # Output + # ---- Output ---- out = df_agg[[ + "year", "indicator_id", "indicator_name", "indicator_name_id", "unit", "direction", - "pillar_name", "pillar_name_id", + "pillar_id", "pillar_name", "pillar_name_id", "framework", - "year_min", "year_max", "n_countries", - "avg_value_first", "avg_value_last", - "asean_avg_value_first", "asean_avg_value_last", - "avg_norm_score_1_100", "performance", - "n_yoy_total", "n_yoy_positive", - "best_yoy_from", "best_yoy_to", - "country_worst", "country_best", - "country_worst_id", "country_best_id", + "value", "norm_score_1_100", "performance", + "yoy_value", + "rank_in_pillar_year", "n_indicators_in_pillar_year", + "country_best", "country_worst", + "country_best_id", "country_worst_id", "narrative_en", "narrative_id", ]].copy() - out = out.sort_values(["pillar_name", "indicator_name"]).reset_index(drop=True) + out = out.sort_values(["year", "pillar_name", "rank_in_pillar_year", "indicator_name"]).reset_index(drop=True) - out["indicator_id"] = out["indicator_id"].astype(int) - out["indicator_name"] = out["indicator_name"].astype(str) - out["indicator_name_id"] = out["indicator_name_id"].astype(str) - out["unit"] = out["unit"].fillna("").astype(str) - out["direction"] = out["direction"].astype(str) - out["pillar_name"] = out["pillar_name"].astype(str) - out["pillar_name_id"] = out["pillar_name_id"].astype(str) - out["framework"] = out["framework"].astype(str) - out["year_min"] = out["year_min"].astype(int) - out["year_max"] = out["year_max"].astype(int) - out["n_countries"] = out["n_countries"].astype(int) - out["avg_value_first"] = pd.to_numeric(out["avg_value_first"], errors="coerce").astype(float) - out["avg_value_last"] = pd.to_numeric(out["avg_value_last"], errors="coerce").astype(float) - out["asean_avg_value_first"]= pd.to_numeric(out["asean_avg_value_first"], errors="coerce").astype(float) - out["asean_avg_value_last"] = pd.to_numeric(out["asean_avg_value_last"], errors="coerce").astype(float) - out["avg_norm_score_1_100"] = pd.to_numeric(out["avg_norm_score_1_100"], errors="coerce").astype(float) - out["performance"] = out["performance"].astype(str).replace("nan", pd.NA).astype("string") - out["n_yoy_total"] = pd.to_numeric(out["n_yoy_total"], errors="coerce").astype("Int64") - out["n_yoy_positive"] = pd.to_numeric(out["n_yoy_positive"], errors="coerce").astype("Int64") - out["best_yoy_from"] = pd.to_numeric(out["best_yoy_from"], errors="coerce").astype("Int64") - out["best_yoy_to"] = pd.to_numeric(out["best_yoy_to"], errors="coerce").astype("Int64") - out["country_worst"] = out["country_worst"].astype(str).replace("nan", pd.NA).astype("string") - out["country_best"] = out["country_best"].astype(str).replace("nan", pd.NA).astype("string") - out["country_worst_id"] = out["country_worst_id"].astype(str).replace("nan", pd.NA).astype("string") - out["country_best_id"] = out["country_best_id"].astype(str).replace("nan", pd.NA).astype("string") - out["narrative_en"] = out["narrative_en"].astype(str) - out["narrative_id"] = out["narrative_id"].astype(str) + out["year"] = out["year"].astype(int) + out["indicator_id"] = out["indicator_id"].astype(int) + out["indicator_name"] = out["indicator_name"].astype(str) + out["indicator_name_id"] = out["indicator_name_id"].astype(str) + out["unit"] = out["unit"].fillna("").astype(str) + out["direction"] = out["direction"].astype(str) + out["pillar_id"] = out["pillar_id"].astype(int) + out["pillar_name"] = out["pillar_name"].astype(str) + out["pillar_name_id"] = out["pillar_name_id"].astype(str) + out["framework"] = out["framework"].astype(str) + out["value"] = pd.to_numeric(out["value"], errors="coerce").astype(float) + out["norm_score_1_100"] = pd.to_numeric(out["norm_score_1_100"], errors="coerce").astype(float) + out["performance"] = out["performance"].astype(str).replace("nan", pd.NA).astype("string") + out["yoy_value"] = pd.to_numeric(out["yoy_value"], errors="coerce").astype(float) + out["rank_in_pillar_year"] = pd.to_numeric(out["rank_in_pillar_year"], errors="coerce").astype("Int64") + out["n_indicators_in_pillar_year"] = pd.to_numeric(out["n_indicators_in_pillar_year"], errors="coerce").astype("Int64") + out["country_best"] = out["country_best"].astype(str).replace("nan", pd.NA).astype("string") + out["country_worst"] = out["country_worst"].astype(str).replace("nan", pd.NA).astype("string") + out["country_best_id"] = out["country_best_id"].astype(str).replace("nan", pd.NA).astype("string") + out["country_worst_id"] = out["country_worst_id"].astype(str).replace("nan", pd.NA).astype("string") + out["narrative_en"] = out["narrative_en"].astype(str) + out["narrative_id"] = out["narrative_id"].astype(str) schema = [ - bigquery.SchemaField("indicator_id", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("indicator_name", "STRING", mode="REQUIRED"), - bigquery.SchemaField("indicator_name_id", "STRING", mode="NULLABLE"), - bigquery.SchemaField("unit", "STRING", mode="NULLABLE"), - bigquery.SchemaField("direction", "STRING", mode="REQUIRED"), - bigquery.SchemaField("pillar_name", "STRING", mode="REQUIRED"), - bigquery.SchemaField("pillar_name_id", "STRING", mode="NULLABLE"), - bigquery.SchemaField("framework", "STRING", mode="REQUIRED"), - bigquery.SchemaField("year_min", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("year_max", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("n_countries", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("avg_value_first", "FLOAT", mode="NULLABLE"), - bigquery.SchemaField("avg_value_last", "FLOAT", mode="NULLABLE"), - bigquery.SchemaField("asean_avg_value_first", "FLOAT", mode="NULLABLE"), - bigquery.SchemaField("asean_avg_value_last", "FLOAT", mode="NULLABLE"), - bigquery.SchemaField("avg_norm_score_1_100", "FLOAT", mode="NULLABLE"), - bigquery.SchemaField("performance", "STRING", mode="NULLABLE"), - bigquery.SchemaField("n_yoy_total", "INTEGER", mode="NULLABLE"), - bigquery.SchemaField("n_yoy_positive", "INTEGER", mode="NULLABLE"), - bigquery.SchemaField("best_yoy_from", "INTEGER", mode="NULLABLE"), - bigquery.SchemaField("best_yoy_to", "INTEGER", mode="NULLABLE"), - bigquery.SchemaField("country_worst", "STRING", mode="NULLABLE"), - bigquery.SchemaField("country_best", "STRING", mode="NULLABLE"), - bigquery.SchemaField("country_worst_id", "STRING", mode="NULLABLE"), - bigquery.SchemaField("country_best_id", "STRING", mode="NULLABLE"), - bigquery.SchemaField("narrative_en", "STRING", mode="NULLABLE"), - bigquery.SchemaField("narrative_id", "STRING", mode="NULLABLE"), + bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"), + bigquery.SchemaField("indicator_id", "INTEGER", mode="REQUIRED"), + bigquery.SchemaField("indicator_name", "STRING", mode="REQUIRED"), + bigquery.SchemaField("indicator_name_id", "STRING", mode="NULLABLE"), + bigquery.SchemaField("unit", "STRING", mode="NULLABLE"), + bigquery.SchemaField("direction", "STRING", mode="REQUIRED"), + bigquery.SchemaField("pillar_id", "INTEGER", mode="REQUIRED"), + bigquery.SchemaField("pillar_name", "STRING", mode="REQUIRED"), + bigquery.SchemaField("pillar_name_id", "STRING", mode="NULLABLE"), + bigquery.SchemaField("framework", "STRING", mode="REQUIRED"), + bigquery.SchemaField("value", "FLOAT", mode="NULLABLE"), + bigquery.SchemaField("norm_score_1_100", "FLOAT", mode="NULLABLE"), + bigquery.SchemaField("performance", "STRING", mode="NULLABLE"), + bigquery.SchemaField("yoy_value", "FLOAT", mode="NULLABLE"), + bigquery.SchemaField("rank_in_pillar_year", "INTEGER", mode="NULLABLE"), + bigquery.SchemaField("n_indicators_in_pillar_year", "INTEGER", mode="NULLABLE"), + bigquery.SchemaField("country_best", "STRING", mode="NULLABLE"), + bigquery.SchemaField("country_worst", "STRING", mode="NULLABLE"), + bigquery.SchemaField("country_best_id", "STRING", mode="NULLABLE"), + bigquery.SchemaField("country_worst_id", "STRING", mode="NULLABLE"), + bigquery.SchemaField("narrative_en", "STRING", mode="NULLABLE"), + bigquery.SchemaField("narrative_id", "STRING", mode="NULLABLE"), ] rows_loaded = load_to_bigquery( @@ -1234,15 +1107,16 @@ class IndicatorNormAggregator: "rows_loaded" : rows_loaded, "completeness_pct" : 100.0, "config_snapshot" : json.dumps({ - "granularity" : "indicator_id only", + "granularity" : "indicator_id x year (1 pillar, 1 tahun)", "narrative_style" : "interpretive, plain text, bilingual EN/ID", - "asean_columns" : ["asean_avg_value_first", "asean_avg_value_last"], - "architecture" : "ASEAN rows included in agg_indicator_norm", + "rank_basis" : "norm_score_1_100 within same pillar_id and year", + "architecture" : "Built from ASEAN rows (country_id=0) in agg_indicator_norm", "pillar_change" : "Renamed to Food Other; all pillars use 'Food ' prefix", }), "validation_metrics" : json.dumps({ "total_rows" : rows_loaded, "n_indicators": int(out["indicator_id"].nunique()), + "n_years" : int(out["year"].nunique()), }), } save_etl_metadata(self.client, metadata) @@ -1260,6 +1134,7 @@ class IndicatorNormAggregator: self.logger.info("INDICATOR NORM AGGREGATION") self.logger.info(" ASEAN rows ditambahkan ke agg_indicator_norm (country_id=0)") self.logger.info(" Rename Food Other; all pillars use Food prefix") + self.logger.info(" agg_narrative_indicator: granularity per indicator per year (1 pillar, 1 tahun)") self.logger.info("=" * 80) self.load_data() diff --git a/scripts/bigquery_aggregate_layer.py b/scripts/bigquery_aggregate_layer.py index d003792..a53462e 100644 --- a/scripts/bigquery_aggregate_layer.py +++ b/scripts/bigquery_aggregate_layer.py @@ -25,6 +25,11 @@ KONDISI PILAR (pillar_condition_en / pillar_condition_id): Kolom tambahan di agg_pillar_by_country untuk mendeskripsikan kondisi tiap pilar per negara per tahun secara kontekstual dan kuantitatif. +KONDISI FRAMEWORK (framework_condition_en / framework_condition_id): + Kolom tambahan di agg_framework_by_country untuk mendeskripsikan kondisi + tiap framework (MDGs/SDGs/Total) per negara per tahun, mengikuti tier + dan referensi yang sama dengan kondisi pilar. + Landasan teori: 1. FAO & CFS (1996 World Food Summit; CFS Reform Document 2009): Definisi 4 pilar ketahanan pangan dan makna substantif masing-masing. @@ -997,7 +1002,10 @@ class FoodSecurityAggregator: "asean_country_id" : ASEAN_COUNTRY_ID, "pillar_change" : "Sustainability renamed to Food Other, all pillars prefixed with Food", "architecture" : "ASEAN merged into country tables (country_id=0)", - "condition_column" : "pillar_condition_en/id added to agg_pillar_by_country", + "condition_column" : ( + "pillar_condition_en/id added to agg_pillar_by_country; " + "framework_condition_en/id added to agg_framework_by_country" + ), "condition_reference" : ( "GFSI 2022 (Economist Impact) score tiers >= 75/60/40/20; " "IPC Technical Manual 2019; FAO/CFS 4-pillar framework 1996/2009; " @@ -1187,7 +1195,7 @@ class FoodSecurityAggregator: return df # ========================================================================= - # STEP 3: agg_framework_by_country (termasuk ASEAN) + # STEP 3: agg_framework_by_country (termasuk ASEAN + kolom kondisi) # ========================================================================= def calc_framework_by_country(self) -> pd.DataFrame: @@ -1196,6 +1204,7 @@ class FoodSecurityAggregator: self.logger.info("\n" + "=" * 70) self.logger.info(f"STEP 3: {table_name} -> [Gold] fs_asean_gold") self.logger.info(" Termasuk baris ASEAN (country_id=0)") + self.logger.info(" Kolom baru: framework_condition_en, framework_condition_id") self.logger.info("=" * 70) try: @@ -1313,6 +1322,31 @@ class FoodSecurityAggregator: df = check_and_dedup(df, ["country_id", "framework", "year"], context=table_name, logger=self.logger) + # ----------------------------------------------------------------- + # TAMBAHAN: kolom kondisi framework (tier + deskripsi dipisah) + # Dibangkitkan SETELAH framework_score_1_100 tersedia dan SETELAH + # dedup, mengikuti pola yang sama dengan pillar_condition di + # agg_pillar_by_country. Referensi tier: GFSI 2022 (Economist + # Impact); IPC 2019; FAO/CFS 1996/2009; FAO SOFI 2024. + # + # Kolom yang dihasilkan: + # framework_condition_en — label tier EN, e.g. "At Risk" + # framework_condition_id — label tier ID, e.g. "Berisiko" + # framework_condition_desc_en — deskripsi kontekstual EN + # framework_condition_desc_id — deskripsi kontekstual ID + # ----------------------------------------------------------------- + fw_conditions = df.apply( + lambda row: get_framework_condition( + row["framework"], + row["framework_score_1_100"] + ), + axis=1 + ) + df["framework_condition_en"] = fw_conditions.apply(lambda x: x[0]) + df["framework_condition_id"] = fw_conditions.apply(lambda x: x[1]) + df["framework_condition_desc_en"] = fw_conditions.apply(lambda x: x[2]) + df["framework_condition_desc_id"] = fw_conditions.apply(lambda x: x[3]) + country_mask = df["country_id"] != ASEAN_COUNTRY_ID df.loc[country_mask, "rank_in_framework_year"] = ( df[country_mask] @@ -1331,6 +1365,18 @@ class FoodSecurityAggregator: df["framework_norm"] = df["framework_norm"].astype(float) df["framework_score_1_100"] = df["framework_score_1_100"].astype(float) df["country_name_id"] = df["country_name_id"].astype(str) + df["framework_condition_en"] = df["framework_condition_en"].astype(str) + df["framework_condition_id"] = df["framework_condition_id"].astype(str) + df["framework_condition_desc_en"] = df["framework_condition_desc_en"].astype(str) + df["framework_condition_desc_id"] = df["framework_condition_desc_id"].astype(str) + + self.logger.info(f"\n Total rows: {len(df):,}") + + # Log distribusi kondisi untuk QA + self.logger.info("\n Distribusi framework_condition_en (sample):") + fw_cond_dist = df["framework_condition_en"].value_counts().head(10) + for cond, cnt in fw_cond_dist.items(): + self.logger.info(f" {cnt:>6,} {cond}") schema = [ bigquery.SchemaField("country_id", "INTEGER", mode="REQUIRED"), @@ -1343,6 +1389,13 @@ class FoodSecurityAggregator: bigquery.SchemaField("framework_score_1_100", "FLOAT", mode="REQUIRED"), bigquery.SchemaField("rank_in_framework_year", "INTEGER", mode="REQUIRED"), bigquery.SchemaField("year_over_year_change", "FLOAT", mode="NULLABLE"), + # --- KOLOM KONDISI (sama pola dengan agg_pillar_by_country) --- + # Tier label (GFSI 2022): Secure / Adequate / Moderate / At Risk / Critical + bigquery.SchemaField("framework_condition_en", "STRING", mode="REQUIRED"), + bigquery.SchemaField("framework_condition_id", "STRING", mode="REQUIRED"), + # Deskripsi kontekstual per framework (MDGs/SDGs/Total) + bigquery.SchemaField("framework_condition_desc_en", "STRING", mode="REQUIRED"), + bigquery.SchemaField("framework_condition_desc_id", "STRING", mode="REQUIRED"), ] rows = load_to_bigquery( self.client, df, table_name, layer='gold', @@ -1600,7 +1653,7 @@ class FoodSecurityAggregator: self.logger.info("\n" + "=" * 70) self.logger.info("FOOD SECURITY AGGREGATION — 3 TABLES -> fs_asean_gold") self.logger.info(" ASEAN aggregate DIGABUNG ke tabel yang sama (country_id=0)") - self.logger.info(" Kolom baru : pillar_condition_en, pillar_condition_id") + self.logger.info(" Kolom baru : pillar_condition_en/id, framework_condition_en/id") self.logger.info(f" Performance threshold: {PERFORMANCE_THRESHOLD}") self.logger.info(f" Condition tiers (GFSI 2022): >=75 Secure | >=60 Adequate |") self.logger.info(f" >=40 Moderate | >=20 At Risk | <20 Critical")