oecd_regional_education |Increasing deletion threshold

statvar_imports/oecd/regional_education/manifest.json

@@ -6,21 +6,24 @@
                 "support@datacommons.org"
             ],
             "provenance_url": "https://stats.oecd.org/Index.aspx?DataSetCode=REGION_EDUCAT",
-            "provenance_description": "dataset contains commodity price and its index value for monthly and annual",
+            "provenance_description": "dataset contains regional educational attainment statistics for OECD regions and cities",
             "scripts": [
                 "../../../util/download_util_script.py --download_url='https://sdmx.oecd.org/public/rest/data/OECD.CFE.EDS,DSD_REG_EDU@DF_ATTAIN,/A.........?dimensionAtObservation=AllDimensions&format=csvfilewithlabels' --output_folder=gcs_output/source_files",
                 "preprocess.py",
-                "../../../tools/statvar_importer/stat_var_processor.py --input_data=gcs_output/source_files/oecd_regional_education_data.csv --pv_map=oecd_regional_education_pvmap.csv --config_file=oecd_regional_education_metadata.csv --places_resolved_csv=oecd_regional_education_places_resolved.csv --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_path=output/oecd_regional_education"
+                "../../../tools/statvar_importer/stat_var_processor.py --input_data=gcs_output/source_files/oecd_regional_education_data.csv --pv_map=oecd_regional_education_pvmap.csv --config_file=oecd_regional_education_metadata.csv --places_resolved_csv=oecd_regional_education_places_resolved.csv --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_path=output/oecd_regional_education --output_counters=counters/oecd_regional_education_counters.csv"
             ],
             "import_inputs": [
                 {
                     "template_mcf": "output/oecd_regional_education.tmcf",
-                    "cleaned_csv": "output/oecd_regional_education.csv"
+                    "cleaned_csv": "output/oecd_regional_education.csv",
+                    "node_mcf": "output/*.mcf"
                 }
             ],
             "cron_schedule": "0 10 1,15 * *",
+            "validation_config_file": "validation_config.json",
             "source_files": [
-                "gcs_output/source_files/*.csv"
+                "gcs_output/source_files/*",
+                "counters/*.csv"
             ]
         }
     ]

statvar_imports/oecd/regional_education/oecd_regional_education_custom_schema.mcf

@@ -0,0 +1,5 @@
+Node: dcid:PostSecondaryNonTertiaryEducation__UpperSecondaryEducation
+typeOf: dcs:ProvisionalNode
+isProvisional: dcs:True
+name: "PostSecondaryNonTertiaryEducation__UpperSecondaryEducation"
+description: "Upper secondary and post-secondary non-tertiary education (ISCED 2011 levels 3 and 4)."

statvar_imports/oecd/regional_education/oecd_regional_education_metadata.csv

@@ -11,4 +11,6 @@ comments,"The source for education data is from 1993 to 2022, unable to selet al
 Path to data - URL - Regions and Cities - Regional Statistics - Regional Education - Educational Attainment"
 output_columns,"observationAbout,observationDate,variableMeasured,value"
 header_rows,1
-word_delimiter,""""""
+word_delimiter,""""""
+reconcile_nodes,
+mapped_columns,"4,5,6"

statvar_imports/oecd/regional_education/oecd_regional_education_places_resolved.csv

@@ -119,7 +119,7 @@ BG41,nuts/BG41,South West,Bulgaria
 BG42,nuts/BG42,South Central,Bulgaria
 CAN,country/CAN,Canada,Canada
 CA10,wikidataId/Q2003,Newfoundland and Labrador,Canada
-CA11, wikidataId/Q1979,Prince Edward Island,Canada
+CA11,wikidataId/Q1979,Prince Edward Island,Canada
 CA12,wikidataId/Q1952,Nova Scotia,Canada
 CA13,wikidataId/Q1965,New Brunswick,Canada
 CA24,wikidataId/Q176,Quebec,Canada
@@ -683,6 +683,8 @@ NL31,nuts/NL31,Utrecht,Netherlands
 NL32,nuts/NL32,North Holland,Netherlands
 NL33,nuts/NL33,South Holland,Netherlands
 NL34,nuts/NL34,Zeeland,Netherlands
+NL35,nuts/NL35,Utrecht,Netherlands
+NL36,nuts/NL36,South Holland,Netherlands
 NL41,nuts/NL41,North Brabant,Netherlands
 NL42,nuts/NL42,Limburg,Netherlands
 NZ23,wikidataId/Q657004,Canterbury,New Zealand
@@ -736,6 +738,11 @@ PT15,nuts/PT15,Algarve,Portugal
 PT16,nuts/PT16,Central Portugal,Portugal
 PT17,nuts/PT17,Metropolitan area of Lisbon,Portugal
 PT18,nuts/PT18,Alentejo,Portugal
