izu/airflow-coolify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d379b2c729bad149a43ef5c92d3ede2ab378de2f
airflow-coolify/scripts/bigquery_aggregate_layer.py
T
Debby d379b2c729 add a condition for pillar
2026-06-29 10:29:44 +07:00

1527 lines
71 KiB
Python
Raw Blame History

"""
BIGQUERY ANALYSIS LAYER - FOOD SECURITY AGGREGATION
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)
KONDISI PILAR (pillar_condition_en / pillar_condition_id):
Kolom tambahan di agg_pillar_by_country untuk mendeskripsikan kondisi
tiap pilar per negara per tahun secara kontekstual dan kuantitatif.
Landasan teori:
1. FAO & CFS (1996 World Food Summit; CFS Reform Document 2009):
Definisi 4 pilar ketahanan pangan dan makna substantif masing-masing.
Referensi: FAO (2009). "Declaration of the World Summit on Food Security."
CFS (2012). "Global Strategic Framework for Food Security & Nutrition."
2. GFSI — Economist Impact (2022):
Threshold klasifikasi skor 0-100:
>= 75 : "Good" environment -> label "Secure / Aman"
>= 60 : above threshold -> label "Adequate / Memadai"
>= 40 : "Moderate" env -> label "Moderate / Sedang"
>= 20 : below moderate -> label "At Risk / Berisiko"
< 20 : severe -> label "Critical / Kritis"
Referensi: Economist Impact (2022). "Global Food Security Index 2022."
3. IPC — Integrated Food Security Phase Classification (2019):
Klasifikasi bertingkat per pilar: dari "Moderate Risk" hingga "Critical".
Referensi: IPC (2019). "IPC Technical Manual Version 3.0."
4. FAO SOFI (2023/2024):
Konteks kondisi per pilar: availability (supply/stok), access (keterjangkauan),
utilization (nutrisi/sanitasi), stability (kerentanan terhadap guncangan).
Referensi: FAO et al. (2024). "The State of Food Security and Nutrition
in the World 2024."
"""
import pandas as pd
import numpy as np
from datetime import datetime
import logging
import json
import sys as _sys
from scripts.bigquery_config import get_bigquery_client
from scripts.bigquery_helpers import (
log_update,
load_to_bigquery,
read_from_bigquery,
setup_logging,
save_etl_metadata,
)
from google.cloud import bigquery
# =============================================================================
# KONSTANTA GLOBAL
# =============================================================================
DIRECTION_INVERT_KEYWORDS = frozenset({
"negative", "lower_better", "lower_is_better", "inverse", "neg",
})
DIRECTION_POSITIVE_KEYWORDS = frozenset({
"positive", "higher_better", "higher_is_better",
})
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)",
"prevalence of severe food insecurity in the total population (percent) (3-year average)",
"prevalence of severe food insecurity in the male adult population (percent) (3-year average)",
"prevalence of severe food insecurity in the female adult population (percent) (3-year average)",
"prevalence of moderate or severe food insecurity in the total population (percent) (3-year average)",
"prevalence of moderate or severe food insecurity in the male adult population (percent) (3-year average)",
"prevalence of moderate or severe food insecurity in the female adult population (percent) (3-year average)",
"number of severely food insecure people (million) (3-year average)",
"number of severely food insecure male adults (million) (3-year average)",
"number of severely food insecure female adults (million) (3-year average)",
"number of moderately or severely food insecure people (million) (3-year average)",
"number of moderately or severely food insecure male adults (million) (3-year average)",
"number of moderately or severely food insecure female adults (million) (3-year average)",
"percentage of children under 5 years of age who are stunted (modelled estimates) (percent)",
"number of children under 5 years of age who are stunted (modeled estimates) (million)",
"percentage of children under 5 years affected by wasting (percent)",
"number of children under 5 years affected by wasting (million)",
"percentage of children under 5 years of age who are overweight (modelled estimates) (percent)",
"number of children under 5 years of age who are overweight (modeled estimates) (million)",
"prevalence of anemia among women of reproductive age (15-49 years) (percent)",
"number of women of reproductive age (15-49 years) affected by anemia (million)",
])
_SDG_ONLY_LOWER: frozenset = frozenset(k.lower() for k in SDG_ONLY_KEYWORDS)
_FIES_DETECTION_LOWER: frozenset = frozenset([
"prevalence of severe food insecurity in the total population (percent) (3-year average)",
"prevalence of moderate or severe food insecurity in the total population (percent) (3-year average)",
"number of severely food insecure people (million) (3-year average)",
"number of moderately or severely food insecure people (million) (3-year average)",
])
# =============================================================================
# TRANSLATION DICTIONARIES
# CHANGED: "Sustainability" / "sustainability" -> "Other" / "Lainnya"
# =============================================================================
COUNTRY_NAME_ID_MAP: dict = {
"Brunei Darussalam" : "Brunei Darussalam",
"Cambodia" : "Kamboja",
"Indonesia" : "Indonesia",
"Lao People's Democratic Republic" : "Laos",
"Lao PDR" : "Laos",
"Malaysia" : "Malaysia",
"Myanmar" : "Myanmar",
"Philippines" : "Filipina",
"Singapore" : "Singapura",
"Thailand" : "Thailand",
"Timor-Leste" : "Timor-Leste",
"Viet Nam" : "Vietnam",
"Vietnam" : "Vietnam",
"ASEAN" : "ASEAN",
}
PILLAR_TRANSLATION_ID: dict = {
# 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",
# 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",
}
def translate_country(name: str) -> str:
if not name:
return name
return COUNTRY_NAME_ID_MAP.get(name.strip(), name)
def translate_pillar(name: str) -> str:
if not name:
return name
return PILLAR_TRANSLATION_ID.get(name, name)
# =============================================================================
# PILLAR CONDITION CLASSIFIER
# =============================================================================
#
# Landasan teori (lihat docstring modul di atas untuk referensi lengkap):
#
# Tier skor (skala 1-100, mengacu GFSI 2022 + IPC Phase Classification):
# >= 75 : Secure / Aman — performa tinggi, kondisi baik
# >= 60 : Adequate / Memadai — di atas threshold, masih ada ruang
# >= 40 : Moderate / Sedang — tantangan nyata, perlu perhatian
# >= 20 : At Risk / Berisiko — kondisi lemah, butuh intervensi
# < 20 : Critical / Kritis — sangat buruk, tindakan segera
#
# Label kontekstual per pilar mengacu definisi FAO/CFS empat pilar:
# Food Availability : ketersediaan pasokan (produksi, stok, impor)
# Food Access : keterjangkauan ekonomi & fisik terhadap pangan
# Food Utilization : pemanfaatan biologis (gizi, sanitasi, kesehatan)
# Food Stability : konsistensi tiga pilar di atas dari waktu ke waktu
# Food Other : indikator multidimensi / suplemen
#
# =============================================================================
# Tier thresholds (urut dari tertinggi)
_CONDITION_TIERS = [
# (min_score, base_label_en, base_label_id)
(75, "Secure", "Aman"),
(60, "Adequate", "Memadai"),
(40, "Moderate", "Sedang"),
(20, "At Risk", "Berisiko"),
( 0, "Critical", "Kritis"),
]
# Konteks kondisi per pilar per tier (EN, ID)
# Mengacu makna substantif pilar (FAO SOFI 2024; FSC Handbook 2020;
# IPC Technical Manual 2019).
_PILLAR_CONTEXT: dict = {
# ---- Food Availability ----
"Food Availability": {
"Secure" : ("Food supply is abundant and well-distributed",
"Pasokan pangan berlimpah dan terdistribusi merata"),
"Adequate" : ("Food supply is sufficient with minor gaps",
"Pasokan pangan cukup dengan kesenjangan minor"),
"Moderate" : ("Food supply shows signs of strain",
"Pasokan pangan menunjukkan tanda-tanda tekanan"),
"At Risk" : ("Food supply is insufficient; stocks are dwindling",
"Pasokan pangan tidak mencukupi; stok mulai menipis"),
"Critical" : ("Severe food supply deficit; stocks critically low",
"Defisit pasokan pangan parah; stok dalam kondisi kritis"),
},
# ---- Food Access ----
"Food Access": {
"Secure" : ("Food is economically and physically accessible to all",
"Pangan terjangkau secara ekonomi dan fisik bagi semua"),
"Adequate" : ("Food access is generally good with limited barriers",
"Akses pangan umumnya baik dengan hambatan terbatas"),
"Moderate" : ("Portions of the population face access constraints",
"Sebagian penduduk menghadapi kendala akses pangan"),
"At Risk" : ("Significant affordability or physical access barriers",
"Hambatan keterjangkauan atau akses fisik yang signifikan"),
"Critical" : ("Widespread inability to access sufficient food",
"Ketidakmampuan meluas dalam mengakses pangan yang cukup"),
},
# ---- Food Utilization ----
"Food Utilization": {
"Secure" : ("Dietary quality, nutrition, and sanitation are strong",
"Kualitas gizi, nutrisi, dan sanitasi dalam kondisi baik"),
"Adequate" : ("Nutrition and sanitation are adequate; minor deficiencies",
"Gizi dan sanitasi memadai; kekurangan minor masih ada"),
"Moderate" : ("Nutritional gaps or sanitation issues are evident",
"Kesenjangan gizi atau masalah sanitasi mulai terlihat"),
"At Risk" : ("Significant nutritional deficiencies or poor sanitation",
"Kekurangan gizi atau sanitasi buruk yang signifikan"),
"Critical" : ("Severe malnutrition and/or critical sanitation deficits",
"Malnutrisi parah dan/atau defisit sanitasi yang kritis"),
},
# ---- Food Stability ----
"Food Stability": {
"Secure" : ("Food security is consistently maintained over time",
"Ketahanan pangan terjaga konsisten dari waktu ke waktu"),
"Adequate" : ("Stability is generally good with manageable risks",
"Stabilitas umumnya baik dengan risiko yang masih terkelola"),
"Moderate" : ("Periodic shocks or vulnerabilities affect stability",
"Guncangan periodik atau kerentanan memengaruhi stabilitas"),
"At Risk" : ("Frequent disruptions threaten food security continuity",
"Gangguan berulang mengancam kesinambungan ketahanan pangan"),
"Critical" : ("Sustained instability; food security is highly fragile",
"Ketidakstabilan berkelanjutan; ketahanan pangan sangat rapuh"),
},
# ---- Food Other / Indikator Tambahan ----
"Food Other": {
"Secure" : ("Supplementary indicators reflect strong food system",
"Indikator tambahan mencerminkan sistem pangan yang kuat"),
"Adequate" : ("Supplementary indicators are at acceptable levels",
"Indikator tambahan berada pada level yang dapat diterima"),
"Moderate" : ("Supplementary indicators signal emerging challenges",
"Indikator tambahan memberi sinyal tantangan yang muncul"),
"At Risk" : ("Supplementary indicators show concerning levels",
"Indikator tambahan menunjukkan level yang mengkhawatirkan"),
"Critical" : ("Supplementary indicators reflect systemic food system failure",
"Indikator tambahan mencerminkan kegagalan sistemik pangan"),
},
}
# Fallback jika pillar_name tidak dikenali
_PILLAR_CONTEXT_FALLBACK: dict = {
"Secure" : ("Performance is high across food security indicators",
"Performa tinggi pada indikator ketahanan pangan"),
"Adequate" : ("Performance is adequate across food security indicators",
"Performa memadai pada indikator ketahanan pangan"),
"Moderate" : ("Performance shows moderate challenges",
"Performa menunjukkan tantangan yang moderat"),
"At Risk" : ("Performance indicates vulnerability in food security",
"Performa mengindikasikan kerentanan ketahanan pangan"),
"Critical" : ("Performance is critically low; urgent action needed",
"Performa sangat rendah; tindakan segera diperlukan"),
}
def get_pillar_condition(pillar_name: str, score: float) -> tuple:
"""
Mengembalikan (condition_en, condition_id) berdasarkan skor dan nama pilar.
Tier mengacu GFSI 2022 (Economist Impact) + IPC Phase Classification (2019):
>= 75 -> Secure / Aman
>= 60 -> Adequate / Memadai
>= 40 -> Moderate / Sedang
>= 20 -> At Risk / Berisiko
< 20 -> Critical / Kritis
Deskripsi kontekstual mengacu FAO/CFS definisi 4 pilar (World Food Summit
1996; CFS 2009) dan FAO SOFI 2024.
Args:
pillar_name : Nama pilar dalam bahasa Inggris (e.g. "Food Availability").
score : Skor ternormalisasi skala 1-100.
Returns:
Tuple (condition_en: str, condition_id: str)
"""
if score is None or (isinstance(score, float) and np.isnan(score)):
return ("N/A", "N/A")
# Tentukan tier
tier_label_en = _CONDITION_TIERS[-1][1] # default: Critical
tier_label_id = _CONDITION_TIERS[-1][2]
for min_score, lbl_en, lbl_id in _CONDITION_TIERS:
if score >= min_score:
tier_label_en = lbl_en
tier_label_id = lbl_id
break
# Ambil konteks per pilar
ctx = _PILLAR_CONTEXT.get(pillar_name, None)
if ctx:
ctx_en, ctx_id = ctx.get(
tier_label_en,
_PILLAR_CONTEXT_FALLBACK.get(tier_label_en, ("", ""))
)
else:
ctx_en, ctx_id = _PILLAR_CONTEXT_FALLBACK.get(tier_label_en, ("", ""))
# Format akhir: "TIER — Context"
condition_en = f"{tier_label_en} — {ctx_en}"
condition_id = f"{tier_label_id} — {ctx_id}"
return condition_en, condition_id
# =============================================================================
# WINDOWS CP1252 SAFE LOGGING
# =============================================================================
class _SafeStreamHandler(logging.StreamHandler):
def emit(self, record):
try:
super().emit(record)
except UnicodeEncodeError:
try:
msg = self.format(record)
self.stream.write(
msg.encode("utf-8", errors="replace").decode("ascii", errors="replace")
+ self.terminator
)
self.flush()
except Exception:
self.handleError(record)
# =============================================================================
# HELPERS
# =============================================================================
def _should_invert(direction: str, logger=None, context: str = "") -> bool:
d = str(direction).lower().strip()
if d in DIRECTION_INVERT_KEYWORDS:
return True
if d in DIRECTION_POSITIVE_KEYWORDS:
return False
if logger:
logger.warning(
f" [DIRECTION WARNING] Unknown direction '{direction}' "
f"{'(' + context + ')' if context else ''}. Defaulting to positive (no invert)."
)
return False
def global_minmax(series: pd.Series, lo: float = 1.0, hi: float = 100.0) -> pd.Series:
values = series.dropna().values
if len(values) == 0:
return pd.Series(np.nan, index=series.index)
v_min, v_max = values.min(), values.max()
if v_min == v_max:
return pd.Series((lo + hi) / 2.0, index=series.index)
result = np.full(len(series), np.nan)
not_nan = series.notna()
raw = series[not_nan].values
result[not_nan.values] = lo + (raw - v_min) / (v_max - v_min) * (hi - lo)
return pd.Series(result, index=series.index)
def add_yoy(df: pd.DataFrame, group_cols: list, score_col: str) -> pd.DataFrame:
df = df.sort_values(group_cols + ["year"]).reset_index(drop=True)
if group_cols:
df["year_over_year_change"] = df.groupby(group_cols)[score_col].diff()
else:
df["year_over_year_change"] = df[score_col].diff()
return df
def safe_int(series: pd.Series, fill: int = 0, col_name: str = "", logger=None) -> pd.Series:
n_nan = series.isna().sum()
if n_nan > 0 and logger:
logger.warning(
f" [NaN WARNING] Kolom '{col_name}' punya {n_nan} NaN -> di-fill dengan {fill}"
)
return series.fillna(fill).astype(int)
def check_and_dedup(df: pd.DataFrame, key_cols: list, context: str = "", logger=None) -> pd.DataFrame:
dupes = df.duplicated(subset=key_cols, keep=False)
if dupes.any():
n_dupes = dupes.sum()
if logger:
logger.warning(
f" [DEDUP WARNING] {context}: {n_dupes} duplikat rows pada {key_cols}. "
f"Di-aggregate dengan mean."
)
numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
agg_dict = {
c: ("mean" if c in numeric_cols else "first")
for c in df.columns if c not in key_cols
}
df = df.groupby(key_cols, as_index=False).agg(agg_dict)
return df
def _performance_status(score) -> str:
if score is None or (isinstance(score, float) and np.isnan(score)):
return "N/A"
return "Good" if score >= PERFORMANCE_THRESHOLD else "Bad"
def _fmt_score(score) -> str:
if score is None or (isinstance(score, float) and np.isnan(score)):
return "N/A"
return f"{score:.2f}"
def _fmt_delta(delta) -> str:
if delta is None or (isinstance(delta, float) and np.isnan(delta)):
return "N/A"
sign = "+" if delta >= 0 else ""
return f"{sign}{delta:.2f}"
# =============================================================================
# NARRATIVE CONDITION DETECTORS (shared)
# =============================================================================
def _detect_series_trend(scores: list) -> str:
if len(scores) < 3:
return "insufficient"
x = np.arange(len(scores))
slope = np.polyfit(x, scores, 1)[0]
cv = np.std(scores) / (np.mean(scores) + 1e-9)
if cv > 0.20:
return "fluctuating"
mid = len(scores) // 2
slope1 = np.polyfit(np.arange(mid), scores[:mid], 1)[0] if mid > 1 else slope
slope2 = np.polyfit(np.arange(len(scores) - mid), scores[mid:], 1)[0] if (len(scores) - mid) > 1 else slope
if slope > 0:
slowing = slope2 < slope1
return "improving_slowing" if slowing else "improving_consistent"
else:
return "deteriorating"
def _detect_country_gap(scores_by_country_year: pd.DataFrame, score_col: str) -> str:
std_by_year = (
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:
return "unknown"
years = sorted(std_by_year.index)
stds = [std_by_year[y] for y in years]
slope = np.polyfit(np.arange(len(stds)), stds, 1)[0]
mean_s = np.mean(stds)
if abs(slope) < 0.02 * mean_s:
return "stable"
return "widening" if slope > 0 else "narrowing"
def _find_anomaly_year(values_by_year: dict) -> tuple:
years = sorted(values_by_year.keys())
deltas = {}
for i in range(1, len(years)):
y0, y1 = years[i-1], years[i]
v0, v1 = values_by_year.get(y0), values_by_year.get(y1)
if v0 is not None and v1 is not None and not (pd.isna(v0) or pd.isna(v1)):
deltas[y1] = v1 - v0
if not deltas:
return None, None
threshold = 1.5 * np.std(list(deltas.values()))
min_y = min(deltas, key=deltas.get)
max_y = max(deltas, key=deltas.get)
if abs(deltas[min_y]) > threshold and deltas[min_y] < 0:
return min_y, "drop"
if abs(deltas[max_y]) > threshold and deltas[max_y] > 0:
return max_y, "rise"
return None, None
# =============================================================================
# NARRATIVE BUILDER — PILLAR
# =============================================================================
def _build_pillar_narrative(
year: int,
pillar_name: str,
pillar_score: float,
rank_in_year: int,
n_pillars: int,
yoy_val,
top_country: str,
top_country_score,
bot_country: str,
bot_country_score,
pillar_scores_history: dict,
all_pillar_scores_year: pd.DataFrame,
country_pillar_all: pd.DataFrame,
is_asean: bool = False,
) -> tuple:
sentences_en = []
sentences_id = []
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})."
)
s1_id = (
f"Pada tahun {year}, pilar {pillar_name_id} menempati peringkat {rank_in_year} dari "
f"{n_pillars} pilar dengan skor {_fmt_score(pillar_score)} ({perf_word_id})."
)
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 = "Performance was relatively stable compared to the previous year."
s2_id = "Performa relatif stabil dibandingkan tahun sebelumnya."
elif yoy_val > 0:
s2_en = f"This is an improvement of {abs(yoy_val):.2f} points from the previous year."
s2_id = f"Ini merupakan peningkatan {abs(yoy_val):.2f} poin dari tahun sebelumnya."
else:
s2_en = f"This marks a decline of {abs(yoy_val):.2f} points from the previous year."
s2_id = f"Ini menandai penurunan {abs(yoy_val):.2f} poin dari tahun sebelumnya."
sentences_en.append(s2_en)
sentences_id.append(s2_id)
hist_years = sorted(pillar_scores_history.keys())
hist_scores = [
pillar_scores_history[y]
for y in hist_years
if not pd.isna(pillar_scores_history.get(y, np.nan))
]
if len(hist_scores) >= 3:
trend = _detect_series_trend(hist_scores)
if trend == "improving_consistent":
s3_en = f"This pillar has shown consistent improvement since {hist_years[0]}."
s3_id = f"Pilar {pillar_name_id} menunjukkan perbaikan yang konsisten sejak {hist_years[0]}."
elif trend == "improving_slowing":
s3_en = f"While the pillar improved since {hist_years[0]}, the pace has slowed in recent years."
s3_id = f"Meskipun pilar {pillar_name_id} membaik sejak {hist_years[0]}, lajunya melambat dalam beberapa tahun terakhir."
elif trend == "deteriorating":
s3_en = f"This pillar has shown a declining trend since {hist_years[0]}, requiring targeted intervention."
s3_id = f"Pilar {pillar_name_id} menunjukkan tren penurunan sejak {hist_years[0]}, memerlukan intervensi yang terarah."
elif trend == "fluctuating":
s3_en = f"Performance in this pillar has been inconsistent since {hist_years[0]}, with no clear trend."
s3_id = f"Performa pilar {pillar_name_id} tidak konsisten sejak {hist_years[0]}, tanpa tren yang jelas."
else:
s3_en = ""
s3_id = ""
if s3_en:
sentences_en.append(s3_en)
sentences_id.append(s3_id)
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"
)
if gap_trend == "widening":
s4_en = "Country disparities within this pillar have widened over time."
s4_id = f"Kesenjangan antar negara dalam pilar {pillar_name_id} semakin melebar seiring waktu."
elif gap_trend == "narrowing":
s4_en = "Country disparities within this pillar have narrowed, indicating more balanced progress."
s4_id = f"Kesenjangan antar negara dalam pilar {pillar_name_id} menyempit, mengindikasikan kemajuan yang lebih merata."
else:
s4_en = ""
s4_id = ""
if s4_en:
sentences_en.append(s4_en)
sentences_id.append(s4_id)
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)})."
)
s5_id = (
f"{top_country_id} memiliki performa terbaik dalam pilar {pillar_name_id} "
f"({_fmt_score(top_country_score)}), "
f"sementara {bot_country_id} memiliki skor terendah ({_fmt_score(bot_country_score)})."
)
sentences_en.append(s5_en)
sentences_id.append(s5_id)
if not all_pillar_scores_year.empty and len(all_pillar_scores_year) > 1:
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_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_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)
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
# =============================================================================
# MAIN CLASS
# =============================================================================
class FoodSecurityAggregator:
def __init__(self, client: bigquery.Client):
self.client = client
self.logger = logging.getLogger(self.__class__.__name__)
self.logger.propagate = False
self.load_metadata = {
"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_narrative_pillar": {"rows_loaded": 0, "status": "pending", "start_time": None, "end_time": None},
}
self.df = None
self.sdgs_start_year = None
self._ind_year_framework: pd.DataFrame = None
# =========================================================================
# STEP 1: Load data
# =========================================================================
def load_data(self):
self.logger.info("=" * 70)
self.logger.info("STEP 1: LOAD DATA from fs_asean_gold")
self.logger.info("=" * 70)
self.df = read_from_bigquery(
self.client, "fact_asean_food_security_selected", layer='gold'
)
self.logger.info(f" fact_asean_food_security_selected : {len(self.df):,} rows")
required_cols = {
"country_id", "country_name",
"indicator_id", "indicator_name", "direction",
"pillar_id", "pillar_name",
"time_id", "year", "value",
}
missing_cols = required_cols - set(self.df.columns)
if missing_cols:
raise ValueError(f"Kolom berikut tidak ditemukan: {missing_cols}")
n_null_dir = self.df["direction"].isna().sum()
if n_null_dir > 0:
self.logger.warning(f" [DIRECTION] {n_null_dir} rows NULL -> diisi 'positive'")
self.df["direction"] = self.df["direction"].fillna("positive")
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)
if "country_name_id" not in self.df.columns:
self.df["country_name_id"] = self.df["country_name"].apply(translate_country)
if "pillar_name_id" not in self.df.columns:
self.df["pillar_name_id"] = self.df["pillar_name"].apply(translate_pillar)
self.logger.info(f"\n Rows : {len(self.df):,}")
self.logger.info(f" Countries : {self.df['country_id'].nunique()}")
self.logger.info(f" Indicators: {self.df['indicator_id'].nunique()}")
self.logger.info(
f" Years : {int(self.df['year'].min())} - {int(self.df['year'].max())}"
)
# =========================================================================
# STEP 1b: Detect sdgs_start_year + assign framework
# =========================================================================
def _detect_sdgs_start_year(self) -> int:
fies_rows = self.df[
self.df["indicator_name"].str.lower().str.strip().isin(_FIES_DETECTION_LOWER)
]
if not fies_rows.empty:
sdgs_start = int(fies_rows["year"].min())
self.logger.info(f" [FIES explicit] sdgs_start_year = {sdgs_start}")
return sdgs_start
ind_min_year = (
self.df.groupby("indicator_id")["year"]
.min().reset_index().rename(columns={"year": "min_year"})
)
unique_years = sorted(ind_min_year["min_year"].unique())
if len(unique_years) == 1:
return int(unique_years[0]) + 9999
gaps = [
(unique_years[i+1] - unique_years[i], unique_years[i], unique_years[i+1])
for i in range(len(unique_years) - 1)
]
gaps.sort(reverse=True)
_, y_before, y_after = gaps[0]
self.logger.info(f" [Fallback gap] sdgs_start_year = {y_after}")
return int(y_after)
def _assign_framework_labels(self):
self.logger.info("\n" + "=" * 70)
self.logger.info("STEP 1b: ASSIGN FRAMEWORK LABELS")
self.logger.info(f" sdgs_start_year = {self.sdgs_start_year}")
self.logger.info("=" * 70)
df = self.df.copy()
df["_is_sdg_kw"] = df["indicator_name"].str.lower().str.strip().isin(_SDG_ONLY_LOWER)
df["framework"] = "MDGs"
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
self._ind_year_framework = (
self.df[["indicator_id", "year", "framework"]]
.drop_duplicates()
.reset_index(drop=True)
)
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")
def _count_framework_indicators(self, year: int, framework: str) -> int:
mask = (
(self._ind_year_framework["year"] == year) &
(self._ind_year_framework["framework"] == framework)
)
return int(self._ind_year_framework.loc[mask, "indicator_id"].nunique())
# =========================================================================
# CORE HELPER: normalisasi raw value per indikator
# =========================================================================
def _get_norm_value_df(self) -> pd.DataFrame:
if "framework" not in self.df.columns:
raise ValueError("Kolom 'framework' tidak ada.")
norm_parts = []
for ind_id, grp in self.df.groupby("indicator_id"):
grp = grp.copy()
direction = str(grp["direction"].iloc[0])
do_invert = _should_invert(direction, self.logger, context=f"indicator_id={ind_id}")
valid_mask = grp["value"].notna()
n_valid = valid_mask.sum()
if n_valid < 2:
grp["norm_value"] = np.nan
norm_parts.append(grp)
continue
raw = grp.loc[valid_mask, "value"].values
v_min, v_max = raw.min(), raw.max()
normed = np.full(len(grp), np.nan)
if v_min == v_max:
normed[valid_mask.values] = 0.5
else:
normed[valid_mask.values] = (raw - v_min) / (v_max - v_min)
if do_invert:
normed = np.where(np.isnan(normed), np.nan, 1.0 - normed)
grp["norm_value"] = normed
norm_parts.append(grp)
return pd.concat(norm_parts, ignore_index=True)
# =========================================================================
# METADATA BUILDER
# =========================================================================
def _build_etl_metadata(
self,
table_name: str,
rows_loaded: int,
start_time: datetime,
end_time: datetime,
status: str,
error_msg: str = None,
) -> dict:
duration = (end_time - start_time).total_seconds() if (start_time and end_time) else 0.0
return {
"source_class" : "FoodSecurityAggregator",
"table_name" : table_name,
"execution_timestamp": start_time or end_time,
"duration_seconds" : round(duration, 4),
"rows_fetched" : rows_loaded,
"rows_transformed" : rows_loaded,
"rows_loaded" : rows_loaded,
"completeness_pct" : 100.0 if status == "success" else 0.0,
"config_snapshot" : json.dumps({
"layer" : "gold",
"write_disposition" : "WRITE_TRUNCATE",
"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)",
"condition_column" : "pillar_condition_en/id added to agg_pillar_by_country",
"condition_reference" : (
"GFSI 2022 (Economist Impact) score tiers >= 75/60/40/20; "
"IPC Technical Manual 2019; FAO/CFS 4-pillar framework 1996/2009; "
"FAO SOFI 2024"
),
}),
"validation_metrics" : json.dumps({
"status" : status,
"error_msg": error_msg or "",
}),
}
# =========================================================================
# HELPER: build ASEAN rows untuk tabel pillar_by_country
# =========================================================================
def _build_asean_pillar_rows(self, df_normed: pd.DataFrame) -> pd.DataFrame:
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 2: agg_pillar_by_country (termasuk ASEAN + kolom kondisi)
# =========================================================================
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 2: {table_name} -> [Gold] fs_asean_gold")
self.logger.info(" Termasuk baris ASEAN (country_id=0) untuk filter Looker Studio")
self.logger.info(" Kolom baru: pillar_condition_en, pillar_condition_id")
self.logger.info("=" * 70)
try:
df_normed = self._get_norm_value_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
df["pillar_country_score_1_100"] = global_minmax(df["pillar_country_norm"])
# ---------------------------------------------------------------
# TAMBAHAN: kolom kondisi pilar
# Dibangkitkan SETELAH score_1_100 tersedia, sehingga tier
# langsung mencerminkan skor dalam skala akhir 1-100.
# Referensi tier: GFSI 2022 (Economist Impact); IPC 2019;
# FAO/CFS 1996/2009; FAO SOFI 2024.
# ---------------------------------------------------------------
conditions = df.apply(
lambda row: get_pillar_condition(
row["pillar_name"],
row["pillar_country_score_1_100"]
),
axis=1
)
df["pillar_condition_en"] = conditions.apply(lambda x: x[0])
df["pillar_condition_id"] = conditions.apply(lambda x: x[1])
# Rank hanya di antara negara asli
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")
# Tipe data
df["country_id"] = df["country_id"].astype(int)
df["pillar_id"] = df["pillar_id"].astype(int)
df["year"] = df["year"].astype(int)
df["rank_in_pillar_year"] = df["rank_in_pillar_year"].astype(int)
df["pillar_country_norm"] = df["pillar_country_norm"].astype(float)
df["pillar_country_score_1_100"] = df["pillar_country_score_1_100"].astype(float)
df["pillar_name_id"] = df["pillar_name_id"].astype(str)
df["country_name_id"] = df["country_name_id"].astype(str)
df["pillar_condition_en"] = df["pillar_condition_en"].astype(str)
df["pillar_condition_id"] = df["pillar_condition_id"].astype(str)
self.logger.info(
f" Total rows: {len(df):,} "
f"({len(df_countries):,} country + {len(asean_only):,} ASEAN)"
)
# Log distribusi kondisi untuk QA
self.logger.info("\n Distribusi pillar_condition_en (sample):")
cond_dist = df["pillar_condition_en"].value_counts().head(10)
for cond, cnt in cond_dist.items():
self.logger.info(f" {cnt:>6,} {cond}")
schema = [
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"),
bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("pillar_country_norm", "FLOAT", mode="REQUIRED"),
bigquery.SchemaField("pillar_country_score_1_100", "FLOAT", mode="REQUIRED"),
bigquery.SchemaField("rank_in_pillar_year", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("year_over_year_change", "FLOAT", mode="NULLABLE"),
# --- KOLOM KONDISI BARU ---
# Tier skor (GFSI 2022) + konteks substantif per pilar (FAO/CFS; IPC 2019)
bigquery.SchemaField("pillar_condition_en", "STRING", mode="REQUIRED"),
bigquery.SchemaField("pillar_condition_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
# =========================================================================
# HELPER: composite per country (untuk framework_by_country)
# =========================================================================
def _calc_country_composite_inmemory(self) -> pd.DataFrame:
df_normed = self._get_norm_value_df()
df = (
df_normed
.groupby(["country_id", "country_name", "year"])
.agg(
composite_score=("norm_value", "mean"),
n_indicators =("indicator_id", "nunique"),
)
.reset_index()
)
df["score_1_100"] = global_minmax(df["composite_score"])
df["rank_in_asean"] = (
df.groupby("year")["score_1_100"]
.rank(method="min", ascending=False)
.astype(int)
)
df = add_yoy(df, ["country_id"], "score_1_100")
df["country_id"] = df["country_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["composite_score"] = df["composite_score"].astype(float)
df["score_1_100"] = df["score_1_100"].astype(float)
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 3: {table_name} -> [Gold] fs_asean_gold")
self.logger.info(" Termasuk baris ASEAN (country_id=0)")
self.logger.info("=" * 70)
try:
country_composite = self._calc_country_composite_inmemory()
df_normed = self._get_norm_value_df()
parts = []
# ---- Per negara (Total) ----
agg_total = (
country_composite[[
"country_id", "country_name", "year",
"score_1_100", "n_indicators", "composite_score"
]]
.copy()
.rename(columns={
"score_1_100" : "framework_score_1_100",
"composite_score": "framework_norm",
})
)
agg_total["framework"] = "Total"
parts.append(agg_total)
pre_sdgs_rows = country_composite[country_composite["year"] < self.sdgs_start_year].copy()
if not pre_sdgs_rows.empty:
mdgs_pre = (
pre_sdgs_rows[[
"country_id", "country_name", "year",
"score_1_100", "n_indicators", "composite_score"
]]
.copy()
.rename(columns={
"score_1_100" : "framework_score_1_100",
"composite_score": "framework_norm",
})
)
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:
agg_mdgs_mixed = (
df_mdgs_mixed
.groupby(["country_id", "country_name", "year"])
.agg(
framework_norm=("norm_value", "mean"),
n_indicators =("indicator_id", "nunique"),
)
.reset_index()
)
if not NORMALIZE_FRAMEWORKS_JOINTLY:
agg_mdgs_mixed["framework_score_1_100"] = global_minmax(agg_mdgs_mixed["framework_norm"])
agg_mdgs_mixed["framework"] = "MDGs"
parts.append(agg_mdgs_mixed)
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:
agg_sdgs = (
df_sdgs
.groupby(["country_id", "country_name", "year"])
.agg(
framework_norm=("norm_value", "mean"),
n_indicators =("indicator_id", "nunique"),
)
.reset_index()
)
if not NORMALIZE_FRAMEWORKS_JOINTLY:
agg_sdgs["framework_score_1_100"] = global_minmax(agg_sdgs["framework_norm"])
agg_sdgs["framework"] = "SDGs"
parts.append(agg_sdgs)
df_countries = pd.concat(parts, ignore_index=True)
# ---- ASEAN aggregate ----
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)
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, ["country_id", "framework", "year"], context=table_name, logger=self.logger)
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)
df["year"] = df["year"].astype(int)
df["n_indicators"] = safe_int(df["n_indicators"], col_name="n_indicators", logger=self.logger)
df["rank_in_framework_year"] = safe_int(df["rank_in_framework_year"], col_name="rank_in_framework_year", logger=self.logger)
df["framework_norm"] = df["framework_norm"].astype(float)
df["framework_score_1_100"] = df["framework_score_1_100"].astype(float)
df["country_name_id"] = df["country_name_id"].astype(str)
schema = [
bigquery.SchemaField("country_id", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("country_name", "STRING", mode="REQUIRED"),
bigquery.SchemaField("country_name_id", "STRING", mode="REQUIRED"),
bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("framework", "STRING", mode="REQUIRED"),
bigquery.SchemaField("n_indicators", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("framework_norm", "FLOAT", mode="REQUIRED"),
bigquery.SchemaField("framework_score_1_100", "FLOAT", mode="REQUIRED"),
bigquery.SchemaField("rank_in_framework_year", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("year_over_year_change", "FLOAT", mode="NULLABLE"),
]
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 4: agg_narrative_pillar (termasuk baris ASEAN)
# =========================================================================
def calc_narrative_pillar(
self,
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 4: {table_name} -> [Gold] fs_asean_gold")
self.logger.info(" Termasuk baris ASEAN (country_id=0)")
self.logger.info("=" * 70)
try:
records = []
years = sorted(df_pillar_by_country["year"].unique())
pillars = df_pillar_by_country["pillar_id"].unique()
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_df = df_pillar_by_country[df_pillar_by_country["year"] == yr]
country_only_yr = yr_df[yr_df["country_id"] != ASEAN_COUNTRY_ID]
for p_id in pillars:
yr_pillar_all = yr_df[yr_df["pillar_id"] == p_id]
if yr_pillar_all.empty:
continue
p_name_row = yr_pillar_all.iloc[0]
p_name = str(p_name_row["pillar_name"])
n_pillars = len(pillars)
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)
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_score = bot_score = None
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)
if is_asean:
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:
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
hist_up = {y: s for y, s in history.get((c_id, p_id), {}).items() if y <= yr}
all_pillar_yr = yr_df[yr_df["country_id"] == c_id][["pillar_name", "pillar_country_score_1_100"]].copy()
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,
)
# Ambil kondisi dari kolom yang sudah dihitung di df_pillar_by_country
cond_en = str(row.get("pillar_condition_en", "N/A"))
cond_id = str(row.get("pillar_condition_id", "N/A"))
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,
"pillar_condition_en": cond_en,
"pillar_condition_id": cond_id,
"narrative_en": narrative_en,
"narrative_id": narrative_id,
})
df = pd.DataFrame(records)
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["pillar_condition_en"] = df["pillar_condition_en"].astype(str)
df["pillar_condition_id"] = df["pillar_condition_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(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():,}")
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"),
bigquery.SchemaField("pillar_score", "FLOAT", mode="REQUIRED"),
bigquery.SchemaField("rank_in_year", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("yoy_change", "FLOAT", mode="NULLABLE"),
bigquery.SchemaField("top_country", "STRING", mode="NULLABLE"),
bigquery.SchemaField("top_country_id", "STRING", mode="NULLABLE"),
bigquery.SchemaField("top_country_score", "FLOAT", mode="NULLABLE"),
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("pillar_condition_en", "STRING", mode="REQUIRED"),
bigquery.SchemaField("pillar_condition_id", "STRING", mode="REQUIRED"),
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
# =========================================================================
# HELPERS
# =========================================================================
def _finalize(self, table_name: str, rows_loaded: int):
end_time = datetime.now()
start_time = self.load_metadata[table_name].get("start_time")
self.load_metadata[table_name].update({
"rows_loaded": rows_loaded,
"status" : "success",
"end_time" : end_time,
})
log_update(self.client, "DW", table_name, "full_load", rows_loaded)
try:
save_etl_metadata(
self.client,
self._build_etl_metadata(
table_name = table_name,
rows_loaded = rows_loaded,
start_time = start_time,
end_time = end_time,
status = "success",
)
)
except Exception as meta_err:
self.logger.warning(f" [METADATA WARNING] {table_name}: {meta_err}")
self.logger.info(f" [OK] {table_name}: {rows_loaded:,} rows -> [Gold] fs_asean_gold")
def _fail(self, table_name: str, error: Exception):
end_time = datetime.now()
start_time = self.load_metadata[table_name].get("start_time")
error_msg = str(error)
self.load_metadata[table_name].update({"status": "failed", "end_time": end_time})
log_update(self.client, "DW", table_name, "full_load", 0, "failed", error_msg)
try:
save_etl_metadata(
self.client,
self._build_etl_metadata(
table_name = table_name,
rows_loaded = 0,
start_time = start_time,
end_time = end_time,
status = "failed",
error_msg = error_msg,
)
)
except Exception as meta_err:
self.logger.warning(f" [METADATA WARNING] {table_name}: {meta_err}")
self.logger.error(f" [FAIL] {table_name}: {error_msg}")
# =========================================================================
# RUN
# =========================================================================
def run(self):
start = datetime.now()
self.logger.info("\n" + "=" * 70)
self.logger.info("FOOD SECURITY AGGREGATION — 3 TABLES -> fs_asean_gold")
self.logger.info(" ASEAN aggregate DIGABUNG ke tabel yang sama (country_id=0)")
self.logger.info(" Kolom baru : pillar_condition_en, pillar_condition_id")
self.logger.info(f" Performance threshold: {PERFORMANCE_THRESHOLD}")
self.logger.info(f" Condition tiers (GFSI 2022): >=75 Secure | >=60 Adequate |")
self.logger.info(f" >=40 Moderate | >=20 At Risk | <20 Critical")
self.logger.info(f" Narrative style : interpretive, plain text, bilingual EN/ID")
self.logger.info(f" Sustainability : renamed to 'Food Other' / 'Indikator Tambahan'")
self.logger.info("=" * 70)
self.load_data()
self.sdgs_start_year = self._detect_sdgs_start_year()
self._assign_framework_labels()
df_pillar_by_country = self.calc_pillar_by_country()
df_framework_by_country = self.calc_framework_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())
self.logger.info("\n" + "=" * 70)
self.logger.info("SELESAI")
self.logger.info("=" * 70)
self.logger.info(f" Durasi : {duration:.2f}s")
self.logger.info(f" Total rows : {total_rows:,}")
for tbl, meta in self.load_metadata.items():
icon = "[OK]" if meta["status"] == "success" else "[FAIL]"
self.logger.info(f" {icon} {tbl:<35} {meta['rows_loaded']:>10,}")
# =============================================================================
# AIRFLOW TASK
# =============================================================================
def run_aggregation():
from scripts.bigquery_config import get_bigquery_client
client = get_bigquery_client()
agg = FoodSecurityAggregator(client)
agg.run()
total = sum(m["rows_loaded"] for m in agg.load_metadata.values())
print(f"Aggregation completed: {total:,} total rows loaded")
# =============================================================================
# MAIN
# =============================================================================
if __name__ == "__main__":
import io
if _sys.stdout.encoding and _sys.stdout.encoding.lower() not in ("utf-8", "utf8"):
_sys.stdout = io.TextIOWrapper(_sys.stdout.buffer, encoding="utf-8", errors="replace")
if _sys.stderr.encoding and _sys.stderr.encoding.lower() not in ("utf-8", "utf8"):
_sys.stderr = io.TextIOWrapper(_sys.stderr.buffer, encoding="utf-8", errors="replace")
print("=" * 70)
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(f" Condition tiers (GFSI 2022) : >=75 Secure | >=60 Adequate | >=40 Moderate | >=20 At Risk | <20 Critical")
print("=" * 70)
logger = setup_logging()
for handler in logger.handlers:
handler.__class__ = _SafeStreamHandler
client = get_bigquery_client()
agg = FoodSecurityAggregator(client)
agg.run()
print("\n" + "=" * 70)
print("[OK] SELESAI")
print("=" * 70)
Reference in New Issue View Git Blame Copy Permalink