From be10572c281d21709c2c6b56d0d8ee4d4c5d730b Mon Sep 17 00:00:00 2001 From: Sanika Prasad Date: Thu, 16 Jul 2026 12:57:31 +0000 Subject: [PATCH 1/5] Oecd job fix and goldens --- .../golden_data/golden_observations.csv | 53 ++++++ .../golden_data/golden_summary_report.csv | 9 ++ .../oecd/wastewater_treatment/manifest.json | 6 +- .../validation_config.json | 29 ++++ .../statvar_importer/merge_historical_data.py | 152 ++++++++++++++++++ 5 files changed, 247 insertions(+), 2 deletions(-) create mode 100644 statvar_imports/oecd/wastewater_treatment/golden_data/golden_observations.csv create mode 100644 statvar_imports/oecd/wastewater_treatment/golden_data/golden_summary_report.csv create mode 100644 statvar_imports/oecd/wastewater_treatment/validation_config.json create mode 100644 tools/statvar_importer/merge_historical_data.py diff --git a/statvar_imports/oecd/wastewater_treatment/golden_data/golden_observations.csv b/statvar_imports/oecd/wastewater_treatment/golden_data/golden_observations.csv new file mode 100644 index 0000000000..72312474d8 --- /dev/null +++ b/statvar_imports/oecd/wastewater_treatment/golden_data/golden_observations.csv @@ -0,0 +1,53 @@ +"observationAbout" +"country/AUS" +"country/AUT" +"country/BEL" +"country/CAN" +"country/CHL" +"country/CZE" +"country/DNK" +"country/EST" +"country/FIN" +"country/DEU" +"country/GRC" +"country/HUN" +"country/ISL" +"country/IRL" +"country/ITA" +"country/JPN" +"country/KOR" +"country/LUX" +"country/NLD" +"country/NOR" +"country/POL" +"country/PRT" +"country/SVK" +"country/SVN" +"country/ESP" +"country/SWE" +"country/CHE" +"country/TUR" +"country/GBR" +"country/BGR" +"country/HRV" +"country/ROU" +"country/FRA" +"country/LVA" +"country/USA" +"country/ARG" +"country/MDA" +"country/ZAF" +"country/ARM" +"country/NZL" +"country/ISR" +"country/LTU" +"country/CRI" +"country/BRA" +"country/AZE" +"country/CHN" +"country/GEO" +"country/SAU" +"country/BLR" +"country/IND" +"country/UKR" +"country/MEX" diff --git a/statvar_imports/oecd/wastewater_treatment/golden_data/golden_summary_report.csv b/statvar_imports/oecd/wastewater_treatment/golden_data/golden_summary_report.csv new file mode 100644 index 0000000000..4985872130 --- /dev/null +++ b/statvar_imports/oecd/wastewater_treatment/golden_data/golden_summary_report.csv @@ -0,0 +1,9 @@ +"MeasurementMethods","observationPeriods","MinDate","Units","NumPlaces","StatVar","ScalingFactors" +"[]","[]","1970","[Percent]","40","Count_Person_WasteWaterTreatment_TertiaryTreatment_WasteWaterTreated_PublicSewerage_WasteWaterTreatmentPlant_AsAFractionOf_Count_Person_WasteWaterTreatment_PublicSewerage","[100]" +"[]","[]","1970","[Percent]","48","Count_Person_WasteWaterTreatment_WasteWaterNotTreated_PublicSewerage_PublicSewerageWithoutTreatment_AsAFractionOf_Count_Person_WasteWaterTreatment_PublicSewerage","[100]" +"[]","[]","1970","[Percent]","39","Count_Person_WasteWaterTreatment_UnspecifiedTreatment_WasteWaterTreated_PublicSewerage_WasteWaterTreatmentPlant_AsAFractionOf_Count_Person_WasteWaterTreatment_PublicSewerage","[100]" +"[]","[]","1970","[Percent]","40","Count_Person_WasteWaterTreatment_PrimaryTreatment_WasteWaterTreated_PublicSewerage_WasteWaterTreatmentPlant_AsAFractionOf_Count_Person_WasteWaterTreatment_PublicSewerage","[100]" +"[]","[]","1970","[Percent]","49","Count_Person_WasteWaterTreatment_WasteWaterTreated_PublicSewerage_WasteWaterTreatmentPlant_AsAFractionOf_Count_Person_WasteWaterTreatment_PublicSewerage","[100]" +"[]","[]","1970","[Percent]","36","Count_Person_WasteWaterTreatment_WasteWaterTreated_WasteWaterIndependentlyTreated_AsAFractionOf_Count_Person","[100]" +"[]","[]","1970","[Percent]","42","Count_Person_WasteWaterTreatment_SecondaryTreatment_WasteWaterTreated_PublicSewerage_WasteWaterTreatmentPlant_AsAFractionOf_Count_Person_WasteWaterTreatment_PublicSewerage","[100]" +"[]","[]","1970","[Percent]","51","Count_Person_WasteWaterTreatment_PublicSewerage_AsAFractionOf_Count_Person","[100]" diff --git a/statvar_imports/oecd/wastewater_treatment/manifest.json b/statvar_imports/oecd/wastewater_treatment/manifest.json index cd3821ddbc..bc4eaac323 100644 --- a/statvar_imports/oecd/wastewater_treatment/manifest.json +++ b/statvar_imports/oecd/wastewater_treatment/manifest.json @@ -10,15 +10,17 @@ "scripts": [ "../../../util/download_util_script.py --download_url='https://sdmx.oecd.org/public/rest/data/OECD.ENV.EPI,DSD_WATER_TREAT@DF_WATER_TREAT,/all?dimensionAtObservation=AllDimensions&format=csvfilewithlabels' --output_folder=input", "preprocess.py", - "../../../tools/statvar_importer/stat_var_processor.py --input_data=input/oecd_wastewatertreatment_data.csv --pv_map=oecd_wastewatertreatment_pvmap.csv --config_file=oecd_wastewatertreatment_metadata.csv --places_resolved_csv=oecd_wastewatertreatment_places_resolved.csv --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_path=output/oecd_wastewatertreatment" + "../../../tools/statvar_importer/stat_var_processor.py --input_data=input/oecd_wastewatertreatment_data.csv --pv_map=oecd_wastewatertreatment_pvmap.csv --config_file=oecd_wastewatertreatment_metadata.csv --places_resolved_csv=oecd_wastewatertreatment_places_resolved.csv --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_path=output/oecd_wastewatertreatment", + "../../../tools/statvar_importer/merge_historical_data.py --current_output=output/oecd_wastewatertreatment.csv --historical_output=https://storage.mtls.cloud.google.com/unresolved_mcf/oecd/oecd_wastewater_treatment/historical_data/oecd_waste_water_historical.csv --output_file=output/oecd_wastewatertreatment_merged.csv" ], "import_inputs": [ { "template_mcf": "output/oecd_wastewatertreatment.tmcf", - "cleaned_csv": "output/oecd_wastewatertreatment.csv" + "cleaned_csv": "output/oecd_wastewatertreatment_merged.csv" } ], "cron_schedule": "0 10 1,15 * *", + "validation_config_file": "validation_config.json", "source_files": [ "input/*.csv" ] diff --git a/statvar_imports/oecd/wastewater_treatment/validation_config.json b/statvar_imports/oecd/wastewater_treatment/validation_config.json new file mode 100644 index 0000000000..e26f7fbed6 --- /dev/null +++ b/statvar_imports/oecd/wastewater_treatment/validation_config.json @@ -0,0 +1,29 @@ +{ + "schema_version": "1.0", + "rules": [ + { + "rule_id": "check_deleted_records_percent", + "description": "Checks that the percentage of deleted points is within the threshold.", + "validator": "DELETED_RECORDS_PERCENT", + "params": { + "threshold": 0.1 + } + }, + { + "rule_id": "check_goldens_summary_report", + "validator": "GOLDENS_CHECK", + "params": { + "golden_files": "../../../../golden_data/golden_summary_report.csv" + } + }, + { + "rule_id": "Check_goldens_output_csv", + "description": "Verifies the generated output CSV data matches established critical golden records", + "validator": "GOLDENS_CHECK", + "params": { + "golden_files": "../../../../golden_data/golden_observations.csv", + "input_files": "../../../../output/oecd_wastewatertreatment_merged.csv" + } + } + ] +} diff --git a/tools/statvar_importer/merge_historical_data.py b/tools/statvar_importer/merge_historical_data.py new file mode 100644 index 0000000000..fae901e20e --- /dev/null +++ b/tools/statvar_importer/merge_historical_data.py @@ -0,0 +1,152 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Utility script to check and retain missing historical data points. + +This script compares a current output CSV with an older historical CSV. It +identifies rows in the historical file that are still missing from the current +output file based on specific composite keys and generates a unified +concatenated CSV. It safely accesses cloud storage locations using internal utilities. +""" + +import os +import sys + +# Dynamically find the path relative to this script's location +_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) +# Navigate relative to tools/statvar_importer/ to find the shared util directory +_UTIL_PATH = os.path.abspath(os.path.join(_SCRIPT_DIR, '../../util/')) + +if _UTIL_PATH not in sys.path: + sys.path.append(_UTIL_PATH) + +try: + import file_util +except ModuleNotFoundError: + # Fallback logic to check alternative root folder configurations + _ALT_UTIL_PATH = os.path.abspath(os.path.join(_SCRIPT_DIR, '../../../util/')) + if _ALT_UTIL_PATH not in sys.path: + sys.path.append(_ALT_UTIL_PATH) + import file_util + +from absl import app +from absl import flags +from absl import logging +import pandas as pd + +FLAGS = flags.FLAGS + +flags.DEFINE_string( + 'current_output', + None, + 'Path to the latest current output CSV file.', +) +flags.DEFINE_string( + 'historical_output', + None, + 'Path to the historical retention CSV file (supports gs:// or storage.mtls paths).', +) +flags.DEFINE_string( + 'output_file', + 'final_combined_output.csv', + 'Path to save the generated unified output file.', +) + + +def _read_csv(file_path: str) -> pd.DataFrame: + """Reads a CSV into a DataFrame, converting HTTP storage links to gs:// paths.""" + # Convert browser mTLS links to gs:// URI syntax if passed to the flag + mtls_prefix = 'https://storage.mtls.cloud.google.com/' + if file_path.startswith(mtls_prefix): + file_path = 'gs://' + file_path[len(mtls_prefix):] + logging.info('Normalized mTLS URL to native path: %s', file_path) + + if file_path.startswith('gs://'): + with file_util.FileIO(file_path, 'r') as f: + return pd.read_csv(f) + return pd.read_csv(file_path) + + +def process_and_reconcile_data(current_path: str, historical_path: str, + output_path: str) -> None: + """Compares historical data against current data and retains missing records. + + Args: + current_path: Path to the current output data file. + historical_path: Path to the historical tracking data file. + output_path: Target path where the merged CSV file should be saved. + """ + logging.info('Loading input CSV datasets...') + try: + current_df = _read_csv(current_path) + historical_df = _read_csv(historical_path) + except Exception as e: + logging.fatal('Failed to read input files: %s', e) + sys.exit(1) + + # Define the columns that serve as the unique identifier/key for each data point + keys = [ + 'observationAbout', + 'observationDate', + 'variableMeasured', + 'unit', + 'scalingFactor', + ] + + logging.info('Generating series footprints for alignment validation...') + + # Extract only the key columns from the current CSV and drop any duplicates + current_keys = current_df[keys].drop_duplicates() + + # Perform a left merge from the historical data onto the current keys + merged = historical_df.merge(current_keys, on=keys, how='left', indicator=True) + + # Keep only the rows from the historical CSV that DO NOT exist in the current CSV + historical_only = historical_df[merged['_merge'] == 'left_only'].copy() + + if historical_only.empty: + logging.info('All historical data points are already active in current.') + final_df = current_df + else: + logging.info( + 'Retaining %d records from the historical framework.', + len(historical_only), + ) + # Combine all data points from current CSV with the unique historical entries + final_df = pd.concat([current_df, historical_only], ignore_index=True) + + try: + final_df.to_csv(output_path, index=False, encoding='utf-8') + logging.info('Successfully saved unified dataset to: %s', output_path) + + print('=' * 80) + print(f'Current Base Records Processed : {len(current_df)}') + print(f'Historical Points Re-inserted : {len(historical_only)}') + print(f'Total Unified Series Exported : {len(final_df)}') + print(f'Target File Location : {output_path}') + print('=' * 80) + except Exception as e: + logging.fatal('Failed to write reconciliation dataset matrix: %s', e) + sys.exit(1) + + +def main(_): + flags.mark_flag_as_required('current_output') + flags.mark_flag_as_required('historical_output') + process_and_reconcile_data( + FLAGS.current_output, FLAGS.historical_output, FLAGS.output_file + ) + + +if __name__ == '__main__': + app.run(main) From 56b84a8e7c13d5b914b2150bc1685075b4b94306 Mon Sep 17 00:00:00 2001 From: Sanika Prasad Date: Tue, 21 Jul 2026 12:44:53 +0000 Subject: [PATCH 2/5] PR comments --- tools/statvar_importer/merge_historical_data.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tools/statvar_importer/merge_historical_data.py b/tools/statvar_importer/merge_historical_data.py index fae901e20e..39f11081e4 100644 --- a/tools/statvar_importer/merge_historical_data.py +++ b/tools/statvar_importer/merge_historical_data.py @@ -126,7 +126,8 @@ def process_and_reconcile_data(current_path: str, historical_path: str, final_df = pd.concat([current_df, historical_only], ignore_index=True) try: - final_df.to_csv(output_path, index=False, encoding='utf-8') + with file_util.FileIO(output_path, 'w') as f: + final_df.to_csv(f, index=False, encoding='utf-8') logging.info('Successfully saved unified dataset to: %s', output_path) print('=' * 80) From 2914d0ff86c86678dc0045bf4925df59817a829c Mon Sep 17 00:00:00 2001 From: Sanika Prasad Date: Wed, 22 Jul 2026 08:49:20 +0000 Subject: [PATCH 3/5] Fix Lint --- tools/statvar_importer/merge_historical_data.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tools/statvar_importer/merge_historical_data.py b/tools/statvar_importer/merge_historical_data.py index 39f11081e4..391771047c 100644 --- a/tools/statvar_importer/merge_historical_data.py +++ b/tools/statvar_importer/merge_historical_data.py @@ -16,7 +16,7 @@ This script compares a current output CSV with an older historical CSV. It identifies rows in the historical file that are still missing from the current output file based on specific composite keys and generates a unified -concatenated CSV. It safely accesses cloud storage locations using internal utilities. +concatenated CSV. It safely accesses cloud storage locations using internal utilities """ import os From 5734e09f7a186804177f45c9ad7e5c499d08f51a Mon Sep 17 00:00:00 2001 From: Sanika Prasad Date: Thu, 23 Jul 2026 12:52:29 +0000 Subject: [PATCH 4/5] Internal PR Comments --- .../oecd/wastewater_treatment/manifest.json | 6 ++-- .../validation_config.json | 2 +- .../statvar_importer/merge_historical_data.py | 29 ++++++++++--------- 3 files changed, 20 insertions(+), 17 deletions(-) diff --git a/statvar_imports/oecd/wastewater_treatment/manifest.json b/statvar_imports/oecd/wastewater_treatment/manifest.json index bc4eaac323..99372d0413 100644 --- a/statvar_imports/oecd/wastewater_treatment/manifest.json +++ b/statvar_imports/oecd/wastewater_treatment/manifest.json @@ -10,7 +10,7 @@ "scripts": [ "../../../util/download_util_script.py --download_url='https://sdmx.oecd.org/public/rest/data/OECD.ENV.EPI,DSD_WATER_TREAT@DF_WATER_TREAT,/all?dimensionAtObservation=AllDimensions&format=csvfilewithlabels' --output_folder=input", "preprocess.py", - "../../../tools/statvar_importer/stat_var_processor.py --input_data=input/oecd_wastewatertreatment_data.csv --pv_map=oecd_wastewatertreatment_pvmap.csv --config_file=oecd_wastewatertreatment_metadata.csv --places_resolved_csv=oecd_wastewatertreatment_places_resolved.csv --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_path=output/oecd_wastewatertreatment", + "../../../tools/statvar_importer/stat_var_processor.py --input_data=input/oecd_wastewatertreatment_data.csv --pv_map=oecd_wastewatertreatment_pvmap.csv --config_file=oecd_wastewatertreatment_metadata.csv --places_resolved_csv=oecd_wastewatertreatment_places_resolved.csv --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_path=output/oecd_wastewatertreatment --output_counters=counters/oecd_wastewatertreatment.csv", "../../../tools/statvar_importer/merge_historical_data.py --current_output=output/oecd_wastewatertreatment.csv --historical_output=https://storage.mtls.cloud.google.com/unresolved_mcf/oecd/oecd_wastewater_treatment/historical_data/oecd_waste_water_historical.csv --output_file=output/oecd_wastewatertreatment_merged.csv" ], "import_inputs": [ @@ -22,7 +22,9 @@ "cron_schedule": "0 10 1,15 * *", "validation_config_file": "validation_config.json", "source_files": [ - "input/*.csv" + "input/*.csv", + "golden_data/*.csv", + "counters/*.csv" ] } ] diff --git a/statvar_imports/oecd/wastewater_treatment/validation_config.json b/statvar_imports/oecd/wastewater_treatment/validation_config.json index e26f7fbed6..3c7cbd14d6 100644 --- a/statvar_imports/oecd/wastewater_treatment/validation_config.json +++ b/statvar_imports/oecd/wastewater_treatment/validation_config.json @@ -17,7 +17,7 @@ } }, { - "rule_id": "Check_goldens_output_csv", + "rule_id": "check_goldens_output_csv", "description": "Verifies the generated output CSV data matches established critical golden records", "validator": "GOLDENS_CHECK", "params": { diff --git a/tools/statvar_importer/merge_historical_data.py b/tools/statvar_importer/merge_historical_data.py index 391771047c..9eb7712326 100644 --- a/tools/statvar_importer/merge_historical_data.py +++ b/tools/statvar_importer/merge_historical_data.py @@ -105,14 +105,13 @@ def process_and_reconcile_data(current_path: str, historical_path: str, logging.info('Generating series footprints for alignment validation...') - # Extract only the key columns from the current CSV and drop any duplicates - current_keys = current_df[keys].drop_duplicates() - - # Perform a left merge from the historical data onto the current keys - merged = historical_df.merge(current_keys, on=keys, how='left', indicator=True) - - # Keep only the rows from the historical CSV that DO NOT exist in the current CSV - historical_only = historical_df[merged['_merge'] == 'left_only'].copy() + # Identify rows in historical_df whose composite keys are missing from current_df + merged = historical_df.merge( + current_df[keys].drop_duplicates(), on=keys, how='left', indicator=True + ) + historical_only = merged[merged['_merge'] == 'left_only'].drop( + columns=['_merge'] + ) if historical_only.empty: logging.info('All historical data points are already active in current.') @@ -130,12 +129,14 @@ def process_and_reconcile_data(current_path: str, historical_path: str, final_df.to_csv(f, index=False, encoding='utf-8') logging.info('Successfully saved unified dataset to: %s', output_path) - print('=' * 80) - print(f'Current Base Records Processed : {len(current_df)}') - print(f'Historical Points Re-inserted : {len(historical_only)}') - print(f'Total Unified Series Exported : {len(final_df)}') - print(f'Target File Location : {output_path}') - print('=' * 80) + logging.info( + '\n' + '=' * 80 + '\n' + f'Current Base Records Processed : {len(current_df)}\n' + f'Historical Points Re-inserted : {len(historical_only)}\n' + f'Total Unified Series Exported : {len(final_df)}\n' + f'Target File Location : {output_path}\n' + + '=' * 80 + ) except Exception as e: logging.fatal('Failed to write reconciliation dataset matrix: %s', e) sys.exit(1) From 7ffe11941260d19b5761b93eb8751a34c2498761 Mon Sep 17 00:00:00 2001 From: Sanika Prasad Date: Mon, 27 Jul 2026 07:08:01 +0000 Subject: [PATCH 5/5] PR comments and lint --- .../statvar_importer/merge_historical_data.py | 32 +++++++------------ 1 file changed, 11 insertions(+), 21 deletions(-) diff --git a/tools/statvar_importer/merge_historical_data.py b/tools/statvar_importer/merge_historical_data.py index 9eb7712326..672eff2b3c 100644 --- a/tools/statvar_importer/merge_historical_data.py +++ b/tools/statvar_importer/merge_historical_data.py @@ -14,9 +14,9 @@ """Utility script to check and retain missing historical data points. This script compares a current output CSV with an older historical CSV. It -identifies rows in the historical file that are still missing from the current +identifies rows in the historical file that are missing from the current output file based on specific composite keys and generates a unified -concatenated CSV. It safely accesses cloud storage locations using internal utilities +concatenated CSV. It safely accesses cloud storage locations using internal utilities. """ import os @@ -103,26 +103,16 @@ def process_and_reconcile_data(current_path: str, historical_path: str, 'scalingFactor', ] - logging.info('Generating series footprints for alignment validation...') + logging.info('Concatenating and deduplicating historical vs current data...') - # Identify rows in historical_df whose composite keys are missing from current_df - merged = historical_df.merge( - current_df[keys].drop_duplicates(), on=keys, how='left', indicator=True - ) - historical_only = merged[merged['_merge'] == 'left_only'].drop( - columns=['_merge'] - ) + # 1. current_df is first, historical_df is second + combined_df = pd.concat([current_df, historical_df], ignore_index=True) - if historical_only.empty: - logging.info('All historical data points are already active in current.') - final_df = current_df - else: - logging.info( - 'Retaining %d records from the historical framework.', - len(historical_only), - ) - # Combine all data points from current CSV with the unique historical entries - final_df = pd.concat([current_df, historical_only], ignore_index=True) + # 2. keep='first' retains records from current_df and drops duplicate keys from historical_df + final_df = combined_df.drop_duplicates(subset=keys, keep='first') + + retained_count = len(final_df) - len(current_df) + logging.info('Retaining %d records from the historical framework.', retained_count) try: with file_util.FileIO(output_path, 'w') as f: @@ -132,7 +122,7 @@ def process_and_reconcile_data(current_path: str, historical_path: str, logging.info( '\n' + '=' * 80 + '\n' f'Current Base Records Processed : {len(current_df)}\n' - f'Historical Points Re-inserted : {len(historical_only)}\n' + f'Historical Points Re-inserted : {retained_count}\n' f'Total Unified Series Exported : {len(final_df)}\n' f'Target File Location : {output_path}\n' + '=' * 80