+PT19,nuts/PT19,Centre,Portugal
+PT1A,nuts/PT1A,Greater Lisbon,Portugal
+PT1B,nuts/PT1B,Setúbal Peninsula,Portugal
+PT1C,nuts/PT1C,Alentejo,Portugal
+PT1D,nuts/PT1D,West and Tagus Valley,Portugal
 PT20,nuts/PT20,Autonomous Region of the Azores,Portugal
 PT30,nuts/PT30,Autonomous Region of Madeira,Portugal
 ROU,country/ROU,Romania,Romania

statvar_imports/oecd/regional_education/preprocess.py

@@ -1,55 +1,236 @@
+import csv
 import os
 import re
-from absl import logging
+import shutil
 
-# --- Add this line to set verbosity ---
-logging.set_verbosity(logging.INFO)
-# For even more detail if you have debug messages:
-# logging.set_verbosity(logging.DEBUG)
-# --------------------------------------
+try:
+    from absl import logging
+    logging.set_verbosity(logging.INFO)
+except ImportError:
+    import logging as std_logging
 
-def rename_target_file(base_path='.'):
+    class _CompatLogger:
+        def __init__(self):
+            self._logger = std_logging.getLogger(__name__)
+            self._logger.setLevel(std_logging.INFO)
+            if not self._logger.handlers:
+                handler = std_logging.StreamHandler()
+                handler.setFormatter(
+                    std_logging.Formatter('%(levelname)s:%(message)s'))
+                self._logger.addHandler(handler)
+
+        def set_verbosity(self, level):
+            self._logger.setLevel(level)
+
+        def info(self, msg, *args, **kwargs):
+            self._logger.info(msg, *args, **kwargs)
+
+        def warning(self, msg, *args, **kwargs):
+            self._logger.warning(msg, *args, **kwargs)
+
+        def error(self, msg, *args, **kwargs):
+            self._logger.error(msg, *args, **kwargs)
+
+    logging = _CompatLogger()
+    logging.set_verbosity(std_logging.INFO)
+
+
+def preprocess(base_path=None):
+    if base_path is None:
+        base_path = os.path.dirname(os.path.abspath(__file__))
     folder_name = 'gcs_output/source_files'
     target_folder = os.path.join(base_path, folder_name)
+    counters_folder = os.path.join(base_path, 'counters')
+    os.makedirs(counters_folder, exist_ok=True)
+    output_folder = os.path.join(base_path, 'output')
+    os.makedirs(output_folder, exist_ok=True)
+
+    custom_schema_src = os.path.join(
+        base_path, 'oecd_regional_education_custom_schema.mcf')
+    if os.path.isfile(custom_schema_src):
+        custom_schema_dst = os.path.join(
+            output_folder, 'oecd_regional_education_custom_schema.mcf')
+        shutil.copy(custom_schema_src, custom_schema_dst)
+        logging.info(f"Copied custom schema to {custom_schema_dst}")
+
+    places_resolved_file = os.path.join(
+        base_path, 'oecd_regional_education_places_resolved.csv')
+    valid_places = {}
+    if not os.path.isfile(places_resolved_file):
+        raise FileNotFoundError(
+            f"Places resolved file not found: {places_resolved_file}. "
+            "Aborting preprocessing to prevent silent data drop.")
+
+    rows_to_rewrite = []
+    needs_rewrite = False
+    with open(places_resolved_file, 'r', encoding='utf-8', newline='') as f:
+        reader = csv.reader(f)
+        for row in reader:
+            stripped_row = [cell.strip() for cell in row]
+            if stripped_row != row:
+                needs_rewrite = True
+            rows_to_rewrite.append(stripped_row)
+
+    if needs_rewrite and rows_to_rewrite:
+        tmp_places = places_resolved_file + '.tmp'
+        with open(tmp_places, 'w', encoding='utf-8', newline='\r\n') as f:
+            writer = csv.writer(f)
+            writer.writerows(rows_to_rewrite)
+        os.replace(tmp_places, places_resolved_file)
+        logging.info(f"Sanitized whitespace in {places_resolved_file}")
+
+    with open(places_resolved_file, 'r', encoding='utf-8') as f:
+        reader = csv.DictReader(f)
+        for row in reader:
+            dcid = row.get('dcid', '').strip()
+            place_name = row.get('place_name', '').strip()
+            if dcid and place_name:
+                if not dcid.startswith('dcid:'):
+                    dcid = f'dcid:{dcid}'
+                valid_places[place_name] = dcid
+
+    if not valid_places:
+        raise ValueError(
+            f"No valid places loaded from {places_resolved_file}. "
+            "Aborting preprocessing to prevent silent data drop.")
+
+    logging.info(f"Loaded {len(valid_places)} valid places from {places_resolved_file}")
 
