From ebc189f2a046b1fe6577386bb10c138a8666014d Mon Sep 17 00:00:00 2001 From: Debby Date: Sat, 27 Jun 2026 13:23:02 +0700 Subject: [PATCH] rename pillar --- .../bigquery_aggraget_fact_selected_layer.py | 689 ++++------ scripts/bigquery_aggregate_layer.py | 1194 +++++------------ scripts/bigquery_analytical_layer.py | 67 +- scripts/bigquery_dimensional_model.py | 313 +++-- 4 files changed, 807 insertions(+), 1456 deletions(-) diff --git a/scripts/bigquery_aggraget_fact_selected_layer.py b/scripts/bigquery_aggraget_fact_selected_layer.py index c258a48..4db7a9e 100644 --- a/scripts/bigquery_aggraget_fact_selected_layer.py +++ b/scripts/bigquery_aggraget_fact_selected_layer.py @@ -1,89 +1,18 @@ """ BIGQUERY ANALYSIS LAYER - INDICATOR NORM AGGREGATION -Tabel 1: agg_indicator_norm -> fs_asean_gold -Tabel 2: agg_narrative_indicator -> fs_asean_gold -============================================================================= -BUGFIX: - - INDICATOR_NAME_ID_MAP: semua key diubah ke lowercase agar cocok dengan - lookup menggunakan .lower().strip(). Sebelumnya key Title Case tidak pernah - match karena fungsi get_indicator_name_id() melakukan .lower() sebelum lookup. -============================================================================= -PERUBAHAN: - - Ditambahkan kolom country_name_id : nama negara dalam Bahasa Indonesia [BARU] - - Ditambahkan kolom indicator_name_id : nama indikator dalam Bahasa Indonesia - - Ditambahkan kolom pillar_name_id : nama pilar dalam Bahasa Indonesia - - Ketiga kolom ikut tersimpan di BigQuery (schema + DataFrame output) - - Narrative versi Indonesia menggunakan nama negara & pilar dalam Bahasa Indonesia - - FIXED: "Access" -> "Akses" (konsisten di semua mapping pilar) -============================================================================= +PERUBAHAN ARSITEKTUR: + - ASEAN aggregate DIGABUNG ke dalam tabel yang sama (country_id=0, country_name="ASEAN") + sehingga Looker Studio dapat memfilter: per negara, atau ASEAN saja. + - agg_narrative_indicator: granularity tetap per indicator_id (all years, all countries), + ASEAN summary ditambahkan sebagai kolom terpisah (asean_avg_value_first/last). -agg_indicator_norm -============================================================================= -Tujuan: - Menghitung norm_value per indikator per negara per tahun, sehingga dapat - melihat performa setiap indikator secara individual (lower_better & higher_better - sudah dibalik). +Output 2 tabel: + 1. agg_indicator_norm -> per baris (year x country x indicator), termasuk ASEAN rows + 2. agg_narrative_indicator -> per indicator_id, ada kolom asean_avg_* tambahan -Framework Classification Logic: - - Semua indikator berlabel "MDGs" secara default. - - Indikator yang ada dalam SDG_ONLY_KEYWORDS akan berlabel "SDGs" mulai dari - sdgs_start_year (tahun pertama FIES hadir, dihitung otomatis). - - Indikator yang SUDAH ADA sebelum sdgs_start_year DAN juga termasuk - SDG_ONLY_KEYWORDS akan memiliki DUA label framework: - * "MDGs" untuk year < sdgs_start_year - * "SDGs" untuk year >= sdgs_start_year - - Indikator yang TIDAK ada dalam SDG_ONLY_KEYWORDS selalu "MDGs". - -YoY Logic: - - yoy_value : selisih absolut value vs tahun sebelumnya (per indikator, negara) - - yoy_norm_value : selisih absolut norm_value vs tahun sebelumnya - -Performance Label Logic: - - performance : "Good" jika norm_score_1_100 >= 60, "Bad" jika < 60, null jika null - -Output Schema (agg_indicator_norm): - year, country_id, country_name, country_name_id, - indicator_id, indicator_name, indicator_name_id, - unit, direction, - pillar_id, pillar_name, pillar_name_id, - framework, - value, - norm_value, - norm_score_1_100, - yoy_value, - yoy_norm_value, - performance - -============================================================================= -agg_narrative_indicator -============================================================================= -Tujuan: - Menghasilkan narasi otomatis per indikator (granularity: indicator_id). - Narasi membaca kondisi nyata dari data: tren, gap, anomali, konsistensi. - Tersedia dalam dua bahasa: Inggris (narrative_en) dan Indonesia (narrative_id). - - narrative_en : menggunakan nama negara & pilar dalam Bahasa Inggris - - narrative_id : menggunakan nama negara & pilar dalam Bahasa Indonesia - Tanpa markdown bold (**) agar aman ditampilkan di Looker Studio. - -Granularity: - indicator_id (all years, all ASEAN countries) - -Output Schema (agg_narrative_indicator): - indicator_id, indicator_name, indicator_name_id, - unit, direction, - pillar_name, pillar_name_id, - framework, - year_min, year_max, n_countries, - avg_value_first, 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, - narrative_en, - narrative_id +BUGFIX (diteruskan dari versi sebelumnya): + - INDICATOR_NAME_ID_MAP: semua key lowercase agar match dengan .lower().strip() lookup. """ import pandas as pd @@ -104,10 +33,19 @@ from google.cloud import bigquery # ============================================================================= -# MAPPING BAHASA INDONESIA +# KONSTANTA +# ============================================================================= + +ASEAN_COUNTRY_ID = 0 +ASEAN_COUNTRY_NAME = "ASEAN" +ASEAN_COUNTRY_NAME_ID = "ASEAN" + + +# ============================================================================= +# MAPPING BAHASA INDONESIA +# CHANGED: Other / Lainnya # ============================================================================= -# Mapping nama negara (Inggris -> Indonesia) COUNTRY_NAME_ID_MAP: dict = { "Brunei Darussalam" : "Brunei Darussalam", "Cambodia" : "Kamboja", @@ -122,40 +60,41 @@ COUNTRY_NAME_ID_MAP: dict = { "Timor-Leste" : "Timor-Leste", "Viet Nam" : "Vietnam", "Vietnam" : "Vietnam", + "ASEAN" : "ASEAN", } -# Mapping nama pilar (Inggris -> Indonesia) -# FIXED: "Access" -> "Akses" (bukan "Keterjangkauan") PILLAR_NAME_ID_MAP: dict = { - "Availability" : "Ketersediaan", - "Access" : "Akses", - "Utilization" : "Pemanfaatan", - "Stability" : "Stabilitas", - "Sustainability": "Keberlanjutan", - "availability" : "Ketersediaan", - "access" : "Akses", - "utilization" : "Pemanfaatan", - "stability" : "Stabilitas", - "sustainability": "Keberlanjutan", + # Mapping nama pilar (Inggris dengan prefix Food) -> Bahasa Indonesia + "Food Availability" : "Ketersediaan Pangan", + "Food Access" : "Akses Pangan", + "Food Utilization" : "Pemanfaatan Pangan", + "Food Stability" : "Stabilitas Pangan", + "Food Other" : "Indikator Tambahan", + # Variasi tanpa prefix Food + "Availability" : "Ketersediaan Pangan", + "Access" : "Akses Pangan", + "Utilization" : "Pemanfaatan Pangan", + "Stability" : "Stabilitas Pangan", + "Other" : "Indikator Tambahan", + # lowercase + "food availability" : "Ketersediaan Pangan", + "food access" : "Akses Pangan", + "food utilization" : "Pemanfaatan Pangan", + "food stability" : "Stabilitas Pangan", + "food other" : "Indikator Tambahan", + "availability" : "Ketersediaan Pangan", + "access" : "Akses Pangan", + "utilization" : "Pemanfaatan Pangan", + "stability" : "Stabilitas Pangan", + "other" : "Indikator Tambahan", } -# ============================================================================= -# BUGFIX: Semua key INDICATOR_NAME_ID_MAP ditulis lowercase -# agar cocok dengan lookup: str(indicator_name).lower().strip() -# Sebelumnya key Title Case tidak pernah match -> kolom selalu fallback ke EN. -# ============================================================================= +# BUGFIX: semua key lowercase INDICATOR_NAME_ID_MAP: dict = { - # ------------------------------------------------------------------------- - # DIETARY ENERGY SUPPLY - # ------------------------------------------------------------------------- "dietary energy supply used in the estimation of the prevalence of undernourishment (kcal/cap/day)": "Pasokan energi makanan yang digunakan dalam estimasi prevalensi kekurangan gizi (kkal/kapita/hari)", "dietary energy supply used in the estimation of the prevalence of undernourishment (kcal/cap/day) (3-year average)": "Pasokan energi makanan yang digunakan dalam estimasi prevalensi kekurangan gizi (kkal/kapita/hari) (rata-rata 3 tahun)", - - # ------------------------------------------------------------------------- - # WATER & SANITATION - # ------------------------------------------------------------------------- "percentage of population using at least basic drinking water services (percent)": "Persentase penduduk yang menggunakan layanan air minum dasar (persen)", "percentage of population using at least basic sanitation services (percent)": @@ -164,16 +103,8 @@ INDICATOR_NAME_ID_MAP: dict = { "Persentase penduduk yang menggunakan layanan air minum yang dikelola dengan aman (persen)", "percentage of population using safely managed sanitation services (percent)": "Persentase penduduk yang menggunakan layanan sanitasi yang dikelola dengan aman (persen)", - - # ------------------------------------------------------------------------- - # INFRASTRUCTURE - # ------------------------------------------------------------------------- "rail lines density (total route in km per 100 square km of land area)": "Kepadatan jalur kereta api (total rute dalam km per 100 km² lahan)", - - # ------------------------------------------------------------------------- - # AVAILABILITY - # ------------------------------------------------------------------------- "average dietary energy requirement (kcal/cap/day)": "Rata-rata kebutuhan energi makanan (kkal/kapita/hari)", "average dietary energy supply adequacy (percent) (3-year average)": @@ -194,10 +125,6 @@ INDICATOR_NAME_ID_MAP: dict = { "Variabilitas pasokan pangan per kapita (kkal/kapita/hari)", "value of food imports in total merchandise exports (percent) (3-year average)": "Nilai impor pangan terhadap total ekspor barang (persen) (rata-rata 3 tahun)", - - # ------------------------------------------------------------------------- - # ACCESS - # ------------------------------------------------------------------------- "gross domestic product per capita, ppp, (constant 2021 international $)": "Produk domestik bruto per kapita, PPP (internasional konstan 2021 US$)", "political stability and absence of violence/terrorism (index)": @@ -208,10 +135,6 @@ INDICATOR_NAME_ID_MAP: dict = { "Jumlah penduduk kekurangan gizi (juta jiwa) (rata-rata 3 tahun)", "minimum dietary energy requirement (kcal/cap/day)": "Kebutuhan energi makanan minimum (kkal/kapita/hari)", - - # ------------------------------------------------------------------------- - # UTILIZATION - # ------------------------------------------------------------------------- "prevalence of exclusive breastfeeding among infants 0-5 months of age (percent)": "Prevalensi pemberian ASI eksklusif pada bayi usia 0-5 bulan (persen)", "number of children under 5 years affected by wasting (million)": @@ -270,31 +193,15 @@ INDICATOR_NAME_ID_MAP: dict = { def get_country_name_id(country_name: str) -> str: - """Kembalikan terjemahan Bahasa Indonesia untuk nama negara.""" - return COUNTRY_NAME_ID_MAP.get( - str(country_name).strip(), - str(country_name), # fallback: kembalikan nama asli - ) + return COUNTRY_NAME_ID_MAP.get(str(country_name).strip(), str(country_name)) def get_indicator_name_id(indicator_name: str) -> str: - """ - Kembalikan terjemahan Bahasa Indonesia untuk nama indikator. - BUGFIX: key di INDICATOR_NAME_ID_MAP sudah lowercase sehingga - .lower().strip() lookup akan selalu match dengan benar. - """ - return INDICATOR_NAME_ID_MAP.get( - str(indicator_name).lower().strip(), - str(indicator_name), # fallback: kembalikan nama asli - ) + return INDICATOR_NAME_ID_MAP.get(str(indicator_name).lower().strip(), str(indicator_name)) def get_pillar_name_id(pillar_name: str) -> str: - """Kembalikan terjemahan Bahasa Indonesia untuk nama pilar.""" - return PILLAR_NAME_ID_MAP.get( - str(pillar_name).strip(), - str(pillar_name), # fallback: kembalikan nama asli - ) + return PILLAR_NAME_ID_MAP.get(str(pillar_name).strip(), str(pillar_name)) # ============================================================================= @@ -415,7 +322,6 @@ def _detect_trend(scores_by_year: pd.Series, lower_better: bool) -> str: x = np.arange(len(vals)) slope = np.polyfit(x, vals, 1)[0] - improving = (slope > 0 and not lower_better) or (slope < 0 and lower_better) mid = len(vals) // 2 @@ -425,7 +331,6 @@ def _detect_trend(scores_by_year: pd.Series, lower_better: bool) -> str: slope2 = np.polyfit(np.arange(len(second_half)), second_half, 1)[0] if len(second_half) > 1 else 0 cv = np.std(vals) / (np.mean(vals) + 1e-9) - if cv > 0.25: return "fluctuating" @@ -440,8 +345,10 @@ def _detect_trend(scores_by_year: pd.Series, lower_better: bool) -> str: def _detect_gap_trend(df_ind: pd.DataFrame, lower_better: bool) -> str: + # Hanya hitung gap di antara negara asli (bukan ASEAN) + df_real = df_ind[df_ind["country_id"] != ASEAN_COUNTRY_ID] std_by_year = ( - df_ind.groupby("year")["value"] + df_real.groupby("year")["value"] .std() .dropna() ) @@ -487,13 +394,10 @@ def _detect_anomaly_year(scores_by_year: pd.Series) -> tuple: def _detect_consistency(df_ind: pd.DataFrame, lower_better: bool) -> tuple: - """ - Mengembalikan (best_country_en, worst_country_en, is_consistent). - Nama negara dikembalikan dalam Bahasa Inggris; penerjemahan dilakukan - di layer narrative builder. - """ + """Hanya negara asli (bukan ASEAN) yang di-analisa konsistensinya.""" + df_real = df_ind[df_ind["country_id"] != ASEAN_COUNTRY_ID] country_avg = ( - df_ind.groupby("country_name")["value"] + df_real.groupby("country_name")["value"] .mean() .dropna() ) @@ -507,8 +411,8 @@ def _detect_consistency(df_ind: pd.DataFrame, lower_better: bool) -> tuple: best = country_avg.idxmax() worst = country_avg.idxmin() - asean_avg_by_year = df_ind.groupby("year")["value"].mean() - country_by_year = df_ind[df_ind["country_name"] == best].set_index("year")["value"] + asean_avg_by_year = df_real.groupby("year")["value"].mean() + country_by_year = df_real[df_real["country_name"] == best].set_index("year")["value"] years_both = set(asean_avg_by_year.index) & set(country_by_year.index) if not years_both: @@ -531,8 +435,7 @@ def _detect_consistency(df_ind: pd.DataFrame, lower_better: bool) -> tuple: # ============================================================================= -# NARRATIVE BUILDER — plain text, no markdown, bilingual -# FIXED: narrative_id menggunakan nama negara & pilar dalam Bahasa Indonesia +# NARRATIVE BUILDER # ============================================================================= def _build_narrative_per_indicator(row: pd.Series, df_full: pd.DataFrame) -> tuple: @@ -542,13 +445,17 @@ def _build_narrative_per_indicator(row: pd.Series, df_full: pd.DataFrame) -> tup 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) # nama pilar dalam Bahasa Indonesia + 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) - df_ind = df_full[df_full["indicator_id"] == ind_id].copy() + # 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() if df_ind.empty: na_en = f"{ind_name_en} ({framework}, {pillar_en}): Insufficient data for analysis." @@ -562,10 +469,8 @@ def _build_narrative_per_indicator(row: pd.Series, df_full: pd.DataFrame) -> tup 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 & worst_country -> nama dalam Bahasa Inggris (dari data) best_country_en, worst_country_en, is_consistent = _detect_consistency(df_ind, lower_better) - # Terjemahan nama negara ke Bahasa Indonesia 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 @@ -582,7 +487,6 @@ def _build_narrative_per_indicator(row: pd.Series, df_full: pd.DataFrame) -> tup sentences_en = [] sentences_id = [] - # Header: EN menggunakan nama Inggris, ID menggunakan nama Indonesia 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}):" sentences_en.append(s1_en) @@ -623,7 +527,6 @@ def _build_narrative_per_indicator(row: pd.Series, df_full: pd.DataFrame) -> tup 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.") - # Kalimat tentang negara: EN pakai nama Inggris, ID pakai nama Indonesia if best_country_en and worst_country_en: if is_consistent: sentences_en.append( @@ -646,7 +549,6 @@ def _build_narrative_per_indicator(row: pd.Series, df_full: pd.DataFrame) -> tup narrative_en = " ".join(s for s in sentences_en if s) narrative_id = " ".join(s for s in sentences_id if s) - return narrative_en, narrative_id @@ -655,11 +557,6 @@ def _build_narrative_per_indicator(row: pd.Series, df_full: pd.DataFrame) -> tup # ============================================================================= class IndicatorNormAggregator: - """ - Hitung norm_value per indikator untuk seluruh data di - fact_asean_food_security_selected, lalu simpan ke agg_indicator_norm. - Setelah selesai, otomatis menjalankan pipeline agg_narrative_indicator. - """ def __init__(self, client: bigquery.Client): self.client = client @@ -680,7 +577,7 @@ class IndicatorNormAggregator: } # ========================================================================= - # STEP 1: Load fact table + # STEP 1: Load data # ========================================================================= def load_data(self): @@ -707,6 +604,16 @@ class IndicatorNormAggregator: self.logger.warning(f" {n_null} rows direction NULL -> diisi 'positive'") self.df["direction"] = self.df["direction"].fillna("positive") + # Rename pillar_name: add 'Food ' prefix, remove + PILLAR_RENAME_MAP = { + 'Availability' : 'Food Availability', + 'Access' : 'Food Access', + 'Utilization' : 'Food Utilization', + 'Stability' : 'Food Stability', + 'Other' : 'Food Other', + } + self.df["pillar_name"] = self.df["pillar_name"].replace(PILLAR_RENAME_MAP) + self.pipeline_metadata["rows_fetched"] = len(self.df) self.logger.info(f" Rows : {len(self.df):,}") self.logger.info(f" Countries : {self.df['country_id'].nunique()}") @@ -716,7 +623,7 @@ class IndicatorNormAggregator: ) # ========================================================================= - # STEP 2: Load unit dari dim_indicator + # STEP 2: Load unit # ========================================================================= def load_units(self): @@ -740,136 +647,40 @@ class IndicatorNormAggregator: self.df_unit["indicator_id"] = self.df_unit["indicator_id"].astype(int) self.df_unit["unit"] = self.df_unit["unit"].fillna("").astype(str) - n_missing_unit = self.df_unit["unit"].eq("").sum() self.logger.info(f" dim_indicator rows (unique indicator_id): {len(self.df_unit):,}") - self.logger.info(f" Indicator dengan unit kosong : {n_missing_unit}") - - fact_ids = set(self.df["indicator_id"].astype(int).unique()) - dim_ids = set(self.df_unit["indicator_id"].unique()) - orphan = fact_ids - dim_ids - if orphan: - self.logger.warning( - f" [WARNING] {len(orphan)} indicator_id di fact tidak ditemukan di " - f"dim_indicator (unit akan diisi ''): {sorted(orphan)}" - ) # ========================================================================= - # STEP 3: Merge unit ke df + # STEP 3: Merge unit # ========================================================================= def _merge_unit(self): - self.logger.info("\n" + "=" * 80) - self.logger.info("STEP 3: MERGE UNIT -> fact df") - self.logger.info("=" * 80) - before = len(self.df) self.df = self.df.merge(self.df_unit, on="indicator_id", how="left") self.df["unit"] = self.df["unit"].fillna("").astype(str) after = len(self.df) - - assert before == after, ( - f"Row count berubah setelah merge unit: {before} -> {after}" - ) - - n_empty = self.df["unit"].eq("").sum() - self.logger.info( - f" Merge OK. Rows: {after:,} | Rows dengan unit kosong: {n_empty}" - ) + assert before == after, f"Row count berubah: {before} -> {after}" + self.logger.info(f" Merge unit OK. Rows: {after:,}") # ========================================================================= # STEP 3b: Tambah kolom nama Bahasa Indonesia # ========================================================================= def _add_indonesia_name_columns(self): - self.logger.info("\n" + "=" * 80) - self.logger.info("STEP 3b: ADD BAHASA INDONESIA NAME COLUMNS") - self.logger.info("=" * 80) + self.df["country_name_id"] = self.df["country_name"].apply(get_country_name_id).astype(str) + self.df["indicator_name_id"] = self.df["indicator_name"].apply(get_indicator_name_id).astype(str) + self.df["pillar_name_id"] = self.df["pillar_name"].apply(get_pillar_name_id).astype(str) - # Nama negara - self.df["country_name_id"] = ( - self.df["country_name"] - .apply(get_country_name_id) - .astype(str) - ) - - # Nama indikator — BUGFIX: key di map sudah lowercase, lookup .lower() match dengan benar - self.df["indicator_name_id"] = ( - self.df["indicator_name"] - .apply(get_indicator_name_id) - .astype(str) - ) - - # Nama pilar - self.df["pillar_name_id"] = ( - self.df["pillar_name"] - .apply(get_pillar_name_id) - .astype(str) - ) - - n_country_mapped = (self.df["country_name_id"] != self.df["country_name"]).sum() - n_indicator_mapped = (self.df["indicator_name_id"] != self.df["indicator_name"]).sum() - n_pillar_mapped = (self.df["pillar_name_id"] != self.df["pillar_name"]).sum() - self.logger.info(f" country_name_id mapped rows : {n_country_mapped:,}") - self.logger.info(f" indicator_name_id mapped rows : {n_indicator_mapped:,}") - self.logger.info(f" pillar_name_id mapped rows : {n_pillar_mapped:,}") - - # Validasi: peringatkan jika ada indikator yang tidak ter-mapping - unmapped_indicators = ( - self.df[self.df["indicator_name_id"] == self.df["indicator_name"]] - ["indicator_name"] - .drop_duplicates() - .tolist() - ) - if unmapped_indicators: - self.logger.warning( - f" [WARNING] {len(unmapped_indicators)} indicator_name tidak ditemukan " - f"di INDICATOR_NAME_ID_MAP (fallback ke nama Inggris):" - ) - for name in unmapped_indicators[:10]: - self.logger.warning(f" - {name}") - if len(unmapped_indicators) > 10: - self.logger.warning(f" ... dan {len(unmapped_indicators) - 10} lainnya") - - # Log sample negara - sample_ctr = ( - self.df[["country_name", "country_name_id"]] - .drop_duplicates() - .sort_values("country_name") - ) - self.logger.info("\n Terjemahan nama negara (EN -> ID):") - for _, r in sample_ctr.iterrows(): - self.logger.info(f" {r['country_name']:<35} -> {r['country_name_id']}") - - # Log sample pilar - sample_pil = ( - self.df[["pillar_name", "pillar_name_id"]] - .drop_duplicates() - ) - self.logger.info("\n Pillar mapping (EN -> ID):") + self.logger.info(" Kolom terjemahan Indonesia ditambahkan.") + sample_pil = self.df[["pillar_name", "pillar_name_id"]].drop_duplicates() + self.logger.info(" Pillar mapping:") for _, r in sample_pil.iterrows(): self.logger.info(f" {r['pillar_name']:<20} -> {r['pillar_name_id']}") - # Log sample indikator - sample_ind = ( - self.df[self.df["indicator_name_id"] != self.df["indicator_name"]] - [["indicator_name", "indicator_name_id"]] - .drop_duplicates() - .head(5) - ) - self.logger.info("\n Sample indicator mapping (EN -> ID) [ter-mapping]:") - for _, r in sample_ind.iterrows(): - self.logger.info(f" EN: {r['indicator_name'][:75]}") - self.logger.info(f" ID: {r['indicator_name_id'][:75]}") - # ========================================================================= # STEP 4: Deteksi sdgs_start_year # ========================================================================= def _detect_sdgs_start_year(self) -> int: - self.logger.info("\n" + "=" * 80) - self.logger.info("STEP 4: DETECT sdgs_start_year (first FIES year)") - self.logger.info("=" * 80) - fies_rows = self.df[ self.df["indicator_name"].str.lower().str.strip().isin(_FIES_DETECTION_LOWER) ] @@ -878,7 +689,6 @@ class IndicatorNormAggregator: self.logger.info(f" [Metode 1 - FIES explicit] sdgs_start_year = {sdgs_start}") return sdgs_start - self.logger.info(" [Metode 1] Tidak ada FIES rows -> fallback gap-terbesar") ind_min_year = ( self.df.groupby("indicator_id")["year"] .min().reset_index() @@ -888,7 +698,6 @@ class IndicatorNormAggregator: if len(unique_years) == 1: sdgs_start = int(unique_years[0]) + 9999 - self.logger.info(" Hanya 1 cluster -> semua MDGs") else: gaps = [ (unique_years[i+1] - unique_years[i], unique_years[i], unique_years[i+1]) @@ -897,9 +706,7 @@ class IndicatorNormAggregator: gaps.sort(reverse=True) _, y_before, y_after = gaps[0] sdgs_start = int(y_after) - self.logger.info( - f" Gap terbesar: {y_before} -> {y_after} -> sdgs_start_year = {sdgs_start}" - ) + self.logger.info(f" Gap terbesar: {y_before} -> {y_after} -> sdgs_start_year = {sdgs_start}") return sdgs_start @@ -908,11 +715,6 @@ class IndicatorNormAggregator: # ========================================================================= def _assign_framework(self): - self.logger.info("\n" + "=" * 80) - self.logger.info("STEP 5: ASSIGN FRAMEWORK PER BARIS") - self.logger.info(f" sdgs_start_year = {self.sdgs_start_year}") - self.logger.info("=" * 80) - df = self.df.copy() df["_is_sdg_kw"] = df["indicator_name"].str.lower().str.strip().isin(_SDG_ONLY_LOWER) df["framework"] = "MDGs" @@ -920,23 +722,18 @@ class IndicatorNormAggregator: mask_sdgs = df["_is_sdg_kw"] & (df["year"] >= self.sdgs_start_year) df.loc[mask_sdgs, "framework"] = "SDGs" df = df.drop(columns=["_is_sdg_kw"]) + self.df = df - fw_dist = df["framework"].value_counts() + fw_dist = self.df["framework"].value_counts() self.logger.info("\n Framework distribution (rows):") for fw, cnt in fw_dist.items(): self.logger.info(f" {fw:<6}: {cnt:,} rows") - self.df = df - # ========================================================================= # STEP 6: Hitung norm_value per indikator # ========================================================================= def _compute_norm_values(self) -> pd.DataFrame: - self.logger.info("\n" + "=" * 80) - self.logger.info("STEP 6: COMPUTE NORM_VALUE PER INDICATOR (direction-aware)") - self.logger.info("=" * 80) - df = self.df.copy() norm_parts = [] @@ -953,9 +750,6 @@ class IndicatorNormAggregator: if n_valid < 2: grp["norm_value"] = np.nan norm_parts.append(grp) - self.logger.warning( - f" [SKIP] indicator_id={ind_id}: only {n_valid} valid values" - ) continue raw = grp.loc[valid_mask, "value"].values @@ -974,93 +768,100 @@ class IndicatorNormAggregator: grp["norm_value"] = normed norm_parts.append(grp) - df_normed = pd.concat(norm_parts, ignore_index=True) - self.logger.info(f" norm_value computed: {df_normed['indicator_id'].nunique()} indicators") - return df_normed + return pd.concat(norm_parts, ignore_index=True) + + # ========================================================================= + # STEP 6b: Tambah baris ASEAN (rata-rata dari semua negara per ind per year) + # ========================================================================= + + def _add_asean_rows(self, df_normed: pd.DataFrame) -> pd.DataFrame: + """ + Buat baris ASEAN aggregate: nilai = rata-rata nilai semua negara asli. + norm_value = rata-rata norm_value semua negara asli. + """ + real_rows = df_normed[df_normed["country_id"] != ASEAN_COUNTRY_ID] + + dim_cols = [ + "indicator_id", "indicator_name", "indicator_name_id", + "unit", "direction", + "pillar_id", "pillar_name", "pillar_name_id", + "framework", + ] + + asean_agg = ( + real_rows.groupby(["indicator_id", "year"]) + .agg( + value =("value", "mean"), + norm_value=("norm_value", "mean"), + ) + .reset_index() + ) + + # Gabung kolom dimensi dari baris pertama per indicator_id + dim_ref = ( + real_rows[dim_cols] + .drop_duplicates(subset=["indicator_id"]) + .copy() + ) + asean_agg = asean_agg.merge(dim_ref, on="indicator_id", how="left") + + asean_agg["country_id"] = ASEAN_COUNTRY_ID + asean_agg["country_name"] = ASEAN_COUNTRY_NAME + asean_agg["country_name_id"] = ASEAN_COUNTRY_NAME_ID + + # Hanya kolom yang ada di df_normed + cols_needed = [c for c in df_normed.columns if c in asean_agg.columns] + for c in df_normed.columns: + if c not in asean_agg.columns: + asean_agg[c] = np.nan + + return pd.concat([df_normed, asean_agg[df_normed.columns]], ignore_index=True) # ========================================================================= # STEP 7: Hitung YoY # ========================================================================= def _compute_yoy_columns(self, df: pd.DataFrame) -> pd.DataFrame: - self.logger.info("\n" + "=" * 80) - self.logger.info("STEP 7: COMPUTE YoY COLUMNS (per indicator, per country)") - self.logger.info("=" * 80) - parts = [] groups = df.groupby(["indicator_id", "country_id"], sort=False) - self.logger.info(f" Processing {groups.ngroups:,} (indicator x country) groups...") for (ind_id, country_id), grp in groups: parts.append(_compute_yoy(grp)) - df_out = pd.concat(parts, ignore_index=True) - self.logger.info(f" yoy_value nulls : {df_out['yoy_value'].isna().sum():,}") - self.logger.info(f" yoy_norm_value nulls: {df_out['yoy_norm_value'].isna().sum():,}") - return df_out + return pd.concat(parts, ignore_index=True) # ========================================================================= # STEP 8: Scale ke 1-100 # ========================================================================= def _compute_scores(self, df: pd.DataFrame) -> pd.DataFrame: - self.logger.info("\n" + "=" * 80) - self.logger.info("STEP 8: SCALE TO 1-100") - self.logger.info("=" * 80) - score_parts = [] for ind_id, grp in df.groupby("indicator_id"): grp = grp.copy() grp["norm_score_1_100"] = global_minmax(grp["norm_value"]) score_parts.append(grp) - df = pd.concat(score_parts, ignore_index=True) - - self.logger.info( - f" norm_score_1_100 range: " - f"{df['norm_score_1_100'].min():.2f} - {df['norm_score_1_100'].max():.2f}" - ) - return df + return pd.concat(score_parts, ignore_index=True) # ========================================================================= # STEP 9: Assign performance label # ========================================================================= def _assign_performance(self, df: pd.DataFrame) -> pd.DataFrame: - self.logger.info("\n" + "=" * 80) - self.logger.info( - f"STEP 9: ASSIGN PERFORMANCE LABEL " - f"(threshold >= {_PERFORMANCE_THRESHOLD} -> Good)" - ) - self.logger.info("=" * 80) - df = df.copy() df["performance"] = pd.NA has_score = df["norm_score_1_100"].notna() df.loc[has_score & (df["norm_score_1_100"] >= _PERFORMANCE_THRESHOLD), "performance"] = "Good" df.loc[has_score & (df["norm_score_1_100"] < _PERFORMANCE_THRESHOLD), "performance"] = "Bad" - - n_good = (df["performance"] == "Good").sum() - n_bad = (df["performance"] == "Bad").sum() - n_null = df["performance"].isna().sum() - total = len(df) - - self.logger.info(f" Good : {n_good:,} ({n_good/total*100:.1f}%)") - self.logger.info(f" Bad : {n_bad:,} ({n_bad/total*100:.1f}%)") - self.logger.info(f" Null : {n_null:,} ({n_null/total*100:.1f}%)") return df # ========================================================================= - # STEP 10: Save agg_indicator_norm + # STEP 10: Save agg_indicator_norm (termasuk ASEAN rows) # ========================================================================= def _save(self, df: pd.DataFrame) -> int: table_name = "agg_indicator_norm" - self.logger.info("\n" + "=" * 80) - self.logger.info(f"STEP 10: SAVE -> [Gold] {table_name}") - self.logger.info("=" * 80) - out = df[[ "year", "country_id", "country_name", "country_name_id", "indicator_id", "indicator_name", "indicator_name_id", @@ -1095,10 +896,9 @@ class IndicatorNormAggregator: out["yoy_norm_value"] = pd.to_numeric(out["yoy_norm_value"], errors="coerce").astype(float) out["performance"] = out["performance"].astype(str).replace("nan", pd.NA).astype("string") - self.logger.info(f" Total rows : {len(out):,}") - self.logger.info(f" Countries : {out['country_id'].nunique()}") - self.logger.info(f" Indicators : {out['indicator_id'].nunique()}") - self.logger.info(f" Years : {int(out['year'].min())} - {int(out['year'].max())}") + n_asean = (out["country_id"] == ASEAN_COUNTRY_ID).sum() + n_country = (out["country_id"] != ASEAN_COUNTRY_ID).sum() + self.logger.info(f" Total rows : {len(out):,} ({n_country:,} country + {n_asean:,} ASEAN)") schema = [ bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"), @@ -1141,75 +941,41 @@ class IndicatorNormAggregator: "completeness_pct" : 100.0, "config_snapshot" : json.dumps({ "sdgs_start_year" : self.sdgs_start_year, - "sdg_only_keywords_n" : len(SDG_ONLY_KEYWORDS), "layer" : "gold", "normalization" : "per_indicator_global_minmax", - "direction_handling" : "lower_better_inverted", - "yoy_columns" : ["yoy_value", "yoy_norm_value"], "performance_threshold": _PERFORMANCE_THRESHOLD, - "unit_source" : "dim_indicator", - "added_columns" : ["country_name_id", "indicator_name_id", "pillar_name_id"], - "bugfix" : "INDICATOR_NAME_ID_MAP keys lowercased to match .lower() lookup", + "asean_country_id" : ASEAN_COUNTRY_ID, + "architecture" : "ASEAN merged into country table (country_id=0)", + "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_countries" : int(out["country_id"].nunique()), - "sdgs_start_year": self.sdgs_start_year, + "n_countries" : int(out[out["country_id"] != ASEAN_COUNTRY_ID]["country_id"].nunique()), + "asean_rows" : int(n_asean), }), } save_etl_metadata(self.client, metadata) return rows_loaded # ========================================================================= - # STEP 11: Summary log - # ========================================================================= - - def _log_summary(self, df: pd.DataFrame): - self.logger.info("\n" + "=" * 80) - self.logger.info("STEP 11: SUMMARY — agg_indicator_norm") - self.logger.info("=" * 80) - - ind_avg = ( - df.groupby(["indicator_id", "indicator_name", "pillar_name", "direction"]) - ["norm_score_1_100"].mean() - .reset_index() - .sort_values("norm_score_1_100", ascending=False) - ) - - self.logger.info("\n TOP 5 Indicators (avg norm_score_1_100):") - for _, r in ind_avg.head(5).iterrows(): - tag = "[lower+]" if r["direction"] in DIRECTION_INVERT_KEYWORDS else "[higher+]" - self.logger.info( - f" [{int(r['indicator_id'])}] {r['indicator_name'][:50]:<52} " - f"{r['norm_score_1_100']:.2f} {tag}" - ) - - self.logger.info("\n BOTTOM 5 Indicators:") - for _, r in ind_avg.tail(5).iterrows(): - tag = "[lower+]" if r["direction"] in DIRECTION_INVERT_KEYWORDS else "[higher+]" - self.logger.info( - f" [{int(r['indicator_id'])}] {r['indicator_name'][:50]:<52} " - f"{r['norm_score_1_100']:.2f} {tag}" - ) - - # ========================================================================= - # STEP 12-17: agg_narrative_indicator + # STEP 11: agg_narrative_indicator (per indicator, ASEAN summary sebagai kolom) # ========================================================================= def _build_narrative_table(self, df_final: pd.DataFrame): self.logger.info("\n" + "=" * 80) - self.logger.info("STEP 12-17: agg_narrative_indicator") - self.logger.info(" Granularity: per indicator_id (all years + all ASEAN countries)") - self.logger.info(" Narrative : interpretatif, plain text, bilingual EN/ID") - self.logger.info(" FIXED : narrative_id pakai nama negara & pilar Bahasa Indonesia") + 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("=" * 80) - df = df_final.copy() + # 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] - # ---- Agregasi statistik per indikator ---- + # ---- Statistik per indikator (negara asli) ---- df_yr = ( - df.groupby(["indicator_id", "year"]) + df_real.groupby(["indicator_id", "year"]) .agg( avg_value =("value", "mean"), avg_norm_score =("norm_score_1_100", "mean"), @@ -1234,14 +1000,31 @@ class IndicatorNormAggregator: .reset_index() ) df_nc = ( - df.groupby("indicator_id")["country_id"] + df_real.groupby("indicator_id")["country_id"] .nunique().reset_index() .rename(columns={"country_id": "n_countries"}) ) - # YoY stats + # 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"}) + ) + + # YoY stats (negara asli) dir_map = ( - df[["indicator_id", "direction"]] + df_real[["indicator_id", "direction"]] .drop_duplicates(subset=["indicator_id"]) .set_index("indicator_id")["direction"] .to_dict() @@ -1291,9 +1074,9 @@ class IndicatorNormAggregator: }) df_yoy_stats = pd.DataFrame(yoy_stats) - # Country best/worst (nama asli Bahasa Inggris) + # Country best/worst df_country_avg = ( - df.groupby(["indicator_id", "country_id", "country_name"]) + df_real.groupby(["indicator_id", "country_id", "country_name"]) .agg(country_avg_value=("value", "mean")) .reset_index() ) @@ -1315,14 +1098,14 @@ class IndicatorNormAggregator: }) df_country_stats = pd.DataFrame(country_stats) - # Dim cols — sertakan kolom Indonesia + # Dim cols dim_cols = [ "indicator_name", "indicator_name_id", "unit", "direction", "pillar_name", "pillar_name_id", "framework", ] - df_dim = df[["indicator_id"] + dim_cols].drop_duplicates(subset=["indicator_id"]) + df_dim = df_real[["indicator_id"] + dim_cols].drop_duplicates(subset=["indicator_id"]) # Merge semua df_agg = ( @@ -1333,6 +1116,8 @@ class IndicatorNormAggregator: .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 @@ -1341,26 +1126,18 @@ class IndicatorNormAggregator: 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 ---- - self.logger.info("\n--- BUILD NARRATIVE (interpretatif, plain text, bilingual EN/ID) ---") + # Build narrative narratives_en = [] narratives_id = [] - for _, row in df_agg.iterrows(): - n_en, n_id = _build_narrative_per_indicator(row, df) + n_en, n_id = _build_narrative_per_indicator(row, df_final) narratives_en.append(n_en) narratives_id.append(n_id) df_agg["narrative_en"] = narratives_en df_agg["narrative_id"] = narratives_id - self.logger.info(f" Narratives generated: {len(df_agg):,}") - self.logger.info("\n Sample EN (first):") - self.logger.info(f" {df_agg.iloc[0]['narrative_en'][:300]}") - self.logger.info("\n Sample ID (first):") - self.logger.info(f" {df_agg.iloc[0]['narrative_id'][:300]}") - - # ---- Save ---- + # Output out = df_agg[[ "indicator_id", "indicator_name", "indicator_name_id", "unit", "direction", @@ -1368,6 +1145,7 @@ class IndicatorNormAggregator: "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", @@ -1389,9 +1167,11 @@ class IndicatorNormAggregator: 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["avg_norm_score_1_100"] = pd.to_numeric(out["avg_norm_score_1_100"], errors="coerce").astype(float) + 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") @@ -1405,31 +1185,33 @@ class IndicatorNormAggregator: 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("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("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"), ] rows_loaded = load_to_bigquery( @@ -1452,15 +1234,11 @@ class IndicatorNormAggregator: "rows_loaded" : rows_loaded, "completeness_pct" : 100.0, "config_snapshot" : json.dumps({ - "source_table" : "agg_indicator_norm (in-memory df_final)", - "granularity" : "indicator_id only (all years, all ASEAN countries)", - "narrative_style" : "interpretive, plain text, no markdown, bilingual EN/ID", - "narrative_dimensions" : ["trend", "gap_trend", "anomaly", "country_consistency"], - "performance_threshold": _PERFORMANCE_THRESHOLD, - "layer" : "gold", - "added_columns" : ["country_name_id", "indicator_name_id", "pillar_name_id", - "country_worst_id", "country_best_id"], - "bugfix" : "INDICATOR_NAME_ID_MAP keys lowercased to match .lower() lookup", + "granularity" : "indicator_id only", + "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", + "pillar_change" : "Renamed to Food Other; all pillars use 'Food ' prefix", }), "validation_metrics" : json.dumps({ "total_rows" : rows_loaded, @@ -1480,12 +1258,8 @@ class IndicatorNormAggregator: self.logger.info("\n" + "=" * 80) self.logger.info("INDICATOR NORM AGGREGATION") - self.logger.info(" Source : fact_asean_food_security_selected") - self.logger.info(" Dim : dim_indicator (unit)") - self.logger.info(" Output : agg_indicator_norm -> fs_asean_gold") - self.logger.info(" agg_narrative_indicator -> fs_asean_gold") - self.logger.info(" Added : country_name_id, indicator_name_id, pillar_name_id (Bahasa Indonesia)") - self.logger.info(" BUGFIX : INDICATOR_NAME_ID_MAP keys -> lowercase (cocok dengan .lower() lookup)") + 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("=" * 80) self.load_data() @@ -1494,13 +1268,13 @@ class IndicatorNormAggregator: self._add_indonesia_name_columns() self.sdgs_start_year = self._detect_sdgs_start_year() self._assign_framework() - df_normed = self._compute_norm_values() - df_yoy = self._compute_yoy_columns(df_normed) - df_scored = self._compute_scores(df_yoy) - df_final = self._assign_performance(df_scored) - rows_loaded = self._save(df_final) + df_normed = self._compute_norm_values() + df_with_asean = self._add_asean_rows(df_normed) # <-- ASEAN ditambahkan di sini + df_yoy = self._compute_yoy_columns(df_with_asean) + df_scored = self._compute_scores(df_yoy) + df_final = self._assign_performance(df_scored) + rows_loaded = self._save(df_final) self.pipeline_metadata["rows_loaded"] = rows_loaded - self._log_summary(df_final) self._build_narrative_table(df_final) self.pipeline_metadata["end_time"] = datetime.now() @@ -1543,6 +1317,7 @@ if __name__ == "__main__": print("=" * 80) print("INDICATOR NORM AGGREGATION -> fs_asean_gold") + print(f" ASEAN merged into country tables (country_id={ASEAN_COUNTRY_ID})") print("=" * 80) logger = setup_logging() diff --git a/scripts/bigquery_aggregate_layer.py b/scripts/bigquery_aggregate_layer.py index a9e1528..fd3d1c2 100644 --- a/scripts/bigquery_aggregate_layer.py +++ b/scripts/bigquery_aggregate_layer.py @@ -1,28 +1,25 @@ """ BIGQUERY ANALYSIS LAYER - FOOD SECURITY AGGREGATION -Semua agregasi pakai norm_value dari _get_norm_value_df() -UPDATED: Simpan 6 tabel ke fs_asean_gold (layer='gold'): - - agg_pillar_composite - - agg_pillar_by_country - - agg_framework_by_country - - agg_framework_asean (+ kolom performance_status: 'Good'/'Bad', threshold=60) - - agg_narrative_overview (bilingual: narrative_en, narrative_id) - - agg_narrative_pillar (bilingual: narrative_en, narrative_id) + +PERUBAHAN ARSITEKTUR: + - ASEAN aggregate DIGABUNG ke dalam tabel yang sama dengan negara-negara, + menggunakan country_name = "ASEAN" dan country_id = 0. + Looker Studio dapat memfilter: semua negara, per negara, atau ASEAN saja. + - 3 tabel DIHAPUS (digantikan oleh filter di tabel gabungan): + * agg_pillar_composite -> cukup filter country_name = "ASEAN" di agg_pillar_by_country + * agg_framework_asean -> cukup filter country_name = "ASEAN" di agg_framework_by_country + * agg_narrative_overview -> cukup filter country_name = "ASEAN" di agg_narrative_pillar + - "Sustainability" diganti "Other" di seluruh mapping pilar. + +Output 3 tabel ke fs_asean_gold: + 1. agg_pillar_by_country (termasuk baris ASEAN per pillar per year) + 2. agg_framework_by_country (termasuk baris ASEAN per framework per year) + 3. agg_narrative_pillar (termasuk baris ASEAN per pillar per year) Narrative style: - Plain text, tanpa markdown bold (**) - Interpretatif: membaca tren, gap, anomali, konsistensi dari data nyata - Bilingual: narrative_en (Inggris) + narrative_id (Indonesia) - * narrative_en : nama negara & pilar dalam Bahasa Inggris - * narrative_id : nama negara & pilar dalam Bahasa Indonesia [FIXED] - - Granularity: per tahun (Overview & Pillar) - -FIXED & ADDED: - - "Access" -> "Akses" di semua mapping pilar (bukan "Keterjangkauan") - - Tambah COUNTRY_NAME_ID_MAP untuk terjemahan nama 10 negara ASEAN - - narrative_id menggunakan nama negara & pilar dalam Bahasa Indonesia - - Tambah kolom country_name_id di tabel yang menyimpan country_name - - Kolom pillar_name_id di semua tabel pilar """ import pandas as pd @@ -57,6 +54,11 @@ DIRECTION_POSITIVE_KEYWORDS = frozenset({ NORMALIZE_FRAMEWORKS_JOINTLY = False PERFORMANCE_THRESHOLD = 60.0 +# country_id fiktif untuk baris ASEAN aggregate +ASEAN_COUNTRY_ID = 0 +ASEAN_COUNTRY_NAME = "ASEAN" +ASEAN_COUNTRY_NAME_ID = "ASEAN" # sama di kedua bahasa + SDG_ONLY_KEYWORDS: frozenset = frozenset([ "prevalence of undernourishment (percent) (3-year average)", "number of people undernourished (million) (3-year average)", @@ -93,9 +95,9 @@ _FIES_DETECTION_LOWER: frozenset = frozenset([ # ============================================================================= # TRANSLATION DICTIONARIES +# CHANGED: "Sustainability" / "sustainability" -> "Other" / "Lainnya" # ============================================================================= -# Nama negara ASEAN -> Bahasa Indonesia [BARU] COUNTRY_NAME_ID_MAP: dict = { "Brunei Darussalam" : "Brunei Darussalam", "Cambodia" : "Kamboja", @@ -110,173 +112,46 @@ COUNTRY_NAME_ID_MAP: dict = { "Timor-Leste" : "Timor-Leste", "Viet Nam" : "Vietnam", "Vietnam" : "Vietnam", + "ASEAN" : "ASEAN", } -# Nama pilar -> Bahasa Indonesia -# FIXED: "Access" -> "Akses" (bukan "Keterjangkauan") PILLAR_TRANSLATION_ID: dict = { - "Availability" : "Ketersediaan", - "Access" : "Akses", - "Utilization" : "Pemanfaatan", - "Stability" : "Stabilitas", - "Sustainability" : "Keberlanjutan", - "availability" : "Ketersediaan", - "access" : "Akses", - "utilization" : "Pemanfaatan", - "stability" : "Stabilitas", - "sustainability" : "Keberlanjutan", + # Mapping nama pilar (Inggris dengan prefix Food) -> Bahasa Indonesia "Food Availability" : "Ketersediaan Pangan", "Food Access" : "Akses Pangan", "Food Utilization" : "Pemanfaatan Pangan", "Food Stability" : "Stabilitas Pangan", - "Food Sustainability": "Keberlanjutan Pangan", + "Food Other" : "Indikator Tambahan", + # Variasi tanpa prefix Food + "Availability" : "Ketersediaan Pangan", + "Access" : "Akses Pangan", + "Utilization" : "Pemanfaatan Pangan", + "Stability" : "Stabilitas Pangan", + "Other" : "Indikator Tambahan", + # Legacy Sustainability + "Sustainability" : "Indikator Tambahan", + "sustainability" : "Indikator Tambahan", + # lowercase + "food availability" : "Ketersediaan Pangan", + "food access" : "Akses Pangan", + "food utilization" : "Pemanfaatan Pangan", + "food stability" : "Stabilitas Pangan", + "food other" : "Indikator Tambahan", + "availability" : "Ketersediaan Pangan", + "access" : "Akses Pangan", + "utilization" : "Pemanfaatan Pangan", + "stability" : "Stabilitas Pangan", + "other" : "Indikator Tambahan", } -INDICATOR_TRANSLATION_ID: dict = { - # ------------------------------------------------------------------------- - # DIETARY ENERGY SUPPLY - # ------------------------------------------------------------------------- - "Dietary energy supply used in the estimation of the prevalence of undernourishment (kcal/cap/day)": - "Pasokan energi makanan yang digunakan dalam estimasi prevalensi kekurangan gizi (kkal/kapita/hari)", - "Dietary energy supply used in the estimation of the prevalence of undernourishment (kcal/cap/day) (3-year average)": - "Pasokan energi makanan yang digunakan dalam estimasi prevalensi kekurangan gizi (kkal/kapita/hari) (rata-rata 3 tahun)", - - # ------------------------------------------------------------------------- - # WATER & SANITATION - # ------------------------------------------------------------------------- - "Percentage of population using at least basic drinking water services (percent)": - "Persentase penduduk yang menggunakan layanan air minum dasar (persen)", - "Percentage of population using at least basic sanitation services (percent)": - "Persentase penduduk yang menggunakan layanan sanitasi dasar (persen)", - "Percentage of population using safely managed drinking water services (percent)": - "Persentase penduduk yang menggunakan layanan air minum yang dikelola dengan aman (persen)", - "Percentage of population using safely managed sanitation services (percent)": - "Persentase penduduk yang menggunakan layanan sanitasi yang dikelola dengan aman (persen)", - - # ------------------------------------------------------------------------- - # INFRASTRUCTURE - # ------------------------------------------------------------------------- - "Rail lines density (total route in km per 100 square km of land area)": - "Kepadatan jalur kereta api (total rute dalam km per 100 km² lahan)", - - # ------------------------------------------------------------------------- - # AVAILABILITY - # ------------------------------------------------------------------------- - "Average dietary energy requirement (kcal/cap/day)": - "Rata-rata kebutuhan energi makanan (kkal/kapita/hari)", - "Average dietary energy supply adequacy (percent) (3-year average)": - "Kecukupan rata-rata pasokan energi makanan (persen) (rata-rata 3 tahun)", - "Average fat supply (g/cap/day) (3-year average)": - "Rata-rata pasokan lemak (g/kapita/hari) (rata-rata 3 tahun)", - "Average protein supply (g/cap/day) (3-year average)": - "Rata-rata pasokan protein (g/kapita/hari) (rata-rata 3 tahun)", - "Average supply of protein of animal origin (g/cap/day) (3-year average)": - "Rata-rata pasokan protein hewani (g/kapita/hari) (rata-rata 3 tahun)", - "Percent of arable land equipped for irrigation (percent) (3-year average)": - "Persentase lahan pertanian yang dilengkapi irigasi (persen) (rata-rata 3 tahun)", - "Cereal import dependency ratio (percent) (3-year average)": - "Rasio ketergantungan impor sereal (persen) (rata-rata 3 tahun)", - "Share of dietary energy supply derived from cereals, roots and tubers (percent) (3-year average)": - "Proporsi pasokan energi makanan dari serealia, akar, dan umbi-umbian (persen) (rata-rata 3 tahun)", - "Per capita food supply variability (kcal/cap/day)": - "Variabilitas pasokan pangan per kapita (kkal/kapita/hari)", - "Value of food imports in total merchandise exports (percent) (3-year average)": - "Nilai impor pangan terhadap total ekspor barang (persen) (rata-rata 3 tahun)", - - # ------------------------------------------------------------------------- - # ACCESS - # ------------------------------------------------------------------------- - "Gross domestic product per capita, PPP, (constant 2021 international $)": - "Produk domestik bruto per kapita, PPP (internasional konstan 2021 US$)", - "Political stability and absence of violence/terrorism (index)": - "Stabilitas politik dan tidak adanya kekerasan/terorisme (indeks)", - "Prevalence of undernourishment (percent) (3-year average)": - "Prevalensi kekurangan gizi (persen) (rata-rata 3 tahun)", - "Number of people undernourished (million) (3-year average)": - "Jumlah penduduk kekurangan gizi (juta jiwa) (rata-rata 3 tahun)", - "Minimum dietary energy requirement (kcal/cap/day)": - "Kebutuhan energi makanan minimum (kkal/kapita/hari)", - - # ------------------------------------------------------------------------- - # UTILIZATION - # ------------------------------------------------------------------------- - "Prevalence of exclusive breastfeeding among infants 0-5 months of age (percent)": - "Prevalensi pemberian ASI eksklusif pada bayi usia 0-5 bulan (persen)", - "Number of children under 5 years affected by wasting (million)": - "Jumlah anak di bawah 5 tahun yang mengalami wasting (juta jiwa)", - "Number of moderately or severely food insecure female adults (million) (3-year average)": - "Jumlah perempuan dewasa yang mengalami kerawanan pangan sedang atau berat (juta jiwa) (rata-rata 3 tahun)", - "Number of moderately or severely food insecure male adults (million) (3-year average)": - "Jumlah laki-laki dewasa yang mengalami kerawanan pangan sedang atau berat (juta jiwa) (rata-rata 3 tahun)", - "Number of moderately or severely food insecure people (million) (3-year average)": - "Jumlah penduduk yang mengalami kerawanan pangan sedang atau berat (juta jiwa) (rata-rata 3 tahun)", - "Number of severely food insecure female adults (million) (3-year average)": - "Jumlah perempuan dewasa yang mengalami kerawanan pangan berat (juta jiwa) (rata-rata 3 tahun)", - "Number of severely food insecure male adults (million) (3-year average)": - "Jumlah laki-laki dewasa yang mengalami kerawanan pangan berat (juta jiwa) (rata-rata 3 tahun)", - "Number of severely food insecure people (million) (3-year average)": - "Jumlah penduduk yang mengalami kerawanan pangan berat (juta jiwa) (rata-rata 3 tahun)", - "Number of women of reproductive age (15-49 years) affected by anemia (million)": - "Jumlah perempuan usia reproduksi (15-49 tahun) yang menderita anemia (juta jiwa)", - "Percentage of children under 5 years affected by wasting (percent)": - "Persentase anak di bawah 5 tahun yang mengalami wasting (persen)", - "Prevalence of anemia among women of reproductive age (15-49 years) (percent)": - "Prevalensi anemia pada perempuan usia reproduksi (15-49 tahun) (persen)", - "Coefficient of variation of habitual caloric consumption distribution (real number)": - "Koefisien variasi distribusi konsumsi kalori kebiasaan (bilangan riil)", - "Incidence of caloric losses at retail distribution level (percent)": - "Insidensi kehilangan kalori pada tingkat distribusi ritel (persen)", - "Number of children under 5 years of age who are overweight (modeled estimates) (million)": - "Jumlah anak di bawah 5 tahun yang mengalami kelebihan berat badan (estimasi model) (juta jiwa)", - "Number of children under 5 years of age who are stunted (modeled estimates) (million)": - "Jumlah anak di bawah 5 tahun yang mengalami stunting (estimasi model) (juta jiwa)", - "Number of newborns with low birthweight (million)": - "Jumlah bayi baru lahir dengan berat badan lahir rendah (juta jiwa)", - "Number of obese adults (18 years and older) (million)": - "Jumlah orang dewasa yang mengalami obesitas (18 tahun ke atas) (juta jiwa)", - "Percentage of children under 5 years of age who are overweight (modelled estimates) (percent)": - "Persentase anak di bawah 5 tahun yang mengalami kelebihan berat badan (estimasi model) (persen)", - "Percentage of children under 5 years of age who are stunted (modelled estimates) (percent)": - "Persentase anak di bawah 5 tahun yang mengalami stunting (estimasi model) (persen)", - "Prevalence of low birthweight (percent)": - "Prevalensi berat badan lahir rendah (persen)", - "Prevalence of moderate or severe food insecurity in the female adult population (percent) (3-year average)": - "Prevalensi kerawanan pangan sedang atau berat pada penduduk perempuan dewasa (persen) (rata-rata 3 tahun)", - "Prevalence of moderate or severe food insecurity in the male adult population (percent) (3-year average)": - "Prevalensi kerawanan pangan sedang atau berat pada penduduk laki-laki dewasa (persen) (rata-rata 3 tahun)", - "Prevalence of moderate or severe food insecurity in the total population (percent) (3-year average)": - "Prevalensi kerawanan pangan sedang atau berat pada total penduduk (persen) (rata-rata 3 tahun)", - "Prevalence of obesity in the adult population (18 years and older) (percent)": - "Prevalensi obesitas pada penduduk dewasa (18 tahun ke atas) (persen)", - "Prevalence of severe food insecurity in the female adult population (percent) (3-year average)": - "Prevalensi kerawanan pangan berat pada penduduk perempuan dewasa (persen) (rata-rata 3 tahun)", - "Prevalence of severe food insecurity in the male adult population (percent) (3-year average)": - "Prevalensi kerawanan pangan berat pada penduduk laki-laki dewasa (persen) (rata-rata 3 tahun)", - "Prevalence of severe food insecurity in the total population (percent) (3-year average)": - "Prevalensi kerawanan pangan berat pada total penduduk (persen) (rata-rata 3 tahun)", -} - - - - - def translate_country(name: str) -> str: - """Terjemahkan nama negara ke Bahasa Indonesia. Fallback ke nama asli.""" if not name: return name return COUNTRY_NAME_ID_MAP.get(name.strip(), name) -def translate_indicator(name: str) -> str: - """Terjemahkan nama indikator ke Bahasa Indonesia. Fallback ke nama asli.""" - if not name: - return name - return INDICATOR_TRANSLATION_ID.get(name, name) - - def translate_pillar(name: str) -> str: - """Terjemahkan nama pillar ke Bahasa Indonesia. Fallback ke nama asli.""" if not name: return name return PILLAR_TRANSLATION_ID.get(name, name) @@ -417,7 +292,8 @@ def _detect_series_trend(scores: list) -> str: def _detect_country_gap(scores_by_country_year: pd.DataFrame, score_col: str) -> str: std_by_year = ( - scores_by_country_year.groupby("year")[score_col] + scores_by_country_year[scores_by_country_year["country_name"] != ASEAN_COUNTRY_NAME] + .groupby("year")[score_col] .std().dropna() ) if len(std_by_year) < 3: @@ -457,156 +333,9 @@ def _find_anomaly_year(values_by_year: dict) -> tuple: # ============================================================================= -# NARRATIVE BUILDER — OVERVIEW (per tahun) -# FIXED: narrative_id pakai nama negara & pilar Bahasa Indonesia -# ============================================================================= - -def _build_overview_narrative( - year: int, - score: float, - performance_status: str, - yoy_val, - n_mdg: int, - n_sdg: int, - ranking_list: list, - most_improved_country, - most_improved_delta, - most_declined_country, - most_declined_delta, - historical_scores: dict, - country_scores_all: pd.DataFrame, -) -> tuple: - sentences_en = [] - sentences_id = [] - - perf_word_en = "good" if performance_status == "Good" else "below target" - perf_word_id = "baik" if performance_status == "Good" else "di bawah target" - - s1_en = ( - f"In {year}, ASEAN food security scored {_fmt_score(score)} out of 100 " - f"({perf_word_en}), covering {n_mdg + n_sdg} indicators " - f"({n_mdg} MDGs and {n_sdg} SDGs)." - ) - s1_id = ( - f"Pada tahun {year}, skor ketahanan pangan ASEAN mencapai {_fmt_score(score)} dari 100 " - f"({perf_word_id}), mencakup {n_mdg + n_sdg} indikator " - f"({n_mdg} MDGs dan {n_sdg} SDGs)." - ) - sentences_en.append(s1_en) - sentences_id.append(s1_id) - - if yoy_val is not None and not pd.isna(yoy_val): - if abs(yoy_val) < 0.5: - s2_en = f"The score was relatively stable compared to the previous year." - s2_id = f"Skor relatif stabil dibandingkan tahun sebelumnya." - elif yoy_val > 0: - s2_en = f"This represents an improvement of {abs(yoy_val):.2f} points from the previous year." - s2_id = f"Ini merupakan peningkatan sebesar {abs(yoy_val):.2f} poin dari tahun sebelumnya." - else: - s2_en = f"This represents a decline of {abs(yoy_val):.2f} points from the previous year." - s2_id = f"Ini merupakan penurunan sebesar {abs(yoy_val):.2f} poin dari tahun sebelumnya." - sentences_en.append(s2_en) - sentences_id.append(s2_id) - - hist_years = sorted(historical_scores.keys()) - hist_scores = [historical_scores[y] for y in hist_years if not pd.isna(historical_scores.get(y, np.nan))] - - if len(hist_scores) >= 3: - trend = _detect_series_trend(hist_scores) - if trend == "improving_consistent": - s3_en = f"The overall trajectory since {hist_years[0]} has been consistently upward." - s3_id = f"Trajektori keseluruhan sejak {hist_years[0]} menunjukkan tren yang konsisten meningkat." - elif trend == "improving_slowing": - s3_en = f"While the long-term trend since {hist_years[0]} is positive, the pace of improvement has slowed in recent years." - s3_id = f"Meskipun tren jangka panjang sejak {hist_years[0]} positif, laju perbaikan melambat dalam beberapa tahun terakhir." - elif trend == "deteriorating": - s3_en = f"The overall trend since {hist_years[0]} shows a declining trajectory that warrants attention." - s3_id = f"Tren keseluruhan sejak {hist_years[0]} menunjukkan trajektori yang menurun dan perlu perhatian." - elif trend == "fluctuating": - s3_en = f"Progress since {hist_years[0]} has been uneven, with scores fluctuating across years." - s3_id = f"Kemajuan sejak {hist_years[0]} tidak merata, dengan skor yang berfluktuasi antar tahun." - else: - s3_en = "" - s3_id = "" - - if s3_en: - sentences_en.append(s3_en) - sentences_id.append(s3_id) - - if not country_scores_all.empty: - gap_trend = _detect_country_gap( - country_scores_all[country_scores_all["year"] <= year], - "framework_score_1_100" - ) - if gap_trend == "widening": - s4_en = "The performance gap among ASEAN member states has widened over time, indicating unequal progress." - s4_id = "Kesenjangan performa antar negara anggota ASEAN semakin melebar, mengindikasikan kemajuan yang tidak merata." - elif gap_trend == "narrowing": - s4_en = "The performance gap among ASEAN member states has narrowed, reflecting more balanced regional development." - s4_id = "Kesenjangan performa antar negara anggota ASEAN semakin menyempit, mencerminkan pembangunan regional yang lebih merata." - elif gap_trend == "stable": - s4_en = "The performance gap among ASEAN member states has remained relatively stable." - s4_id = "Kesenjangan performa antar negara anggota ASEAN relatif stabil." - else: - s4_en = "" - s4_id = "" - - if s4_en: - sentences_en.append(s4_en) - sentences_id.append(s4_id) - - # Ranking: EN pakai nama Inggris, ID pakai nama Indonesia - if ranking_list and len(ranking_list) >= 2: - top = ranking_list[0] - bottom = ranking_list[-1] - - top_name_en = top['country_name'] - top_name_id = translate_country(top_name_en) - bottom_name_en = bottom['country_name'] - bottom_name_id = translate_country(bottom_name_en) - - s5_en = ( - f"In {year}, {top_name_en} led the region with a score of " - f"{_fmt_score(top['score'])}, while {bottom_name_en} ranked last " - f"at {_fmt_score(bottom['score'])}." - ) - s5_id = ( - f"Pada tahun {year}, {top_name_id} memimpin kawasan dengan skor " - f"{_fmt_score(top['score'])}, sementara {bottom_name_id} berada di " - f"posisi terbawah dengan skor {_fmt_score(bottom['score'])}." - ) - sentences_en.append(s5_en) - sentences_id.append(s5_id) - - # Peningkatan/penurunan: EN pakai nama Inggris, ID pakai nama Indonesia - if most_improved_country and most_declined_country: - improved_name_id = translate_country(most_improved_country) - declined_name_id = translate_country(most_declined_country) - - if most_improved_country != most_declined_country: - s6_en = ( - f"{most_improved_country} showed the biggest improvement " - f"({_fmt_delta(most_improved_delta)} pts), " - f"while {most_declined_country} experienced the largest decline " - f"({_fmt_delta(most_declined_delta)} pts)." - ) - s6_id = ( - f"{improved_name_id} mencatat peningkatan terbesar " - f"({_fmt_delta(most_improved_delta)} poin), " - f"sementara {declined_name_id} mengalami penurunan terbesar " - f"({_fmt_delta(most_declined_delta)} poin)." - ) - sentences_en.append(s6_en) - sentences_id.append(s6_id) - - narrative_en = " ".join(s for s in sentences_en if s) - narrative_id = " ".join(s for s in sentences_id if s) - return narrative_en, narrative_id - - -# ============================================================================= -# NARRATIVE BUILDER — PILLAR (per tahun per pilar) -# FIXED: narrative_id pakai nama negara & pilar Bahasa Indonesia +# NARRATIVE BUILDER — PILLAR +# Digunakan untuk SEMUA baris: per negara dan ASEAN aggregate. +# Jika is_asean=True, narasi tidak menyebut "country" melainkan "ASEAN region". # ============================================================================= def _build_pillar_narrative( @@ -623,17 +352,20 @@ def _build_pillar_narrative( pillar_scores_history: dict, all_pillar_scores_year: pd.DataFrame, country_pillar_all: pd.DataFrame, + is_asean: bool = False, ) -> tuple: sentences_en = [] sentences_id = [] - # Terjemahan nama pilar pillar_name_id = translate_pillar(pillar_name) rank_suffix = {1: "st", 2: "nd", 3: "rd"}.get(rank_in_year, "th") perf_word_en = "good" if pillar_score >= PERFORMANCE_THRESHOLD else "below target" perf_word_id = "baik" if pillar_score >= PERFORMANCE_THRESHOLD else "di bawah target" + subject_en = "ASEAN region" if is_asean else "this region" + subject_id = "kawasan ASEAN" if is_asean else "kawasan ini" + s1_en = ( f"In {year}, the {pillar_name} pillar ranked {rank_in_year}{rank_suffix} out of " f"{n_pillars} pillars with a score of {_fmt_score(pillar_score)} ({perf_word_en})." @@ -682,12 +414,12 @@ def _build_pillar_narrative( else: s3_en = "" s3_id = "" - if s3_en: sentences_en.append(s3_en) sentences_id.append(s3_id) - if not country_pillar_all.empty: + # Gap antar negara hanya relevan untuk ASEAN narrative + if is_asean and not country_pillar_all.empty: gap_trend = _detect_country_gap( country_pillar_all[country_pillar_all["year"] <= year], "pillar_country_score_1_100" @@ -701,16 +433,14 @@ def _build_pillar_narrative( else: s4_en = "" s4_id = "" - if s4_en: sentences_en.append(s4_en) sentences_id.append(s4_id) - # Negara terbaik/terburuk: EN pakai nama Inggris, ID pakai nama Indonesia - if top_country and bot_country and top_country != bot_country: + # Top/bottom hanya ditampilkan untuk baris ASEAN + if is_asean and top_country and bot_country and top_country != bot_country: top_country_id = translate_country(top_country) bot_country_id = translate_country(bot_country) - s5_en = ( f"{top_country} performed best in this pillar ({_fmt_score(top_country_score)}), " f"while {bot_country} had the lowest score ({_fmt_score(bot_country_score)})." @@ -723,25 +453,24 @@ def _build_pillar_narrative( sentences_en.append(s5_en) sentences_id.append(s5_id) - # Perbandingan antar pilar: EN pakai nama Inggris, ID pakai nama Indonesia + # Perbandingan antar pilar if not all_pillar_scores_year.empty and len(all_pillar_scores_year) > 1: - sorted_pillars = all_pillar_scores_year.sort_values("pillar_score_1_100", ascending=False) + sorted_pillars = all_pillar_scores_year.sort_values("pillar_country_score_1_100", ascending=False) strongest = sorted_pillars.iloc[0] weakest = sorted_pillars.iloc[-1] if strongest["pillar_name"] != pillar_name and weakest["pillar_name"] != pillar_name: strongest_id = translate_pillar(strongest["pillar_name"]) weakest_id = translate_pillar(weakest["pillar_name"]) - s6_en = ( f"Across all pillars in {year}, {strongest['pillar_name']} scored highest " - f"({_fmt_score(strongest['pillar_score_1_100'])}) and {weakest['pillar_name']} " - f"scored lowest ({_fmt_score(weakest['pillar_score_1_100'])})." + f"({_fmt_score(strongest['pillar_country_score_1_100'])}) and {weakest['pillar_name']} " + f"scored lowest ({_fmt_score(weakest['pillar_country_score_1_100'])})." ) s6_id = ( f"Di antara semua pilar pada tahun {year}, {strongest_id} mendapat skor " - f"tertinggi ({_fmt_score(strongest['pillar_score_1_100'])}) dan {weakest_id} " - f"mendapat skor terendah ({_fmt_score(weakest['pillar_score_1_100'])})." + f"tertinggi ({_fmt_score(strongest['pillar_country_score_1_100'])}) dan {weakest_id} " + f"mendapat skor terendah ({_fmt_score(weakest['pillar_country_score_1_100'])})." ) sentences_en.append(s6_en) sentences_id.append(s6_id) @@ -763,11 +492,8 @@ class FoodSecurityAggregator: self.logger.propagate = False self.load_metadata = { - "agg_pillar_composite": {"rows_loaded": 0, "status": "pending", "start_time": None, "end_time": None}, "agg_pillar_by_country": {"rows_loaded": 0, "status": "pending", "start_time": None, "end_time": None}, "agg_framework_by_country": {"rows_loaded": 0, "status": "pending", "start_time": None, "end_time": None}, - "agg_framework_asean": {"rows_loaded": 0, "status": "pending", "start_time": None, "end_time": None}, - "agg_narrative_overview": {"rows_loaded": 0, "status": "pending", "start_time": None, "end_time": None}, "agg_narrative_pillar": {"rows_loaded": 0, "status": "pending", "start_time": None, "end_time": None}, } @@ -804,32 +530,23 @@ class FoodSecurityAggregator: self.logger.warning(f" [DIRECTION] {n_null_dir} rows NULL -> diisi 'positive'") self.df["direction"] = self.df["direction"].fillna("positive") - # Pastikan kolom terjemahan Indonesia tersedia + # Rename pillar_name: add 'Food ' prefix, remove Sustainability + PILLAR_RENAME_MAP = { + 'Availability' : 'Food Availability', + 'Access' : 'Food Access', + 'Utilization' : 'Food Utilization', + 'Stability' : 'Food Stability', + 'Other' : 'Food Other', + 'Sustainability': 'Food Other', + 'sustainability': 'Food Other', + } + self.df["pillar_name"] = self.df["pillar_name"].replace(PILLAR_RENAME_MAP) + + # Kolom terjemahan Indonesia if "country_name_id" not in self.df.columns: self.df["country_name_id"] = self.df["country_name"].apply(translate_country) - self.logger.info(" [TRANSLATION] Kolom country_name_id dibuat dari mapping.") - if "indicator_name_id" not in self.df.columns: - self.df["indicator_name_id"] = self.df["indicator_name"].apply(translate_indicator) - self.logger.info(" [TRANSLATION] Kolom indicator_name_id dibuat dari mapping.") if "pillar_name_id" not in self.df.columns: self.df["pillar_name_id"] = self.df["pillar_name"].apply(translate_pillar) - self.logger.info(" [TRANSLATION] Kolom pillar_name_id dibuat dari mapping.") - - # Log terjemahan negara - sample_ctr = ( - self.df[["country_name", "country_name_id"]] - .drop_duplicates() - .sort_values("country_name") - ) - self.logger.info("\n Terjemahan nama negara (EN -> ID):") - for _, r in sample_ctr.iterrows(): - self.logger.info(f" {r['country_name']:<35} -> {r['country_name_id']}") - - # Log terjemahan pilar - sample_pil = self.df[["pillar_name", "pillar_name_id"]].drop_duplicates() - self.logger.info("\n Terjemahan nama pilar (EN -> ID):") - for _, r in sample_pil.iterrows(): - self.logger.info(f" {r['pillar_name']:<20} -> {r['pillar_name_id']}") self.logger.info(f"\n Rows : {len(self.df):,}") self.logger.info(f" Countries : {self.df['country_id'].nunique()}") @@ -966,6 +683,9 @@ class FoodSecurityAggregator: "normalize_frameworks_jointly": NORMALIZE_FRAMEWORKS_JOINTLY, "performance_threshold" : PERFORMANCE_THRESHOLD, "status" : status, + "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)", }), "validation_metrics" : json.dumps({ "status" : status, @@ -974,108 +694,75 @@ class FoodSecurityAggregator: } # ========================================================================= - # STEP 2: agg_pillar_composite - # Kolom tambahan: pillar_name_id + # HELPER: build ASEAN rows untuk tabel pillar_by_country # ========================================================================= - def calc_pillar_composite(self) -> pd.DataFrame: - table_name = "agg_pillar_composite" - self.load_metadata[table_name]["start_time"] = datetime.now() - self.logger.info("\n" + "=" * 70) - self.logger.info(f"STEP 2: {table_name} -> [Gold] fs_asean_gold") - self.logger.info("=" * 70) - - try: - df_normed = self._get_norm_value_df() - - df = ( - df_normed - .groupby(["pillar_id", "pillar_name", "year"]) - .agg( - pillar_norm =("norm_value", "mean"), - n_indicators=("indicator_id", "nunique"), - n_countries =("country_id", "nunique"), - ) - .reset_index() - ) - - df["pillar_score_1_100"] = global_minmax(df["pillar_norm"]) - df["rank_in_year"] = ( - df.groupby("year")["pillar_score_1_100"] - .rank(method="min", ascending=False) - .astype(int) - ) - df = add_yoy(df, ["pillar_id"], "pillar_score_1_100") - - # Kolom terjemahan Indonesia — FIXED: "Access" -> "Akses" - df["pillar_name_id"] = df["pillar_name"].apply(translate_pillar) - - df["pillar_id"] = df["pillar_id"].astype(int) - df["year"] = df["year"].astype(int) - df["n_indicators"] = safe_int(df["n_indicators"], col_name="n_indicators", logger=self.logger) - df["n_countries"] = safe_int(df["n_countries"], col_name="n_countries", logger=self.logger) - df["rank_in_year"] = df["rank_in_year"].astype(int) - df["pillar_norm"] = df["pillar_norm"].astype(float) - df["pillar_score_1_100"] = df["pillar_score_1_100"].astype(float) - df["pillar_name_id"] = df["pillar_name_id"].astype(str) - - schema = [ - bigquery.SchemaField("pillar_id", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("pillar_name", "STRING", mode="REQUIRED"), - bigquery.SchemaField("pillar_name_id", "STRING", mode="REQUIRED"), - bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("pillar_norm", "FLOAT", mode="REQUIRED"), - bigquery.SchemaField("n_indicators", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("n_countries", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("pillar_score_1_100", "FLOAT", mode="REQUIRED"), - bigquery.SchemaField("rank_in_year", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("year_over_year_change", "FLOAT", mode="NULLABLE"), - ] - rows = load_to_bigquery( - self.client, df, table_name, layer='gold', - write_disposition="WRITE_TRUNCATE", schema=schema - ) - self._finalize(table_name, rows) - return df - - except Exception as e: - self._fail(table_name, e) - raise + def _build_asean_pillar_rows(self, df_normed: pd.DataFrame) -> pd.DataFrame: + """ + Hitung rata-rata ASEAN per pillar per year dari norm_value semua negara, + kemudian scale ulang ke 1-100 dalam konteks SELURUH tabel (negara + ASEAN). + Return DataFrame dengan format sama seperti baris per-negara. + """ + asean_agg = ( + df_normed + .groupby(["pillar_id", "pillar_name", "year"]) + .agg(pillar_country_norm=("norm_value", "mean")) + .reset_index() + ) + asean_agg["country_id"] = ASEAN_COUNTRY_ID + asean_agg["country_name"] = ASEAN_COUNTRY_NAME + asean_agg["country_name_id"] = ASEAN_COUNTRY_NAME_ID + asean_agg["pillar_name_id"] = asean_agg["pillar_name"].apply(translate_pillar) + return asean_agg # ========================================================================= - # STEP 3: agg_pillar_by_country - # Kolom tambahan: pillar_name_id, country_name_id + # STEP 2: agg_pillar_by_country (termasuk ASEAN) # ========================================================================= def calc_pillar_by_country(self) -> pd.DataFrame: table_name = "agg_pillar_by_country" self.load_metadata[table_name]["start_time"] = datetime.now() self.logger.info("\n" + "=" * 70) - self.logger.info(f"STEP 3: {table_name} -> [Gold] fs_asean_gold") + self.logger.info(f"STEP 2: {table_name} -> [Gold] fs_asean_gold") + self.logger.info(" Termasuk baris ASEAN (country_id=0) untuk filter Looker Studio") self.logger.info("=" * 70) try: df_normed = self._get_norm_value_df() - df = ( + # Baris per negara + df_countries = ( df_normed .groupby(["country_id", "country_name", "pillar_id", "pillar_name", "year"]) .agg(pillar_country_norm=("norm_value", "mean")) .reset_index() ) + df_countries["pillar_name_id"] = df_countries["pillar_name"].apply(translate_pillar) + df_countries["country_name_id"] = df_countries["country_name"].apply(translate_country) + # Baris ASEAN aggregate + df_asean = self._build_asean_pillar_rows(df_normed) + + # Gabung + df = pd.concat([df_countries, df_asean], ignore_index=True) + + # Scale 1-100 secara BERSAMA (negara + ASEAN dalam satu ruang skala) df["pillar_country_score_1_100"] = global_minmax(df["pillar_country_norm"]) - df["rank_in_pillar_year"] = ( - df.groupby(["pillar_id", "year"])["pillar_country_score_1_100"] - .rank(method="min", ascending=False) - .astype(int) + + # Rank hanya di antara negara asli (ASEAN tidak di-rank melawan dirinya sendiri) + country_only = df[df["country_id"] != ASEAN_COUNTRY_ID].copy() + country_only["rank_in_pillar_year"] = ( + country_only.groupby(["pillar_id", "year"])["pillar_country_score_1_100"] + .rank(method="min", ascending=False) + .astype(int) ) + asean_only = df[df["country_id"] == ASEAN_COUNTRY_ID].copy() + asean_only["rank_in_pillar_year"] = 0 # 0 = ASEAN aggregate + + df = pd.concat([country_only, asean_only], ignore_index=True) df = add_yoy(df, ["country_id", "pillar_id"], "pillar_country_score_1_100") - # Kolom terjemahan Indonesia — FIXED - df["pillar_name_id"] = df["pillar_name"].apply(translate_pillar) - df["country_name_id"] = df["country_name"].apply(translate_country) - + # Tipe data df["country_id"] = df["country_id"].astype(int) df["pillar_id"] = df["pillar_id"].astype(int) df["year"] = df["year"].astype(int) @@ -1085,6 +772,11 @@ class FoodSecurityAggregator: df["pillar_name_id"] = df["pillar_name_id"].astype(str) df["country_name_id"] = df["country_name_id"].astype(str) + self.logger.info( + f" Total rows: {len(df):,} " + f"({len(df_countries):,} country + {len(asean_only):,} ASEAN)" + ) + schema = [ bigquery.SchemaField("country_id", "INTEGER", mode="REQUIRED"), bigquery.SchemaField("country_name", "STRING", mode="REQUIRED"), @@ -1110,8 +802,7 @@ class FoodSecurityAggregator: raise # ========================================================================= - # STEP 4: agg_framework_by_country - # Tambah kolom: country_name_id + # HELPER: composite per country (untuk framework_by_country) # ========================================================================= def _calc_country_composite_inmemory(self) -> pd.DataFrame: @@ -1140,11 +831,16 @@ class FoodSecurityAggregator: df["rank_in_asean"] = df["rank_in_asean"].astype(int) return df + # ========================================================================= + # STEP 3: agg_framework_by_country (termasuk ASEAN) + # ========================================================================= + def calc_framework_by_country(self) -> pd.DataFrame: table_name = "agg_framework_by_country" self.load_metadata[table_name]["start_time"] = datetime.now() self.logger.info("\n" + "=" * 70) - self.logger.info(f"STEP 4: {table_name} -> [Gold] fs_asean_gold") + self.logger.info(f"STEP 3: {table_name} -> [Gold] fs_asean_gold") + self.logger.info(" Termasuk baris ASEAN (country_id=0)") self.logger.info("=" * 70) try: @@ -1152,6 +848,7 @@ class FoodSecurityAggregator: df_normed = self._get_norm_value_df() parts = [] + # ---- Per negara (Total) ---- agg_total = ( country_composite[[ "country_id", "country_name", "year", @@ -1228,7 +925,31 @@ class FoodSecurityAggregator: agg_sdgs["framework"] = "SDGs" parts.append(agg_sdgs) - df = pd.concat(parts, ignore_index=True) + df_countries = pd.concat(parts, ignore_index=True) + + # ---- ASEAN aggregate (rata-rata dari semua negara per framework per year) ---- + asean_parts = [] + for fw in df_countries["framework"].unique(): + fw_df = df_countries[ + (df_countries["framework"] == fw) & + (df_countries["country_id"] != ASEAN_COUNTRY_ID) + ] + asean_fw = ( + fw_df.groupby(["year", "framework"]) + .agg( + framework_norm =("framework_norm", "mean"), + framework_score_1_100 =("framework_score_1_100", "mean"), + n_indicators =("n_indicators", "mean"), + ) + .reset_index() + ) + asean_fw["country_id"] = ASEAN_COUNTRY_ID + asean_fw["country_name"] = ASEAN_COUNTRY_NAME + asean_parts.append(asean_fw) + + df_asean_fw = pd.concat(asean_parts, ignore_index=True) + + df = pd.concat([df_countries, df_asean_fw], ignore_index=True) if NORMALIZE_FRAMEWORKS_JOINTLY: mixed_mask = (df["framework"].isin(["MDGs", "SDGs"])) & (df["year"] >= self.sdgs_start_year) @@ -1236,14 +957,17 @@ class FoodSecurityAggregator: df.loc[mixed_mask, "framework_score_1_100"] = global_minmax(df.loc[mixed_mask, "framework_norm"]) df = check_and_dedup(df, ["country_id", "framework", "year"], context=table_name, logger=self.logger) - df["rank_in_framework_year"] = ( - df.groupby(["framework", "year"])["framework_score_1_100"] - .rank(method="min", ascending=False) - .astype(int) - ) - df = add_yoy(df, ["country_id", "framework"], "framework_score_1_100") - # Tambah kolom nama negara Indonesia + # Rank hanya di antara negara asli + country_mask = df["country_id"] != ASEAN_COUNTRY_ID + df.loc[country_mask, "rank_in_framework_year"] = ( + df[country_mask] + .groupby(["framework", "year"])["framework_score_1_100"] + .rank(method="min", ascending=False) + ) + df.loc[~country_mask, "rank_in_framework_year"] = 0 + + df = add_yoy(df, ["country_id", "framework"], "framework_score_1_100") df["country_name_id"] = df["country_name"].apply(translate_country) df["country_id"] = df["country_id"].astype(int) @@ -1254,8 +978,6 @@ class FoodSecurityAggregator: df["framework_score_1_100"] = df["framework_score_1_100"].astype(float) df["country_name_id"] = df["country_name_id"].astype(str) - self._validate_mdgs_equals_total(df, level="country") - schema = [ bigquery.SchemaField("country_id", "INTEGER", mode="REQUIRED"), bigquery.SchemaField("country_name", "STRING", mode="REQUIRED"), @@ -1280,455 +1002,162 @@ class FoodSecurityAggregator: raise # ========================================================================= - # STEP 5: agg_framework_asean - # ========================================================================= - - def calc_framework_asean(self) -> pd.DataFrame: - table_name = "agg_framework_asean" - self.load_metadata[table_name]["start_time"] = datetime.now() - self.logger.info("\n" + "=" * 70) - self.logger.info(f"STEP 5: {table_name} -> [Gold] fs_asean_gold") - self.logger.info("=" * 70) - - try: - df_normed = self._get_norm_value_df() - country_composite = self._calc_country_composite_inmemory() - - country_norm = ( - df_normed - .groupby(["country_id", "country_name", "year"])["norm_value"] - .mean().reset_index() - .rename(columns={"norm_value": "country_norm"}) - ) - asean_overall = ( - country_norm.groupby("year") - .agg( - asean_norm =("country_norm", "mean"), - std_norm =("country_norm", "std"), - n_countries=("country_norm", "count"), - ) - .reset_index() - ) - asean_overall["asean_score_1_100"] = global_minmax(asean_overall["asean_norm"]) - asean_comp = ( - country_composite.groupby("year")["composite_score"] - .mean().reset_index() - .rename(columns={"composite_score": "asean_composite"}) - ) - asean_overall = asean_overall.merge(asean_comp, on="year", how="left") - - parts = [] - - def _n_ind(year_val, framework_val): - return self._count_framework_indicators(year_val, framework_val) - - total_cols = asean_overall[[ - "year", "asean_score_1_100", "asean_norm", "std_norm", "n_countries" - ]].copy().rename(columns={ - "asean_score_1_100": "framework_score_1_100", - "asean_norm" : "framework_norm", - "n_countries" : "n_countries_with_data", - }) - total_cols["n_indicators"] = total_cols["year"].apply( - lambda y: int(self._ind_year_framework[ - self._ind_year_framework["year"] == y - ]["indicator_id"].nunique()) - ) - total_cols["framework"] = "Total" - parts.append(total_cols) - - pre_sdgs = asean_overall[asean_overall["year"] < self.sdgs_start_year].copy() - if not pre_sdgs.empty: - mdgs_pre = pre_sdgs[[ - "year", "asean_score_1_100", "asean_norm", "std_norm", "n_countries" - ]].copy().rename(columns={ - "asean_score_1_100": "framework_score_1_100", - "asean_norm" : "framework_norm", - "n_countries" : "n_countries_with_data", - }) - mdgs_pre["n_indicators"] = mdgs_pre["year"].apply(lambda y: _n_ind(y, "MDGs")) - mdgs_pre["framework"] = "MDGs" - parts.append(mdgs_pre) - - mdgs_indicator_ids = set( - self._ind_year_framework[self._ind_year_framework["framework"] == "MDGs"]["indicator_id"] - ) - if mdgs_indicator_ids: - df_mdgs_mixed = df_normed[ - (df_normed["indicator_id"].isin(mdgs_indicator_ids)) & - (df_normed["year"] >= self.sdgs_start_year) - ].copy() - if not df_mdgs_mixed.empty: - cn = ( - df_mdgs_mixed - .groupby(["country_id", "year"])["norm_value"].mean() - .reset_index().rename(columns={"norm_value": "country_norm"}) - ) - asean_mdgs = cn.groupby("year").agg( - framework_norm =("country_norm", "mean"), - std_norm =("country_norm", "std"), - n_countries_with_data=("country_id", "count"), - ).reset_index() - asean_mdgs["n_indicators"] = asean_mdgs["year"].apply(lambda y: _n_ind(y, "MDGs")) - if not NORMALIZE_FRAMEWORKS_JOINTLY: - asean_mdgs["framework_score_1_100"] = global_minmax(asean_mdgs["framework_norm"]) - asean_mdgs["framework"] = "MDGs" - parts.append(asean_mdgs) - - sdgs_indicator_ids = set( - self._ind_year_framework[self._ind_year_framework["framework"] == "SDGs"]["indicator_id"] - ) - if sdgs_indicator_ids: - df_sdgs = df_normed[ - (df_normed["indicator_id"].isin(sdgs_indicator_ids)) & - (df_normed["year"] >= self.sdgs_start_year) - ].copy() - if not df_sdgs.empty: - cn = ( - df_sdgs - .groupby(["country_id", "year"])["norm_value"].mean() - .reset_index().rename(columns={"norm_value": "country_norm"}) - ) - asean_sdgs = cn.groupby("year").agg( - framework_norm =("country_norm", "mean"), - std_norm =("country_norm", "std"), - n_countries_with_data=("country_id", "count"), - ).reset_index() - asean_sdgs["n_indicators"] = asean_sdgs["year"].apply(lambda y: _n_ind(y, "SDGs")) - if not NORMALIZE_FRAMEWORKS_JOINTLY: - asean_sdgs["framework_score_1_100"] = global_minmax(asean_sdgs["framework_norm"]) - asean_sdgs["framework"] = "SDGs" - parts.append(asean_sdgs) - - df = pd.concat(parts, ignore_index=True) - - if NORMALIZE_FRAMEWORKS_JOINTLY: - mixed_mask = (df["framework"].isin(["MDGs", "SDGs"])) & (df["year"] >= self.sdgs_start_year) - if mixed_mask.any(): - df.loc[mixed_mask, "framework_score_1_100"] = global_minmax(df.loc[mixed_mask, "framework_norm"]) - - df = check_and_dedup(df, ["framework", "year"], context=table_name, logger=self.logger) - df = add_yoy(df, ["framework"], "framework_score_1_100") - - df["performance_status"] = df["framework_score_1_100"].apply(_performance_status) - df["year"] = df["year"].astype(int) - df["n_indicators"] = safe_int(df["n_indicators"], col_name="n_indicators", logger=self.logger) - df["n_countries_with_data"] = safe_int(df["n_countries_with_data"], col_name="n_countries_with_data", logger=self.logger) - for col in ["framework_norm", "std_norm", "framework_score_1_100"]: - df[col] = df[col].astype(float) - - self._validate_mdgs_equals_total(df, level="asean") - - schema = [ - bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("framework", "STRING", mode="REQUIRED"), - bigquery.SchemaField("n_indicators", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("n_countries_with_data", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("framework_norm", "FLOAT", mode="REQUIRED"), - bigquery.SchemaField("std_norm", "FLOAT", mode="NULLABLE"), - bigquery.SchemaField("framework_score_1_100", "FLOAT", mode="REQUIRED"), - bigquery.SchemaField("year_over_year_change", "FLOAT", mode="NULLABLE"), - bigquery.SchemaField("performance_status", "STRING", mode="REQUIRED"), - ] - rows = load_to_bigquery( - self.client, df, table_name, layer='gold', - write_disposition="WRITE_TRUNCATE", schema=schema - ) - self._finalize(table_name, rows) - return df - - except Exception as e: - self._fail(table_name, e) - raise - - # ========================================================================= - # STEP 6: agg_narrative_overview - # FIXED: narrative_id pakai nama negara Indonesia - # ========================================================================= - - def calc_narrative_overview( - self, - df_framework_asean: pd.DataFrame, - df_framework_by_country: pd.DataFrame, - ) -> pd.DataFrame: - table_name = "agg_narrative_overview" - self.load_metadata[table_name]["start_time"] = datetime.now() - self.logger.info("\n" + "=" * 70) - self.logger.info(f"STEP 6: {table_name} -> [Gold] fs_asean_gold") - self.logger.info(" Narrative: interpretatif, plain text, bilingual EN/ID") - self.logger.info(" FIXED : narrative_id pakai nama negara Bahasa Indonesia") - self.logger.info("=" * 70) - - try: - asean_total = ( - df_framework_asean[df_framework_asean["framework"] == "Total"] - .sort_values("year") - .reset_index(drop=True) - ) - - score_by_year = dict(zip(asean_total["year"].astype(int), asean_total["framework_score_1_100"].astype(float))) - status_by_year = dict(zip(asean_total["year"].astype(int), asean_total["performance_status"].astype(str))) - country_total = df_framework_by_country[df_framework_by_country["framework"] == "Total"].copy() - - records = [] - - for _, row in asean_total.iterrows(): - yr = int(row["year"]) - score = float(row["framework_score_1_100"]) - perf_status = str(row["performance_status"]) - yoy = row["year_over_year_change"] - yoy_val = float(yoy) if pd.notna(yoy) else None - - n_mdg = self._count_framework_indicators(yr, "MDGs") - n_sdg = self._count_framework_indicators(yr, "SDGs") - n_total_ind = int( - self._ind_year_framework[ - self._ind_year_framework["year"] == yr - ]["indicator_id"].nunique() - ) - - prev_score = score_by_year.get(yr - 1, None) - prev_status = status_by_year.get(yr - 1, "N/A") - yoy_pct = ( - (yoy_val / prev_score * 100) - if (yoy_val is not None and prev_score is not None and prev_score != 0) - else None - ) - - yr_country = ( - country_total[country_total["year"] == yr] - .sort_values("rank_in_framework_year") - .reset_index(drop=True) - ) - - ranking_list = [] - for _, cr in yr_country.iterrows(): - cr_yoy = cr.get("year_over_year_change", None) - ranking_list.append({ - "rank": int(cr["rank_in_framework_year"]), - "country_name": str(cr["country_name"]), - "score": round(float(cr["framework_score_1_100"]), 2), - "yoy_change": round(float(cr_yoy), 2) if pd.notna(cr_yoy) else None, - }) - country_ranking_json = json.dumps(ranking_list, ensure_ascii=False) - - yr_country_yoy = yr_country.dropna(subset=["year_over_year_change"]) - if not yr_country_yoy.empty: - best_idx = yr_country_yoy["year_over_year_change"].idxmax() - worst_idx = yr_country_yoy["year_over_year_change"].idxmin() - most_improved_country = str(yr_country_yoy.loc[best_idx, "country_name"]) - most_improved_delta = round(float(yr_country_yoy.loc[best_idx, "year_over_year_change"]), 2) - most_declined_country = str(yr_country_yoy.loc[worst_idx, "country_name"]) - most_declined_delta = round(float(yr_country_yoy.loc[worst_idx, "year_over_year_change"]), 2) - else: - most_improved_country = most_declined_country = None - most_improved_delta = most_declined_delta = None - - country_scores_all = country_total[["year", "country_id", "framework_score_1_100"]].copy() - - narrative_en, narrative_id = _build_overview_narrative( - year = yr, - score = score, - performance_status = perf_status, - yoy_val = yoy_val, - n_mdg = n_mdg, - n_sdg = n_sdg, - ranking_list = ranking_list, - most_improved_country = most_improved_country, - most_improved_delta = most_improved_delta, - most_declined_country = most_declined_country, - most_declined_delta = most_declined_delta, - historical_scores = score_by_year, - country_scores_all = country_scores_all, - ) - - records.append({ - "year": yr, - "n_mdg_indicators": n_mdg, - "n_sdg_indicators": n_sdg, - "n_total_indicators": n_total_ind, - "asean_total_score": round(score, 2), - "performance_status": perf_status, - "yoy_change": yoy_val, - "yoy_change_pct": round(yoy_pct, 2) if yoy_pct is not None else None, - "country_ranking_json": country_ranking_json, - "most_improved_country": most_improved_country, - "most_improved_delta": most_improved_delta, - "most_declined_country": most_declined_country, - "most_declined_delta": most_declined_delta, - "narrative_en": narrative_en, - "narrative_id": narrative_id, - }) - - df = pd.DataFrame(records) - df["year"] = df["year"].astype(int) - df["n_mdg_indicators"] = df["n_mdg_indicators"].astype(int) - df["n_sdg_indicators"] = df["n_sdg_indicators"].astype(int) - df["n_total_indicators"] = df["n_total_indicators"].astype(int) - df["asean_total_score"] = df["asean_total_score"].astype(float) - df["performance_status"] = df["performance_status"].astype(str) - df["narrative_en"] = df["narrative_en"].astype(str) - df["narrative_id"] = df["narrative_id"].astype(str) - for col in ["yoy_change", "yoy_change_pct", "most_improved_delta", "most_declined_delta"]: - df[col] = pd.to_numeric(df[col], errors="coerce").astype(float) - - self.logger.info("\n Sample narrative_en (year 1):") - self.logger.info(f" {df.iloc[0]['narrative_en'][:300]}") - self.logger.info("\n Sample narrative_id (year 1):") - self.logger.info(f" {df.iloc[0]['narrative_id'][:300]}") - - schema = [ - bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("n_mdg_indicators", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("n_sdg_indicators", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("n_total_indicators", "INTEGER", mode="REQUIRED"), - bigquery.SchemaField("asean_total_score", "FLOAT", mode="REQUIRED"), - bigquery.SchemaField("performance_status", "STRING", mode="REQUIRED"), - bigquery.SchemaField("yoy_change", "FLOAT", mode="NULLABLE"), - bigquery.SchemaField("yoy_change_pct", "FLOAT", mode="NULLABLE"), - bigquery.SchemaField("country_ranking_json", "STRING", mode="REQUIRED"), - bigquery.SchemaField("most_improved_country", "STRING", mode="NULLABLE"), - bigquery.SchemaField("most_improved_delta", "FLOAT", mode="NULLABLE"), - bigquery.SchemaField("most_declined_country", "STRING", mode="NULLABLE"), - bigquery.SchemaField("most_declined_delta", "FLOAT", mode="NULLABLE"), - bigquery.SchemaField("narrative_en", "STRING", mode="REQUIRED"), - bigquery.SchemaField("narrative_id", "STRING", mode="REQUIRED"), - ] - rows = load_to_bigquery( - self.client, df, table_name, layer='gold', - write_disposition="WRITE_TRUNCATE", schema=schema, - ) - self._finalize(table_name, rows) - return df - - except Exception as e: - self._fail(table_name, e) - raise - - # ========================================================================= - # STEP 7: agg_narrative_pillar - # Kolom tambahan: pillar_name_id - # FIXED: narrative_id pakai nama negara & pilar Bahasa Indonesia + # STEP 4: agg_narrative_pillar (termasuk baris ASEAN) # ========================================================================= def calc_narrative_pillar( self, - df_pillar_composite: pd.DataFrame, df_pillar_by_country: pd.DataFrame, ) -> pd.DataFrame: table_name = "agg_narrative_pillar" self.load_metadata[table_name]["start_time"] = datetime.now() self.logger.info("\n" + "=" * 70) - self.logger.info(f"STEP 7: {table_name} -> [Gold] fs_asean_gold") - self.logger.info(" Narrative: interpretatif, plain text, bilingual EN/ID") - self.logger.info(" FIXED : narrative_id pakai nama negara & pilar Bahasa Indonesia") + self.logger.info(f"STEP 4: {table_name} -> [Gold] fs_asean_gold") + self.logger.info(" Termasuk baris ASEAN (country_id=0)") + self.logger.info(" Filter country_name='ASEAN' untuk overview regional") self.logger.info("=" * 70) try: records = [] - years = sorted(df_pillar_composite["year"].unique()) + years = sorted(df_pillar_by_country["year"].unique()) + pillars = df_pillar_by_country["pillar_id"].unique() - # Precompute history per pillar - pillar_history = {} - for p_id, grp in df_pillar_composite.groupby("pillar_id"): - pillar_history[p_id] = dict( - zip(grp["year"].astype(int), grp["pillar_score_1_100"].astype(float)) + # Precompute history per country x pillar + history = {} + for (c_id, p_id), grp in df_pillar_by_country.groupby(["country_id", "pillar_id"]): + history[(c_id, p_id)] = dict( + zip(grp["year"].astype(int), grp["pillar_country_score_1_100"].astype(float)) ) for yr in years: - yr_pillars = ( - df_pillar_composite[df_pillar_composite["year"] == yr] - .sort_values("rank_in_year") - .reset_index(drop=True) - ) - yr_country_pillar = df_pillar_by_country[df_pillar_by_country["year"] == yr] + yr_df = df_pillar_by_country[df_pillar_by_country["year"] == yr] - for _, prow in yr_pillars.iterrows(): - p_id = int(prow["pillar_id"]) - p_name = str(prow["pillar_name"]) - p_score = float(prow["pillar_score_1_100"]) - p_rank = int(prow["rank_in_year"]) - p_yoy = prow["year_over_year_change"] - p_yoy_val = float(p_yoy) if pd.notna(p_yoy) else None + # Semua negara asli untuk referensi top/bottom dalam narasi ASEAN + country_only_yr = yr_df[yr_df["country_id"] != ASEAN_COUNTRY_ID] - # Terjemahan Indonesia nama pillar — FIXED - p_name_id = translate_pillar(p_name) + for p_id in pillars: + yr_pillar_all = yr_df[yr_df["pillar_id"] == p_id] + if yr_pillar_all.empty: + continue - p_country = ( - yr_country_pillar[yr_country_pillar["pillar_id"] == p_id] - .sort_values("rank_in_pillar_year") - .reset_index(drop=True) - ) - if not p_country.empty: - top_country = str(p_country.iloc[0]["country_name"]) - top_country_score = round(float(p_country.iloc[0]["pillar_country_score_1_100"]), 2) - bot_country = str(p_country.iloc[-1]["country_name"]) - bot_country_score = round(float(p_country.iloc[-1]["pillar_country_score_1_100"]), 2) + p_name_row = yr_pillar_all.iloc[0] + p_name = str(p_name_row["pillar_name"]) + n_pillars = len(pillars) + + # Ranking di antara semua pillar (gunakan skor ASEAN untuk rank antar pillar) + asean_yr_all_pillars = yr_df[yr_df["country_id"] == ASEAN_COUNTRY_ID] + asean_sorted = asean_yr_all_pillars.sort_values("pillar_country_score_1_100", ascending=False).reset_index(drop=True) + + # Top/bottom di antara negara asli (untuk narasi ASEAN) + country_pillar_yr = country_only_yr[country_only_yr["pillar_id"] == p_id] + if not country_pillar_yr.empty: + top_row = country_pillar_yr.loc[country_pillar_yr["pillar_country_score_1_100"].idxmax()] + bot_row = country_pillar_yr.loc[country_pillar_yr["pillar_country_score_1_100"].idxmin()] + top_country = str(top_row["country_name"]) + top_score = round(float(top_row["pillar_country_score_1_100"]), 2) + bot_country = str(bot_row["country_name"]) + bot_score = round(float(bot_row["pillar_country_score_1_100"]), 2) else: top_country = bot_country = None - top_country_score = bot_country_score = None + top_score = bot_score = None - hist_up_to_yr = { - y: s for y, s in pillar_history.get(p_id, {}).items() if y <= yr - } + # Iterasi setiap baris (negara + ASEAN) pada pillar ini + for _, row in yr_pillar_all.iterrows(): + c_id = int(row["country_id"]) + c_name = str(row["country_name"]) + c_name_id = translate_country(c_name) + p_score = float(row["pillar_country_score_1_100"]) + p_yoy = row.get("year_over_year_change", None) + p_yoy_val = float(p_yoy) if pd.notna(p_yoy) else None + p_name_id = translate_pillar(p_name) + is_asean = (c_id == ASEAN_COUNTRY_ID) - country_pillar_all = df_pillar_by_country[ - df_pillar_by_country["pillar_id"] == p_id - ][["year", "country_id", "pillar_country_score_1_100"]].copy() + # Rank pilar ini dalam konteks yang sesuai + if is_asean: + # ASEAN: rank pilar ini di antara semua pilar ASEAN tahun ini + rank_sorted = asean_sorted.reset_index(drop=True) + p_rank = int(rank_sorted[rank_sorted["pillar_id"] == p_id].index[0]) + 1 if p_id in rank_sorted["pillar_id"].values else 0 + else: + # Negara: rank pillar ini di antara semua pillar negara ini + country_all_pillars = yr_df[yr_df["country_id"] == c_id].sort_values("pillar_country_score_1_100", ascending=False).reset_index(drop=True) + p_rank = int(country_all_pillars[country_all_pillars["pillar_id"] == p_id].index[0]) + 1 if p_id in country_all_pillars["pillar_id"].values else 0 - narrative_en, narrative_id = _build_pillar_narrative( - year = yr, - pillar_name = p_name, - pillar_score = p_score, - rank_in_year = p_rank, - n_pillars = len(yr_pillars), - yoy_val = p_yoy_val, - top_country = top_country, - top_country_score = top_country_score, - bot_country = bot_country, - bot_country_score = bot_country_score, - pillar_scores_history = hist_up_to_yr, - all_pillar_scores_year= yr_pillars[["pillar_name", "pillar_score_1_100"]].copy(), - country_pillar_all = country_pillar_all, - ) + hist_up = {y: s for y, s in history.get((c_id, p_id), {}).items() if y <= yr} - records.append({ - "year": yr, - "pillar_id": p_id, - "pillar_name": p_name, - "pillar_name_id": p_name_id, - "pillar_score": round(p_score, 2), - "rank_in_year": p_rank, - "yoy_change": p_yoy_val, - "top_country": top_country, - "top_country_id": translate_country(top_country) if top_country else None, - "top_country_score": top_country_score, - "bottom_country": bot_country, - "bottom_country_id": translate_country(bot_country) if bot_country else None, - "bottom_country_score": bot_country_score, - "narrative_en": narrative_en, - "narrative_id": narrative_id, - }) + # all_pillar_scores_year untuk perbandingan lintas pilar + all_pillar_yr = yr_df[yr_df["country_id"] == c_id][["pillar_name", "pillar_country_score_1_100"]].copy() + + # country_pillar_all untuk gap trend (hanya relevan untuk ASEAN) + cpa = df_pillar_by_country[ + (df_pillar_by_country["pillar_id"] == p_id) & + (df_pillar_by_country["country_id"] != ASEAN_COUNTRY_ID) + ][["year", "country_id", "country_name", "pillar_country_score_1_100"]].copy() + + narrative_en, narrative_id = _build_pillar_narrative( + year = yr, + pillar_name = p_name, + pillar_score = p_score, + rank_in_year = p_rank, + n_pillars = n_pillars, + yoy_val = p_yoy_val, + top_country = top_country if is_asean else None, + top_country_score = top_score if is_asean else None, + bot_country = bot_country if is_asean else None, + bot_country_score = bot_score if is_asean else None, + pillar_scores_history = hist_up, + all_pillar_scores_year= all_pillar_yr, + country_pillar_all = cpa, + is_asean = is_asean, + ) + + records.append({ + "year": yr, + "country_id": c_id, + "country_name": c_name, + "country_name_id": c_name_id, + "pillar_id": int(row["pillar_id"]), + "pillar_name": p_name, + "pillar_name_id": p_name_id, + "pillar_score": round(p_score, 2), + "rank_in_year": p_rank, + "yoy_change": p_yoy_val, + "top_country": top_country if is_asean else None, + "top_country_id": translate_country(top_country) if (is_asean and top_country) else None, + "top_country_score": top_score if is_asean else None, + "bottom_country": bot_country if is_asean else None, + "bottom_country_id": translate_country(bot_country) if (is_asean and bot_country) else None, + "bottom_country_score": bot_score if is_asean else None, + "is_asean_aggregate": is_asean, + "narrative_en": narrative_en, + "narrative_id": narrative_id, + }) df = pd.DataFrame(records) - df["year"] = df["year"].astype(int) - df["pillar_id"] = df["pillar_id"].astype(int) - df["rank_in_year"] = df["rank_in_year"].astype(int) - df["pillar_name_id"] = df["pillar_name_id"].astype(str) - df["narrative_en"] = df["narrative_en"].astype(str) - df["narrative_id"] = df["narrative_id"].astype(str) - df["top_country_id"] = df["top_country_id"].astype(str).replace("None", pd.NA).astype("string") - df["bottom_country_id"] = df["bottom_country_id"].astype(str).replace("None", pd.NA).astype("string") + df["year"] = df["year"].astype(int) + df["country_id"] = df["country_id"].astype(int) + df["pillar_id"] = df["pillar_id"].astype(int) + df["rank_in_year"] = df["rank_in_year"].astype(int) + df["is_asean_aggregate"] = df["is_asean_aggregate"].astype(bool) + df["pillar_name_id"] = df["pillar_name_id"].astype(str) + df["country_name_id"] = df["country_name_id"].astype(str) + df["narrative_en"] = df["narrative_en"].astype(str) + df["narrative_id"] = df["narrative_id"].astype(str) for col in ["pillar_score", "yoy_change", "top_country_score", "bottom_country_score"]: df[col] = pd.to_numeric(df[col], errors="coerce").astype(float) - self.logger.info("\n Sample narrative_en (first row):") - self.logger.info(f" {df.iloc[0]['narrative_en'][:300]}") - self.logger.info("\n Sample narrative_id (first row):") - self.logger.info(f" {df.iloc[0]['narrative_id'][:300]}") + self.logger.info(f"\n Total rows: {len(df):,}") + self.logger.info(f" ASEAN rows: {df['is_asean_aggregate'].sum():,}") + self.logger.info(f" Country rows: {(~df['is_asean_aggregate']).sum():,}") + self.logger.info("\n Sample ASEAN narrative_en (first):") + asean_sample = df[df["is_asean_aggregate"]].head(1) + if not asean_sample.empty: + self.logger.info(f" {asean_sample.iloc[0]['narrative_en'][:300]}") schema = [ bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"), + bigquery.SchemaField("country_id", "INTEGER", mode="REQUIRED"), + bigquery.SchemaField("country_name", "STRING", mode="REQUIRED"), + bigquery.SchemaField("country_name_id", "STRING", mode="REQUIRED"), bigquery.SchemaField("pillar_id", "INTEGER", mode="REQUIRED"), bigquery.SchemaField("pillar_name", "STRING", mode="REQUIRED"), bigquery.SchemaField("pillar_name_id", "STRING", mode="REQUIRED"), @@ -1741,6 +1170,7 @@ class FoodSecurityAggregator: bigquery.SchemaField("bottom_country", "STRING", mode="NULLABLE"), bigquery.SchemaField("bottom_country_id", "STRING", mode="NULLABLE"), bigquery.SchemaField("bottom_country_score", "FLOAT", mode="NULLABLE"), + bigquery.SchemaField("is_asean_aggregate", "BOOL", mode="REQUIRED"), bigquery.SchemaField("narrative_en", "STRING", mode="REQUIRED"), bigquery.SchemaField("narrative_id", "STRING", mode="REQUIRED"), ] @@ -1759,22 +1189,6 @@ class FoodSecurityAggregator: # HELPERS # ========================================================================= - def _validate_mdgs_equals_total(self, df: pd.DataFrame, level: str = ""): - self.logger.info(f"\n Validasi MDGs < {self.sdgs_start_year} == Total [{level}]:") - group_by = ["year"] if level.startswith("asean") else ["country_id", "year"] - mdgs_pre = df[(df["framework"] == "MDGs") & (df["year"] < self.sdgs_start_year)][group_by + ["framework_score_1_100"]].rename(columns={"framework_score_1_100": "mdgs_score"}) - total_pre = df[(df["framework"] == "Total") & (df["year"] < self.sdgs_start_year)][group_by + ["framework_score_1_100"]].rename(columns={"framework_score_1_100": "total_score"}) - if mdgs_pre.empty and total_pre.empty: - self.logger.info(f" -> Tidak ada data pre-{self.sdgs_start_year} (skip)") - return - if mdgs_pre.empty or total_pre.empty: - self.logger.warning(f" -> [WARNING] Salah satu kosong") - return - check = mdgs_pre.merge(total_pre, on=group_by) - max_diff = (check["mdgs_score"] - check["total_score"]).abs().max() - status = "OK (identik)" if max_diff < 0.01 else f"MISMATCH! max_diff={max_diff:.6f}" - self.logger.info(f" -> {status} (n_checked={len(check)})") - def _finalize(self, table_name: str, rows_loaded: int): end_time = datetime.now() start_time = self.load_metadata[table_name].get("start_time") @@ -1828,29 +1242,22 @@ class FoodSecurityAggregator: def run(self): start = datetime.now() self.logger.info("\n" + "=" * 70) - self.logger.info("FOOD SECURITY AGGREGATION — 6 TABLES -> fs_asean_gold") + 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(" Tabel dihapus: agg_pillar_composite, agg_framework_asean,") + self.logger.info(" agg_narrative_overview") self.logger.info(f" Performance threshold: {PERFORMANCE_THRESHOLD}") self.logger.info(f" Narrative style : interpretive, plain text, bilingual EN/ID") - self.logger.info(f" FIXED : 'Access' -> 'Akses', nama negara & pilar ID") + self.logger.info(f" Sustainability : renamed to 'Food Other' (EN) / 'Indikator Tambahan' (ID)") self.logger.info("=" * 70) self.load_data() self.sdgs_start_year = self._detect_sdgs_start_year() self._assign_framework_labels() - df_pillar_composite = self.calc_pillar_composite() df_pillar_by_country = self.calc_pillar_by_country() df_framework_by_country = self.calc_framework_by_country() - df_framework_asean = self.calc_framework_asean() - - self.calc_narrative_overview( - df_framework_asean = df_framework_asean, - df_framework_by_country = df_framework_by_country, - ) - self.calc_narrative_pillar( - df_pillar_composite = df_pillar_composite, - df_pillar_by_country = df_pillar_by_country, - ) + self.calc_narrative_pillar(df_pillar_by_country=df_pillar_by_country) duration = (datetime.now() - start).total_seconds() total_rows = sum(m["rows_loaded"] for m in self.load_metadata.values()) @@ -1894,6 +1301,7 @@ if __name__ == "__main__": print("FOOD SECURITY AGGREGATION -> fs_asean_gold") print(f" NORMALIZE_FRAMEWORKS_JOINTLY : {NORMALIZE_FRAMEWORKS_JOINTLY}") print(f" PERFORMANCE_THRESHOLD : {PERFORMANCE_THRESHOLD}") + print(f" ASEAN_COUNTRY_ID : {ASEAN_COUNTRY_ID}") print("=" * 70) logger = setup_logging() diff --git a/scripts/bigquery_analytical_layer.py b/scripts/bigquery_analytical_layer.py index 50e3e41..dd94f1d 100644 --- a/scripts/bigquery_analytical_layer.py +++ b/scripts/bigquery_analytical_layer.py @@ -11,6 +11,8 @@ Filtering Order: 6. Save analytical table (dengan nama/label lengkap untuk Looker Studio) ADDED: Kolom indicator_name_id dan pillar_name_id (terjemahan Bahasa Indonesia) +CHANGED: pillar_name sekarang pakai prefix 'Food ' (Food Availability, Food Access, dst.) + 'Sustainability' -> 'Food Other', nama Indonesia: Ketersediaan Pangan, Akses Pangan, dst. """ import pandas as pd @@ -38,26 +40,36 @@ from google.cloud import bigquery # ============================================================================= # TRANSLATION DICTIONARIES +# CHANGED: "Sustainability" / "sustainability" -> "Other" / "Lainnya" # ============================================================================= PILLAR_TRANSLATION_ID: dict = { - # 4 pilar utama Food Security - "Availability" : "Ketersediaan", - "Access" : "Keterjangkauan", - "Utilization" : "Pemanfaatan", - "Stability" : "Stabilitas", - "Sustainability": "Keberlanjutan", - # Variasi penulisan yang mungkin muncul - "availability" : "Ketersediaan", - "access" : "Keterjangkauan", - "utilization" : "Pemanfaatan", - "stability" : "Stabilitas", - "sustainability": "Keberlanjutan", + # Mapping nama pilar (Inggris dengan prefix Food) -> Bahasa Indonesia "Food Availability" : "Ketersediaan Pangan", - "Food Access" : "Keterjangkauan Pangan", + "Food Access" : "Akses Pangan", "Food Utilization" : "Pemanfaatan Pangan", "Food Stability" : "Stabilitas Pangan", - "Food Sustainability": "Keberlanjutan Pangan", + "Food Other" : "Indikator Tambahan", + # Variasi tanpa prefix Food (dari data lama) + "Availability" : "Ketersediaan Pangan", + "Access" : "Akses Pangan", + "Utilization" : "Pemanfaatan Pangan", + "Stability" : "Stabilitas Pangan", + "Other" : "Indikator Tambahan", + # Legacy Sustainability -> Food Other -> Indikator Tambahan + "Sustainability" : "Indikator Tambahan", + "sustainability" : "Indikator Tambahan", + # lowercase + "food availability" : "Ketersediaan Pangan", + "food access" : "Akses Pangan", + "food utilization" : "Pemanfaatan Pangan", + "food stability" : "Stabilitas Pangan", + "food other" : "Indikator Tambahan", + "availability" : "Ketersediaan Pangan", + "access" : "Akses Pangan", + "utilization" : "Pemanfaatan Pangan", + "stability" : "Stabilitas Pangan", + "other" : "Indikator Tambahan", } @@ -194,7 +206,11 @@ def translate_indicator(name: str) -> str: def translate_pillar(name: str) -> str: - """Terjemahkan nama pillar ke Bahasa Indonesia. Fallback ke nama asli.""" + """ + Terjemahkan nama pillar ke Bahasa Indonesia. Fallback ke nama asli. + CHANGED: pillar_name menggunakan prefix 'Food ' (Food Availability, Food Access, dll.) + 'Sustainability' -> 'Food Other' (EN) / 'Indikator Tambahan' (ID). + """ if not name: return name return PILLAR_TRANSLATION_ID.get(name, name) @@ -284,6 +300,18 @@ class AnalyticalLayerLoader: self.df_clean = self.client.query(query).result().to_dataframe(create_bqstorage_client=False) self.logger.info(f" Loaded: {len(self.df_clean):,} rows") + # Rename pillar_name: add 'Food ' prefix, remove Sustainability + PILLAR_RENAME_MAP = { + 'Availability' : 'Food Availability', + 'Access' : 'Food Access', + 'Utilization' : 'Food Utilization', + 'Stability' : 'Food Stability', + 'Other' : 'Food Other', + 'Sustainability': 'Food Other', + 'sustainability': 'Food Other', + } + self.df_clean['pillar_name'] = self.df_clean['pillar_name'].replace(PILLAR_RENAME_MAP) + if 'is_year_range' in self.df_clean.columns: yr = self.df_clean['is_year_range'].value_counts() self.logger.info(f" Breakdown:") @@ -614,11 +642,7 @@ class AnalyticalLayerLoader: 'value', ]].copy() - # ------------------------------------------------------------------ - # TAMBAHAN: kolom terjemahan Bahasa Indonesia - # indicator_name_id : terjemahan Bahasa Indonesia dari indicator_name - # pillar_name_id : terjemahan Bahasa Indonesia dari pillar_name - # ------------------------------------------------------------------ + # Terjemahan Bahasa Indonesia analytical_df['indicator_name_id'] = analytical_df['indicator_name'].apply(translate_indicator) analytical_df['pillar_name_id'] = analytical_df['pillar_name'].apply(translate_pillar) @@ -701,7 +725,8 @@ class AnalyticalLayerLoader: 'fixed_countries': len(self.selected_country_ids), 'no_gaps' : True, 'layer' : 'gold', - 'columns' : 'id + name + name_id (Looker Studio ready)' + 'columns' : 'id + name + name_id (Looker Studio ready)', + 'pillar_change' : 'Sustainability -> Food Other; all pillars use Food prefix', }), 'validation_metrics' : json.dumps({ 'fixed_countries' : len(self.selected_country_ids), diff --git a/scripts/bigquery_dimensional_model.py b/scripts/bigquery_dimensional_model.py index 735f7a2..36d6467 100644 --- a/scripts/bigquery_dimensional_model.py +++ b/scripts/bigquery_dimensional_model.py @@ -10,6 +10,13 @@ Kimball ETL Flow yang dijalankan file ini: Classes: DimensionalModelLoader — Build Star Schema & load ke Gold layer +Pilar resmi FAO yang digunakan (5 pilar dengan prefix "Food "): + - Food Availability (Ketersediaan Pangan) + - Food Access (Akses Pangan) + - Food Utilization (Pemanfaatan Pangan) + - Food Stability (Stabilitas Pangan) + - Food Other (Indikator Tambahan) — additional useful statistics + Usage: python bigquery_dimensional_model.py """ @@ -37,6 +44,53 @@ if hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8') +# ============================================================================= +# PILLAR CONSTANTS +# Satu-satunya sumber kebenaran untuk nama pilar di seluruh pipeline. +# Tidak ada lagi "Sustainability" — digantikan "Food Other". +# ============================================================================= + +# Mapping dari nilai lama/raw -> nama pilar resmi (dengan prefix "Food ") +PILLAR_RENAME_MAP: dict = { + # Nilai lama tanpa prefix + 'Availability' : 'Food Availability', + 'Access' : 'Food Access', + 'Utilization' : 'Food Utilization', + 'Stability' : 'Food Stability', + 'Other' : 'Food Other', + # Nilai yang sudah benar (idempotent) + 'Food Availability': 'Food Availability', + 'Food Access' : 'Food Access', + 'Food Utilization' : 'Food Utilization', + 'Food Stability' : 'Food Stability', + 'Food Other' : 'Food Other', + # lowercase + 'food availability': 'Food Availability', + 'food access' : 'Food Access', + 'food utilization' : 'Food Utilization', + 'food stability' : 'Food Stability', + 'food other' : 'Food Other', +} + +# Kode resmi per pilar +PILLAR_CODE_MAP: dict = { + 'Food Availability': 'AVL', + 'Food Access' : 'ACC', + 'Food Utilization' : 'UTL', + 'Food Stability' : 'STB', + 'Food Other' : 'OTH', +} + +# Nama 5 pilar resmi (urutan tampilan) +OFFICIAL_PILLARS: list = [ + 'Food Availability', + 'Food Access', + 'Food Utilization', + 'Food Stability', + 'Food Other', +] + + # ============================================================================= # DIMENSIONAL MODEL LOADER # ============================================================================= @@ -62,11 +116,28 @@ class DimensionalModelLoader: def __init__(self, client: bigquery.Client, df_clean: pd.DataFrame): self.client = client - self.df_clean = df_clean + self.df_clean = df_clean.copy() self.logger = logging.getLogger(self.__class__.__name__) self.logger.propagate = False self.target_layer = 'gold' + # Normalisasi pillar column sekarang, satu kali, di awal + if 'pillar' in self.df_clean.columns: + self.df_clean['pillar'] = ( + self.df_clean['pillar'] + .astype(str) + .str.strip() + .map(lambda x: PILLAR_RENAME_MAP.get(x, 'Food Other')) + ) + unknown = set(self.df_clean['pillar'].unique()) - set(OFFICIAL_PILLARS) + if unknown: + self.logger.warning( + f" [WARN] Pillar values tidak dikenali (di-set ke 'Food Other'): {unknown}" + ) + self.df_clean['pillar'] = self.df_clean['pillar'].replace( + {u: 'Food Other' for u in unknown} + ) + self.load_metadata = { 'dim_country' : {'start_time': None, 'end_time': None, 'rows_loaded': 0, 'status': 'pending'}, 'dim_indicator' : {'start_time': None, 'end_time': None, 'rows_loaded': 0, 'status': 'pending'}, @@ -117,7 +188,7 @@ class DimensionalModelLoader: """ try: self.client.query(query).result() - self.logger.info(f" [OK] FK: {table_name}.{fk_column} → {ref_table}.{ref_column}") + self.logger.info(f" [OK] FK: {table_name}.{fk_column} -> {ref_table}.{ref_column}") except Exception as e: if "already exists" in str(e).lower(): self.logger.info(f" [INFO] FK already exists: {constraint_name}") @@ -140,12 +211,16 @@ class DimensionalModelLoader: 'rows_transformed' : meta['rows_loaded'], 'rows_loaded' : meta['rows_loaded'], 'completeness_pct' : 100.0 if meta['status'] == 'success' else 0.0, - 'config_snapshot' : json.dumps({'load_mode': 'full_refresh', 'layer': 'gold'}), + 'config_snapshot' : json.dumps({ + 'load_mode' : 'full_refresh', + 'layer' : 'gold', + 'pillar_names': OFFICIAL_PILLARS, + }), 'validation_metrics' : json.dumps({'status': meta['status'], 'rows': meta['rows_loaded']}) } try: save_etl_metadata(self.client, metadata) - self.logger.info(f" Metadata → [AUDIT] etl_metadata") + self.logger.info(f" Metadata -> [AUDIT] etl_metadata") except Exception as e: self.logger.warning(f" [WARN] Could not save metadata for {table_name}: {e}") @@ -156,13 +231,13 @@ class DimensionalModelLoader: def load_dim_time(self): table_name = 'dim_time' self.load_metadata[table_name]['start_time'] = datetime.now() - self.logger.info("Loading dim_time → [DW/Gold] fs_asean_gold...") + self.logger.info("Loading dim_time -> [DW/Gold] fs_asean_gold...") try: if 'year_range' in self.df_clean.columns: dim_time = self.df_clean[['year', 'year_range']].drop_duplicates().copy() else: - dim_time = self.df_clean[['year']].drop_duplicates().copy() + dim_time = self.df_clean[['year']].drop_duplicates().copy() dim_time['year_range'] = None dim_time['year'] = dim_time['year'].astype(int) @@ -194,10 +269,10 @@ class DimensionalModelLoader: pass return pd.Series({'year': year, 'start_year': start_year, 'end_year': end_year}) - parsed = dim_time.apply(parse_year_range, axis=1) - dim_time['year'] = parsed['year'].astype(int) - dim_time['start_year'] = parsed['start_year'].astype(int) - dim_time['end_year'] = parsed['end_year'].astype(int) + parsed = dim_time.apply(parse_year_range, axis=1) + dim_time['year'] = parsed['year'].astype(int) + dim_time['start_year'] = parsed['start_year'].astype(int) + dim_time['end_year'] = parsed['end_year'].astype(int) dim_time['is_year_range'] = (dim_time['start_year'] != dim_time['end_year']) dim_time['decade'] = (dim_time['year'] // 10) * 10 dim_time['is_range'] = (dim_time['start_year'] != dim_time['end_year']).astype(int) @@ -229,7 +304,7 @@ class DimensionalModelLoader: ) log_update(self.client, 'DW', table_name, 'full_load', rows_loaded) self._save_table_metadata(table_name) - self.logger.info(f" ✓ dim_time: {rows_loaded} rows\n") + self.logger.info(f" [OK] dim_time: {rows_loaded} rows\n") return rows_loaded except Exception as e: @@ -240,11 +315,11 @@ class DimensionalModelLoader: def load_dim_country(self): table_name = 'dim_country' self.load_metadata[table_name]['start_time'] = datetime.now() - self.logger.info("Loading dim_country → [DW/Gold] fs_asean_gold...") + self.logger.info("Loading dim_country -> [DW/Gold] fs_asean_gold...") try: - dim_country = self.df_clean[['country']].drop_duplicates().copy() - dim_country.columns = ['country_name'] + dim_country = self.df_clean[['country']].drop_duplicates().copy() + dim_country.columns = ['country_name'] region_mapping = { 'Brunei Darussalam': ('Southeast Asia', 'ASEAN'), @@ -293,7 +368,7 @@ class DimensionalModelLoader: ) log_update(self.client, 'DW', table_name, 'full_load', rows_loaded) self._save_table_metadata(table_name) - self.logger.info(f" ✓ dim_country: {rows_loaded} rows\n") + self.logger.info(f" [OK] dim_country: {rows_loaded} rows\n") return rows_loaded except Exception as e: @@ -304,27 +379,27 @@ class DimensionalModelLoader: def load_dim_indicator(self): table_name = 'dim_indicator' self.load_metadata[table_name]['start_time'] = datetime.now() - self.logger.info("Loading dim_indicator → [DW/Gold] fs_asean_gold...") + self.logger.info("Loading dim_indicator -> [DW/Gold] fs_asean_gold...") try: has_direction = 'direction' in self.df_clean.columns has_unit = 'unit' in self.df_clean.columns has_category = 'indicator_category' in self.df_clean.columns - dim_indicator = self.df_clean[['indicator_standardized']].drop_duplicates().copy() - dim_indicator.columns = ['indicator_name'] + dim_indicator = self.df_clean[['indicator_standardized']].drop_duplicates().copy() + dim_indicator.columns = ['indicator_name'] if has_unit: unit_map = self.df_clean[['indicator_standardized', 'unit']].drop_duplicates() - unit_map.columns = ['indicator_name', 'unit'] - dim_indicator = dim_indicator.merge(unit_map, on='indicator_name', how='left') + unit_map.columns = ['indicator_name', 'unit'] + dim_indicator = dim_indicator.merge(unit_map, on='indicator_name', how='left') else: dim_indicator['unit'] = None if has_direction: dir_map = self.df_clean[['indicator_standardized', 'direction']].drop_duplicates() - dir_map.columns = ['indicator_name', 'direction'] - dim_indicator = dim_indicator.merge(dir_map, on='indicator_name', how='left') + dir_map.columns = ['indicator_name', 'direction'] + dim_indicator = dim_indicator.merge(dir_map, on='indicator_name', how='left') self.logger.info(" [OK] direction column from cleaned_integrated") else: dim_indicator['direction'] = 'higher_better' @@ -332,16 +407,21 @@ class DimensionalModelLoader: if has_category: cat_map = self.df_clean[['indicator_standardized', 'indicator_category']].drop_duplicates() - cat_map.columns = ['indicator_name', 'indicator_category'] - dim_indicator = dim_indicator.merge(cat_map, on='indicator_name', how='left') + cat_map.columns = ['indicator_name', 'indicator_category'] + dim_indicator = dim_indicator.merge(cat_map, on='indicator_name', how='left') else: - def categorize_indicator(name): + # Kategorisasi otomatis — tidak ada lagi kata "Sustainability" + def categorize_indicator(name: str) -> str: n = str(name).lower() - if any(w in n for w in ['undernourishment', 'malnutrition', 'stunting', - 'wasting', 'anemia', 'food security', 'food insecure', 'hunger']): + if any(w in n for w in [ + 'undernourishment', 'malnutrition', 'stunting', + 'wasting', 'anemia', 'food security', 'food insecure', 'hunger' + ]): return 'Health & Nutrition' - elif any(w in n for w in ['production', 'yield', 'cereal', 'crop', - 'import dependency', 'share of dietary']): + elif any(w in n for w in [ + 'production', 'yield', 'cereal', 'crop', + 'import dependency', 'share of dietary' + ]): return 'Agricultural Production' elif any(w in n for w in ['import', 'export', 'trade']): return 'Trade' @@ -350,10 +430,14 @@ class DimensionalModelLoader: elif any(w in n for w in ['water', 'sanitation', 'infrastructure', 'rail']): return 'Infrastructure' else: - return 'Sustainability' - dim_indicator['indicator_category'] = dim_indicator['indicator_name'].apply(categorize_indicator) + # Fallback: "Additional Statistics" menggantikan "Sustainability" + return 'Additional Statistics' - dim_indicator = dim_indicator.drop_duplicates(subset=['indicator_name'], keep='first') + dim_indicator['indicator_category'] = ( + dim_indicator['indicator_name'].apply(categorize_indicator) + ) + + dim_indicator = dim_indicator.drop_duplicates(subset=['indicator_name'], keep='first') dim_indicator_final = dim_indicator[ ['indicator_name', 'indicator_category', 'unit', 'direction'] ].copy() @@ -384,7 +468,7 @@ class DimensionalModelLoader: ) log_update(self.client, 'DW', table_name, 'full_load', rows_loaded) self._save_table_metadata(table_name) - self.logger.info(f" ✓ dim_indicator: {rows_loaded} rows\n") + self.logger.info(f" [OK] dim_indicator: {rows_loaded} rows\n") return rows_loaded except Exception as e: @@ -395,7 +479,7 @@ class DimensionalModelLoader: def load_dim_source(self): table_name = 'dim_source' self.load_metadata[table_name]['start_time'] = datetime.now() - self.logger.info("Loading dim_source → [DW/Gold] fs_asean_gold...") + self.logger.info("Loading dim_source -> [DW/Gold] fs_asean_gold...") try: source_details = { @@ -455,7 +539,7 @@ class DimensionalModelLoader: ) log_update(self.client, 'DW', table_name, 'full_load', rows_loaded) self._save_table_metadata(table_name) - self.logger.info(f" ✓ dim_source: {rows_loaded} rows\n") + self.logger.info(f" [OK] dim_source: {rows_loaded} rows\n") return rows_loaded except Exception as e: @@ -464,26 +548,45 @@ class DimensionalModelLoader: raise def load_dim_pillar(self): + """ + Load dim_pillar dengan 5 pilar resmi FAO (prefix 'Food '). + 'Sustainability' tidak ada — digantikan 'Food Other' (Indikator Tambahan). + Pilar dibuat dari OFFICIAL_PILLARS (bukan dari data) agar selalu lengkap. + """ table_name = 'dim_pillar' self.load_metadata[table_name]['start_time'] = datetime.now() - self.logger.info("Loading dim_pillar → [DW/Gold] fs_asean_gold...") + self.logger.info("Loading dim_pillar -> [DW/Gold] fs_asean_gold...") try: - pillar_codes = { - 'Availability': 'AVL', 'Access' : 'ACC', - 'Utilization' : 'UTL', 'Stability': 'STB', 'Sustainability': 'STN', - } - pillars_data = [ - {'pillar_name': p, 'pillar_code': pillar_codes.get(p, 'STN')} - for p in self.df_clean['pillar'].unique() - ] + # Ambil pilar yang benar-benar ada di data (sudah dinormalisasi di __init__) + pillars_in_data = set(self.df_clean['pillar'].unique()) if 'pillar' in self.df_clean.columns else set() - dim_pillar_final = pd.DataFrame(pillars_data).sort_values('pillar_name')[ - ['pillar_name', 'pillar_code'] - ].copy() + # Bangun dari OFFICIAL_PILLARS — urutan tampilan konsisten + pillars_data = [] + for pillar_name in OFFICIAL_PILLARS: + code = PILLAR_CODE_MAP.get(pillar_name, 'OTH') + pillars_data.append({ + 'pillar_name': pillar_name, + 'pillar_code': code, + }) + if pillar_name not in pillars_in_data: + self.logger.warning( + f" [INFO] Pilar '{pillar_name}' tidak ada di data — " + f"tetap disertakan di dim_pillar untuk kelengkapan." + ) + + dim_pillar_final = pd.DataFrame(pillars_data)[['pillar_name', 'pillar_code']].copy() dim_pillar_final = dim_pillar_final.reset_index(drop=True) dim_pillar_final.insert(0, 'pillar_id', range(1, len(dim_pillar_final) + 1)) + self.logger.info(" Pillar list yang akan di-load:") + for _, row in dim_pillar_final.iterrows(): + in_data = "[ada di data]" if row['pillar_name'] in pillars_in_data else "[tidak ada di data]" + self.logger.info( + f" {int(row['pillar_id'])}. {row['pillar_name']:<20} " + f"({row['pillar_code']}) {in_data}" + ) + schema = [ bigquery.SchemaField("pillar_id", "INTEGER", mode="REQUIRED"), bigquery.SchemaField("pillar_name", "STRING", mode="REQUIRED"), @@ -501,7 +604,7 @@ class DimensionalModelLoader: ) log_update(self.client, 'DW', table_name, 'full_load', rows_loaded) self._save_table_metadata(table_name) - self.logger.info(f" ✓ dim_pillar: {rows_loaded} rows\n") + self.logger.info(f" [OK] dim_pillar: {rows_loaded} rows\n") return rows_loaded except Exception as e: @@ -516,7 +619,7 @@ class DimensionalModelLoader: def load_fact_food_security(self): table_name = 'fact_food_security' self.load_metadata[table_name]['start_time'] = datetime.now() - self.logger.info("Loading fact_food_security → [DW/Gold] fs_asean_gold...") + self.logger.info("Loading fact_food_security -> [DW/Gold] fs_asean_gold...") try: # Load dims dari Gold untuk FK resolution @@ -563,7 +666,8 @@ class DimensionalModelLoader: # Resolve FKs fact_table = fact_table.merge( - dim_country[['country_id', 'country_name']].rename(columns={'country_name': 'country'}), + dim_country[['country_id', 'country_name']].rename( + columns={'country_name': 'country'}), on='country', how='left' ) fact_table = fact_table.merge( @@ -576,14 +680,28 @@ class DimensionalModelLoader: on=['start_year', 'end_year'], how='left' ) fact_table = fact_table.merge( - dim_source[['source_id', 'source_name']].rename(columns={'source_name': 'source'}), + dim_source[['source_id', 'source_name']].rename( + columns={'source_name': 'source'}), on='source', how='left' ) + # pillar kolom sudah dinormalisasi ke nama resmi di __init__ fact_table = fact_table.merge( - dim_pillar[['pillar_id', 'pillar_name']].rename(columns={'pillar_name': 'pillar'}), + dim_pillar[['pillar_id', 'pillar_name']].rename( + columns={'pillar_name': 'pillar'}), on='pillar', how='left' ) + # Log FK resolution stats + n_total = len(fact_table) + n_no_pillar = fact_table['pillar_id'].isna().sum() + if n_no_pillar > 0: + self.logger.warning( + f" [WARN] {n_no_pillar}/{n_total} rows tidak dapat di-resolve pillar_id" + ) + unmatched = fact_table[fact_table['pillar_id'].isna()]['pillar'].value_counts() + for val, cnt in unmatched.items(): + self.logger.warning(f" pillar='{val}': {cnt} rows") + # Filter hanya row dengan FK lengkap fact_table = fact_table[ fact_table['country_id'].notna() & @@ -634,7 +752,7 @@ class DimensionalModelLoader: ) log_update(self.client, 'DW', table_name, 'full_load', rows_loaded) self._save_table_metadata(table_name) - self.logger.info(f" ✓ fact_food_security: {rows_loaded:,} rows\n") + self.logger.info(f" [OK] fact_food_security: {rows_loaded:,} rows\n") return rows_loaded except Exception as e: @@ -712,12 +830,28 @@ class DimensionalModelLoader: self.logger.info(f" Unique Sources : {int(stats['unique_sources']):>10,}") self.logger.info(f" Unique Pillars : {int(stats['unique_pillars']):>10,}") + # Validasi pillar — pastikan tidak ada "Sustainability" di BigQuery + query_pillar = f""" + SELECT pillar_name, pillar_code + FROM `{get_table_id('dim_pillar', layer='gold')}` + ORDER BY pillar_id + """ + df_pillar = self.client.query(query_pillar).result().to_dataframe( + create_bqstorage_client=False + ) + self.logger.info(f"\n Pillar Dimension:") + for _, row in df_pillar.iterrows(): + self.logger.info(f" [{row['pillar_code']}] {row['pillar_name']}") + + # Cek arah indikator query_dir = f""" SELECT direction, COUNT(*) AS count FROM `{get_table_id('dim_indicator', layer='gold')}` GROUP BY direction ORDER BY direction """ - df_dir = self.client.query(query_dir).result().to_dataframe(create_bqstorage_client=False) + df_dir = self.client.query(query_dir).result().to_dataframe( + create_bqstorage_client=False + ) if len(df_dir) > 0: self.logger.info(f"\n Direction Distribution:") for _, row in df_dir.iterrows(): @@ -738,11 +872,14 @@ class DimensionalModelLoader: self.pipeline_metadata['rows_fetched'] = len(self.df_clean) self.logger.info("\n" + "=" * 60) - self.logger.info("DIMENSIONAL MODEL LOAD — DW (Gold) → fs_asean_gold") + self.logger.info("DIMENSIONAL MODEL LOAD — DW (Gold) -> fs_asean_gold") self.logger.info("=" * 60) + self.logger.info(" Pilar resmi (5 pilar, prefix 'Food '):") + for p in OFFICIAL_PILLARS: + self.logger.info(f" - {p} [{PILLAR_CODE_MAP[p]}]") # Dimensions - self.logger.info("\nLOADING DIMENSION TABLES → fs_asean_gold") + self.logger.info("\nLOADING DIMENSION TABLES -> fs_asean_gold") self.load_dim_country() self.load_dim_indicator() self.load_dim_time() @@ -750,7 +887,7 @@ class DimensionalModelLoader: self.load_dim_pillar() # Fact - self.logger.info("\nLOADING FACT TABLE → fs_asean_gold") + self.logger.info("\nLOADING FACT TABLE -> fs_asean_gold") self.load_fact_food_security() # Validate @@ -762,15 +899,19 @@ class DimensionalModelLoader: total_loaded = sum(m['rows_loaded'] for m in self.load_metadata.values()) self.pipeline_metadata.update({ - 'end_time' : pipeline_end, - 'duration_seconds' : duration, - 'rows_transformed' : total_loaded, - 'rows_loaded' : total_loaded, + 'end_time' : pipeline_end, + 'duration_seconds' : duration, + 'rows_transformed' : total_loaded, + 'rows_loaded' : total_loaded, 'execution_timestamp': self.pipeline_metadata['start_time'], - 'completeness_pct' : 100.0, - 'config_snapshot' : json.dumps({'load_mode': 'full_refresh', 'layer': 'gold'}), - 'validation_metrics': json.dumps({t: m['status'] for t, m in self.load_metadata.items()}), - 'table_name' : 'dimensional_model_pipeline', + 'completeness_pct' : 100.0, + 'config_snapshot' : json.dumps({ + 'load_mode' : 'full_refresh', + 'layer' : 'gold', + 'pillar_names': OFFICIAL_PILLARS, + }), + 'validation_metrics' : json.dumps({t: m['status'] for t, m in self.load_metadata.items()}), + 'table_name' : 'dimensional_model_pipeline', }) try: save_etl_metadata(self.client, self.pipeline_metadata) @@ -785,20 +926,19 @@ class DimensionalModelLoader: self.logger.info(f" Duration : {duration:.2f}s") self.logger.info(f" Tables :") for tbl, meta in self.load_metadata.items(): - icon = "✓" if meta['status'] == 'success' else "✗" + icon = "[OK]" if meta['status'] == 'success' else "[FAIL]" self.logger.info(f" {icon} {tbl:25s}: {meta['rows_loaded']:>10,} rows") - self.logger.info(f"\n Metadata → [AUDIT] etl_metadata") + self.logger.info(f"\n Metadata -> [AUDIT] etl_metadata") self.logger.info("=" * 60) # ============================================================================= -# AIRFLOW TASK FUNCTIONS ← sama polanya dengan raw & cleaned layer +# AIRFLOW TASK FUNCTIONS # ============================================================================= def run_dimensional_model(): """ Airflow task: Load dimensional model dari cleaned_integrated. - Dipanggil oleh DAG setelah task cleaned_integration_to_silver selesai. """ from scripts.bigquery_config import get_bigquery_client @@ -817,9 +957,10 @@ if __name__ == "__main__": print("=" * 60) print("BIGQUERY DIMENSIONAL MODEL LOAD") print("Kimball DW Architecture") - print(" Input : STAGING (Silver) → cleaned_integrated (fs_asean_silver)") - print(" Output : DW (Gold) → dim_*, fact_* (fs_asean_gold)") - print(" Audit : AUDIT → etl_logs, etl_metadata (fs_asean_audit)") + print(" Input : STAGING (Silver) -> cleaned_integrated (fs_asean_silver)") + print(" Output : DW (Gold) -> dim_*, fact_* (fs_asean_gold)") + print(" Audit : AUDIT -> etl_logs, etl_metadata (fs_asean_audit)") + print(f" Pillars: {', '.join(OFFICIAL_PILLARS)}") print("=" * 60) logger = setup_logging() @@ -827,24 +968,26 @@ if __name__ == "__main__": print("\nLoading cleaned_integrated (fs_asean_silver)...") df_clean = read_from_bigquery(client, 'cleaned_integrated', layer='silver') - print(f" ✓ Loaded : {len(df_clean):,} rows") - print(f" Columns : {len(df_clean.columns)}") - print(f" Sources : {df_clean['source'].nunique()}") - print(f" Indicators : {df_clean['indicator_standardized'].nunique()}") - print(f" Countries : {df_clean['country'].nunique()}") - print(f" Year range : {int(df_clean['year'].min())}–{int(df_clean['year'].max())}") + print(f" [OK] Loaded : {len(df_clean):,} rows") + print(f" Columns : {len(df_clean.columns)}") + print(f" Sources : {df_clean['source'].nunique()}") + print(f" Indicators : {df_clean['indicator_standardized'].nunique()}") + print(f" Countries : {df_clean['country'].nunique()}") + print(f" Year range : {int(df_clean['year'].min())}-{int(df_clean['year'].max())}") + if 'pillar' in df_clean.columns: + print(f" Pillars raw : {sorted(df_clean['pillar'].unique())}") if 'direction' in df_clean.columns: - print(f" Direction : {df_clean['direction'].value_counts().to_dict()}") + print(f" Direction : {df_clean['direction'].value_counts().to_dict()}") else: - print(f" [WARN] direction column not found — run bigquery_cleaned_layer.py first") + print(f" [WARN] direction column not found — run bigquery_cleaned_layer.py first") - print("\n[1/1] Dimensional Model Load → DW (Gold)...") + print("\n[1/1] Dimensional Model Load -> DW (Gold)...") loader = DimensionalModelLoader(client, df_clean) loader.run() print("\n" + "=" * 60) - print("✓ DIMENSIONAL MODEL ETL COMPLETED") - print(" 🥇 DW (Gold) : dim_country, dim_indicator, dim_time,") - print(" dim_source, dim_pillar, fact_food_security") - print(" 📋 AUDIT : etl_logs, etl_metadata") + print("[OK] DIMENSIONAL MODEL ETL COMPLETED") + print(" DW (Gold) : dim_country, dim_indicator, dim_time,") + print(" dim_source, dim_pillar, fact_food_security") + print(" AUDIT : etl_logs, etl_metadata") print("=" * 60) \ No newline at end of file