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..99372d0413 100644 --- a/statvar_imports/oecd/wastewater_treatment/manifest.json +++ b/statvar_imports/oecd/wastewater_treatment/manifest.json @@ -10,17 +10,21 @@ "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": [ { "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" + "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 new file mode 100644 index 0000000000..3c7cbd14d6 --- /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..672eff2b3c --- /dev/null +++ b/tools/statvar_importer/merge_historical_data.py @@ -0,0 +1,144 @@ +# 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 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('Concatenating and deduplicating historical vs current data...') + + # 1. current_df is first, historical_df is second + combined_df = pd.concat([current_df, historical_df], 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: + final_df.to_csv(f, index=False, encoding='utf-8') + logging.info('Successfully saved unified dataset to: %s', output_path) + + logging.info( + '\n' + '=' * 80 + '\n' + f'Current Base Records Processed : {len(current_df)}\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 + ) + 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)