-    try:
-        # Check if the folder exists
-        if not os.path.isdir(target_folder):
-            logging.fatal(f"Folder '{folder_name}' not found in '{base_path}'") # Changed to error for non-fatal issues
-            return # Exit function if folder not found
+    if not os.path.isdir(target_folder):
+        logging.error(f"Folder '{folder_name}' not found in '{base_path}'")
+        raise FileNotFoundError(f"Folder '{folder_name}' not found in '{base_path}'")
 
-        # Pattern to match any file starting with 'A'
-        pattern = re.compile(r'^A.*$', re.IGNORECASE)
-        renamed = False
+    pattern = re.compile(r'^A\.{9}$', re.IGNORECASE)
+    candidate_files = sorted([f for f in os.listdir(target_folder) if pattern.match(f)])
+    raw_file = candidate_files[0] if candidate_files else None
 
-        # Search through files in the folder
-        for filename in os.listdir(target_folder):
-            logging.info(f"Checking file: {filename}")
-            if pattern.match(filename):
-                old_path = os.path.join(target_folder, filename)
-                new_path = os.path.join(target_folder, 'oecd_regional_education_data.csv')
+    target_csv = os.path.join(target_folder, 'oecd_regional_education_data.csv')
+    unmapped_log_path = os.path.join(counters_folder, 'unresolved_places.csv')
 
+    if raw_file:
+        src_path = os.path.join(target_folder, raw_file)
+        tmp_path = os.path.join(target_folder, 'filtered_tmp.csv')
+        logging.info(f"Filtering '{raw_file}' into 'oecd_regional_education_data.csv'...")
+        _filter_csv(src_path, tmp_path, valid_places, unmapped_log_path=unmapped_log_path)
+        os.replace(tmp_path, target_csv)
+        # Preserve original downloaded raw file in GCS source_files per Data Commons guidelines
+        logging.info(f"Retained raw downloaded source file at '{src_path}'.")
+        logging.info("Preprocessing and filtering completed successfully.")
+    elif os.path.isfile(target_csv) and valid_places:
+        tmp_path = os.path.join(target_folder, 'filtered_tmp.csv')
+        logging.info(f"Checking and filtering existing '{target_csv}'...")
+        _filter_csv(target_csv, tmp_path, valid_places, unmapped_log_path=unmapped_log_path)
+        os.replace(tmp_path, target_csv)
+        logging.info("Filtering completed successfully.")
+    else:
+        logging.error("No matching source data file found to process.")
+        raise FileNotFoundError(
+            f"No candidate raw data file found to process in '{target_folder}'.")
+
+
+def _filter_csv(src_path: str, dst_path: str, valid_places: dict, unmapped_log_path: str = None):
+    required_columns = [
+        'REF_AREA',
+        'TIME_PERIOD',
+        'UNIT_MULT',
+        'SEX',
+        'Education level',
+        'AGE',
+        'OBS_VALUE',
+    ]
+    with open(src_path, 'r', encoding='utf-8', errors='replace') as fin, \
+         open(dst_path, 'w', encoding='utf-8', newline='') as fout:
+        reader = csv.reader(fin)
+        writer = csv.writer(fout)
+
+        header = next(reader, None)
+        if not header:
+            raise ValueError(f"Source file '{src_path}' is empty.")
+
+        ref_area_idx = header.index('REF_AREA') if 'REF_AREA' in header else None
+        if ref_area_idx is None:
+            logging.warning("REF_AREA column not found in header, copying all rows.")
+            writer.writerow(header)
+            kept = 0
+            for row in reader:
+                writer.writerow(row)
+                kept += 1
+            if kept == 0:
+                raise ValueError(f"Source file '{src_path}' has header but no data rows.")
+            return
+
+        # Determine indices of required columns if all exist in header
+        col_indices = [header.index(c) for c in required_columns if c in header]
+        use_subset = len(col_indices) == len(required_columns)
+        obs_val_idx = header.index('OBS_VALUE') if 'OBS_VALUE' in header else None
+        stat_op_idx = header.index('STATISTICAL_OPERATION') if 'STATISTICAL_OPERATION' in header else None
+
+        if use_subset:
+            writer.writerow(required_columns)
+            out_ref_area_idx = required_columns.index('REF_AREA')
+        else:
+            writer.writerow(header)
+            out_ref_area_idx = ref_area_idx
+
+        kept = 0
+        dropped = 0
+        unmapped_places = set()
+        for row in reader:
+            if len(row) > ref_area_idx:
+                # Skip rows with empty OBS_VALUE or ignored STATISTICAL_OPERATION (SE)
+                if obs_val_idx is not None and len(row) > obs_val_idx:
+                    if not row[obs_val_idx].strip():
+                        dropped += 1
+                        continue
+                if stat_op_idx is not None and len(row) > stat_op_idx:
+                    if row[stat_op_idx].strip() == 'SE':
+                        dropped += 1
+                        continue
+
+                ref_area = row[ref_area_idx].strip()
+                # Handle both raw ref_area codes and already-prefixed dcid values
+                clean_ref = ref_area
+                resolved_dcid = valid_places.get(clean_ref)
+                if not resolved_dcid and clean_ref.startswith('dcid:'):
+                    # Check reverse lookup if already dcid-prefixed
+                    resolved_dcid = clean_ref
+
+                if resolved_dcid:
+                    if use_subset:
+                        out_row = [row[idx] if len(row) > idx else '' for idx in col_indices]
+                    else:
+                        out_row = list(row)
+                    out_row[out_ref_area_idx] = resolved_dcid
+                    writer.writerow(out_row)
+                    kept += 1
+                else:
+                    dropped += 1
+                    unmapped_places.add(ref_area)
+            else:
+                dropped += 1
+
+        if kept == 0:
+            raise ValueError(
+                f"Critical: All {dropped} rows in '{src_path}' were filtered out! "
+                "Output CSV would be completely empty. Aborting preprocessing to prevent silent data drop.")
+
+        logging.info(f"Filtered source data: {kept} rows kept, {dropped} rows dropped.")
+        if unmapped_places:
+            logging.warning(
+                f"Encountered {len(unmapped_places)} unmapped REF_AREA codes. "
+                f"Sample unmapped places: {sorted(list(unmapped_places))[:25]}")
+            if unmapped_log_path:
                 try:
