OECD_Waste_Water_Treatment cloud job fix and goldens

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"

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]"

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"
             ]

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"
+            }
+        }
+    ]
+}

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)