izu/airflow-coolify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

country name indo version

Browse Source
This commit is contained in:
Debby
2026-06-07 08:09:14 +07:00
parent 512c08bff3
commit ca1e0d3949
3 changed files with 344 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/bigquery_aggraget_fact_selected_layer.py
+129 -48
View File
@@ -5,9 +5,12 @@ Tabel 2: agg_narrative_indicator -> fs_asean_gold
=============================================================================
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
- Kedua kolom ikut tersimpan di BigQuery (schema + DataFrame output)
- 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)
=============================================================================
agg_indicator_norm
@@ -35,7 +38,7 @@ 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,
year, country_id, country_name, country_name_id,
indicator_id, indicator_name, indicator_name_id,
unit, direction,
pillar_id, pillar_name, pillar_name_id,
@@ -54,6 +57,8 @@ 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:
@@ -71,6 +76,7 @@ Output Schema (agg_narrative_indicator):
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
"""
@@ -96,7 +102,25 @@ from google.cloud import bigquery
# MAPPING BAHASA INDONESIA
# =============================================================================
# Mapping nama negara (Inggris -> Indonesia)
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",
}
# Mapping nama pilar (Inggris -> Indonesia)
# FIXED: "Access" -> "Akses" (bukan "Keterjangkauan")
PILLAR_NAME_ID_MAP: dict = {
"Availability" : "Ketersediaan",
"Access" : "Akses",
@@ -189,8 +213,6 @@ INDICATOR_NAME_ID_MAP: dict = {
"Stabilitas politik dan ketiadaan kekerasan/terorisme",
"domestic food price volatility index":
"Indeks volatilitas harga pangan domestik",
"per capita food supply variability (kcal/capita/day)":
"Variabilitas pasokan pangan per kapita (kkal/kapita/hari)",
"cereal import dependency ratio (percent) (3-year average)":
"Rasio ketergantungan impor sereal (persen) (rata-rata 3 tahun)",
"value of food imports in total merchandise exports (percent) (3-year average)":
@@ -200,11 +222,19 @@ 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
)
def get_indicator_name_id(indicator_name: str) -> str:
"""Kembalikan terjemahan Bahasa Indonesia untuk nama indikator."""
return INDICATOR_NAME_ID_MAP.get(
str(indicator_name).lower().strip(),
str(indicator_name), # fallback: kembalikan nama asli jika tidak ada mapping
str(indicator_name), # fallback: kembalikan nama asli
)
@@ -212,7 +242,7 @@ 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 jika tidak ada mapping
str(pillar_name), # fallback: kembalikan nama asli
)
@@ -406,6 +436,11 @@ 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.
"""
country_avg = (
df_ind.groupby("country_name")["value"]
.mean()
@@ -446,34 +481,42 @@ 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
# =============================================================================
def _build_narrative_per_indicator(row: pd.Series, df_full: pd.DataFrame) -> tuple:
ind_id = int(row["indicator_id"])
ind_name = str(row["indicator_name"]).strip()
unit = str(row["unit"]).strip() if row["unit"] else ""
direction = str(row["direction"]).strip()
pillar = str(row["pillar_name"]).strip()
framework = str(row["framework"]).strip()
year_min = int(row["year_min"])
year_max = int(row["year_max"])
ind_id = int(row["indicator_id"])
ind_name_en = str(row["indicator_name"]).strip()
ind_name_id = str(row.get("indicator_name_id", ind_name_en)).strip()
unit = str(row["unit"]).strip() if row["unit"] else ""
direction = str(row["direction"]).strip()
pillar_en = str(row["pillar_name"]).strip()
pillar_id_ = get_pillar_name_id(pillar_en) # nama pilar dalam Bahasa Indonesia
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()
if df_ind.empty:
na_en = f"{ind_name} ({framework}, {pillar}): Insufficient data for analysis."
na_id = f"{ind_name} ({framework}, {pillar}): Data tidak cukup untuk dianalisis."
na_en = f"{ind_name_en} ({framework}, {pillar_en}): Insufficient data for analysis."
na_id = f"{ind_name_id} ({framework}, {pillar_id_}): Data tidak cukup untuk dianalisis."
return na_en, na_id
asean_avg_by_year = (
df_ind.groupby("year")["value"].mean().dropna()
)
trend_label = _detect_trend(asean_avg_by_year, lower_better)
gap_label = _detect_gap_trend(df_ind, lower_better)
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, is_consistent = _detect_consistency(df_ind, lower_better)
# 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
avg_first = row.get("avg_value_first", np.nan)
avg_last = row.get("avg_value_last", np.nan)
@@ -488,8 +531,9 @@ def _build_narrative_per_indicator(row: pd.Series, df_full: pd.DataFrame) -> tup
sentences_en = []
sentences_id = []
s1_en = f"{ind_name} ({framework}, {pillar}, {year_min}-{year_max}):"
s1_id = f"{ind_name} ({framework}, {pillar}, {year_min}-{year_max}):"
# 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)
sentences_id.append(s1_id)
@@ -528,24 +572,25 @@ 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.")
if best_country and worst_country:
# 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(
f"{best_country} consistently performed above the regional average, "
f"while {worst_country} consistently lagged behind."
f"{best_country_en} consistently performed above the regional average, "
f"while {worst_country_en} consistently lagged behind."
)
sentences_id.append(
f"{best_country} secara konsisten berada di atas rata-rata regional, "
f"sementara {worst_country} secara konsisten tertinggal."
f"{best_country_id} secara konsisten berada di atas rata-rata regional, "
f"sementara {worst_country_id} secara konsisten tertinggal."
)
else:
sentences_en.append(
f"Overall, {best_country} showed the best performance, "
f"while {worst_country} had the weakest results across the period."
f"Overall, {best_country_en} showed the best performance, "
f"while {worst_country_en} had the weakest results across the period."
)
sentences_id.append(
f"Secara keseluruhan, {best_country} menunjukkan performa terbaik, "
f"sementara {worst_country} memiliki hasil terlemah sepanjang periode."
f"Secara keseluruhan, {best_country_id} menunjukkan performa terbaik, "
f"sementara {worst_country_id} memiliki hasil terlemah sepanjang periode."
)
narrative_en = " ".join(s for s in sentences_en if s)
@@ -689,23 +734,54 @@ class IndicatorNormAggregator:
self.logger.info("STEP 3b: ADD BAHASA INDONESIA NAME COLUMNS")
self.logger.info("=" * 80)
# Nama negara
self.df["country_name_id"] = (
self.df["country_name"]
.apply(get_country_name_id)
.astype(str)
)
# Nama indikator
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:,}")
# Log sample mapping
# 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):")
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[["indicator_name", "indicator_name_id"]]
.drop_duplicates()
@@ -716,14 +792,6 @@ class IndicatorNormAggregator:
self.logger.info(f" EN: {r['indicator_name'][:55]}")
self.logger.info(f" ID: {r['indicator_name_id'][:55]}")
sample_pil = (
self.df[["pillar_name", "pillar_name_id"]]
.drop_duplicates()
)
self.logger.info("\n Pillar mapping (EN -> ID):")
for _, r in sample_pil.iterrows():
self.logger.info(f" {r['pillar_name']:<20} -> {r['pillar_name_id']}")
# =========================================================================
# STEP 4: Deteksi sdgs_start_year
# =========================================================================
@@ -925,7 +993,7 @@ class IndicatorNormAggregator:
self.logger.info("=" * 80)
out = df[[
"year", "country_id", "country_name",
"year", "country_id", "country_name", "country_name_id",
"indicator_id", "indicator_name", "indicator_name_id",
"unit", "direction",
"pillar_id", "pillar_name", "pillar_name_id",
@@ -941,6 +1009,7 @@ class IndicatorNormAggregator:
out["year"] = out["year"].astype(int)
out["country_id"] = out["country_id"].astype(int)
out["country_name"] = out["country_name"].astype(str)
out["country_name_id"] = out["country_name_id"].astype(str)
out["indicator_id"] = out["indicator_id"].astype(int)
out["indicator_name"] = out["indicator_name"].astype(str)
out["indicator_name_id"] = out["indicator_name_id"].astype(str)
@@ -966,6 +1035,7 @@ class IndicatorNormAggregator:
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="NULLABLE"),
bigquery.SchemaField("indicator_id", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("indicator_name", "STRING", mode="REQUIRED"),
bigquery.SchemaField("indicator_name_id", "STRING", mode="NULLABLE"),
@@ -1009,7 +1079,7 @@ class IndicatorNormAggregator:
"yoy_columns" : ["yoy_value", "yoy_norm_value"],
"performance_threshold": _PERFORMANCE_THRESHOLD,
"unit_source" : "dim_indicator",
"added_columns" : ["indicator_name_id", "pillar_name_id"],
"added_columns" : ["country_name_id", "indicator_name_id", "pillar_name_id"],
}),
"validation_metrics" : json.dumps({
"total_rows" : rows_loaded,
@@ -1062,6 +1132,7 @@ class IndicatorNormAggregator:
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("=" * 80)
df = df_final.copy()
@@ -1150,7 +1221,7 @@ class IndicatorNormAggregator:
})
df_yoy_stats = pd.DataFrame(yoy_stats)
# Country best/worst
# Country best/worst (nama asli Bahasa Inggris)
df_country_avg = (
df.groupby(["indicator_id", "country_id", "country_name"])
.agg(country_avg_value=("value", "mean"))
@@ -1166,9 +1237,12 @@ class IndicatorNormAggregator:
worst_row = grp.loc[grp["country_avg_value"].idxmin()]
best_row = grp.loc[grp["country_avg_value"].idxmax()]
country_stats.append({
"indicator_id" : ind_id,
"country_worst": worst_row["country_name"],
"country_best" : best_row["country_name"],
"indicator_id" : ind_id,
"country_worst" : worst_row["country_name"], # nama Inggris
"country_best" : best_row["country_name"], # nama Inggris
# Tambahan: nama Indonesia untuk kedua negara
"country_worst_id": get_country_name_id(worst_row["country_name"]),
"country_best_id" : get_country_name_id(best_row["country_name"]),
})
df_country_stats = pd.DataFrame(country_stats)
@@ -1229,6 +1303,7 @@ class IndicatorNormAggregator:
"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",
]].copy()
@@ -1255,6 +1330,8 @@ class IndicatorNormAggregator:
out["best_yoy_to"] = pd.to_numeric(out["best_yoy_to"], errors="coerce").astype("Int64")
out["country_worst"] = out["country_worst"].astype(str).replace("nan", pd.NA).astype("string")
out["country_best"] = out["country_best"].astype(str).replace("nan", pd.NA).astype("string")
out["country_worst_id"] = out["country_worst_id"].astype(str).replace("nan", pd.NA).astype("string")
out["country_best_id"] = out["country_best_id"].astype(str).replace("nan", pd.NA).astype("string")
out["narrative_en"] = out["narrative_en"].astype(str)
out["narrative_id"] = out["narrative_id"].astype(str)
@@ -1280,6 +1357,8 @@ class IndicatorNormAggregator:
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"),
]
@@ -1310,7 +1389,8 @@ class IndicatorNormAggregator:
"narrative_dimensions" : ["trend", "gap_trend", "anomaly", "country_consistency"],
"performance_threshold": _PERFORMANCE_THRESHOLD,
"layer" : "gold",
"added_columns" : ["indicator_name_id", "pillar_name_id"],
"added_columns" : ["country_name_id", "indicator_name_id", "pillar_name_id",
"country_worst_id", "country_best_id"],
}),
"validation_metrics" : json.dumps({
"total_rows" : rows_loaded,
@@ -1334,13 +1414,14 @@ class IndicatorNormAggregator:
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 : indicator_name_id, pillar_name_id (Bahasa Indonesia)")
self.logger.info(" Added : country_name_id, indicator_name_id, pillar_name_id (Bahasa Indonesia)")
self.logger.info(" FIXED : 'Access' -> 'Akses', narrative_id pakai nama ID")
self.logger.info("=" * 80)
self.load_data()
self.load_units()
self._merge_unit()
self._add_indonesia_name_columns() # <-- BARU
self._add_indonesia_name_columns()
self.sdgs_start_year = self._detect_sdgs_start_year()
self._assign_framework()
df_normed = self._compute_norm_values()