-                    os.rename(old_path, new_path)
-                    logging.info(f"Renamed '{filename}' to 'oecd_regional_education_data.csv'")
-                    renamed = True
-                except PermissionError:
-                    logging.warning(f"Permission denied while renaming '{filename}'.") # Changed to warning
-                except OSError as e:
-                    logging.fatal(f"OS error while renaming '{filename}': {e}") # Changed to error
-                break  # Rename only the first match
-
-        if not renamed:
-            logging.info("No matching file starting with 'A' found to rename.")
-
-    except FileNotFoundError as e:
-        # This block might not be hit if caught earlier, but good for other FileNotFoundError
-        logging.fatal(f"File system error: {e}")
-    except PermissionError as e:
-        # This block might not be hit if caught earlier, but good for other PermissionError
-        logging.fatal(f"Global permission error: {e}")
-    except Exception as e:
-        logging.fatal(f"An unexpected critical error occurred: {e}") # Changed to critical for unexpected errors
-
-# Run it
-rename_target_file()
+                    os.makedirs(os.path.dirname(unmapped_log_path), exist_ok=True)
+                    with open(unmapped_log_path, 'w', encoding='utf-8', newline='') as uf:
+                        u_writer = csv.writer(uf)
+                        u_writer.writerow(['unmapped_ref_area'])
+                        for p in sorted(unmapped_places):
+                            u_writer.writerow([p])
+                    logging.info(f"Wrote {len(unmapped_places)} unmapped places to {unmapped_log_path}")
+                except Exception as e:
+                    logging.warning(f"Failed to write unmapped places log: {e}")
+
+
+if __name__ == '__main__':
+    preprocess()

statvar_imports/oecd/regional_education/validation_config.json

@@ -0,0 +1,36 @@
+{
+    "schema_version": "1.0",
+    "rules": [
+        {
+            "rule_id": "check_deleted_records_percent",
+            "description": "Allow up to 5% deleted records due to OECD dataflow 2.5 NUTS 2024 regional restructuring and historical series revisions.",
+            "validator": "DELETED_RECORDS_PERCENT",
+            "params": {
+                "threshold": 5
+            }
+        },
+        {
+            "rule_id": "check_max_date_consistent",
+            "description": "Ensure MaxDate is uniform across all StatVars",
+            "validator": "MAX_DATE_CONSISTENT"
+        },
+        {
+            "rule_id": "check_max_date_freshness",
+            "description": "Ensure MaxDate is fresh within allowable 3-year reporting lag for OECD annual data",
+            "validator": "SQL_VALIDATOR",
+            "params": {
+                "query": "SELECT StatVar, MaxDate FROM stats",
+                "condition": "COALESCE(TRY_CAST(SUBSTRING(CAST(MaxDate AS VARCHAR), 1, 4) AS INTEGER) >= date_part('year', current_date) - 3, FALSE)"
+            }
+        },
+        {
+            "rule_id": "check_no_statvar_extinction",
+            "description": "Ensure that the 5% deletion budget did not completely eliminate all observations for any StatVar",
+            "validator": "SQL_VALIDATOR",
+            "params": {
+                "query": "SELECT COUNT(DISTINCT StatVar) AS active_statvars FROM stats",
+                "condition": "active_statvars >= 42"
+            }
+        }
+    ]
+}