ford_qc/checks/linkingrecord_header_check.py

import os
import re
import json
from typing import Dict, Any, List, Optional


def run_check(config: Dict[str, Any]) -> Dict[str, Any]:
    """
    Validate the top-level header fields in linkingrecord.json.

    Checks: market (whitelist), model (5-char alphanumeric uppercase), year (literal pattern),
    fullLoad (must be "Y"), header.specMarket (5 uppercase), header.ptvl (matches model prefix),
    header.numericalMmy (digits only, no dots).

    All rules are configurable via profile config keys — see ford_bnp.json for defaults.

    :param config: Profile config dict (see ford_bnp.json for available keys and defaults)
    :return: {"status": "passed"/"failed"/"error", "details": {...}}
    """
    working_dir = config.get("working_dir", "working")
    linkingrecord_filename = config.get("linkingrecord_filename", "linkingrecord.json")
    linkingrecord_path = os.path.join(working_dir, linkingrecord_filename)

    if not os.path.exists(linkingrecord_path):
        return {
            "status": "error",
            "error_message": f"Linking record file '{linkingrecord_filename}' not found in {working_dir}."
        }

    try:
        with open(linkingrecord_path, 'r', encoding='utf-8') as f:
            data = json.load(f)
    except json.JSONDecodeError as e:
        return {"status": "error", "error_message": f"Malformed JSON in '{linkingrecord_filename}': {str(e)}"}
    except Exception as e:
        return {"status": "error", "error_message": f"Error reading '{linkingrecord_filename}': {str(e)}"}

    if not isinstance(data, dict):
        return {"status": "error", "error_message": "Invalid linkingrecord.json structure: root is not an object."}

    market_whitelist: Optional[List[str]] = config.get("market_whitelist")
    model_pattern: str = config.get("model_pattern", r'^[A-Z0-9]{5}$')
    year_pattern: str = config.get("year_pattern", r'^YYY$')
    full_load_value: str = config.get("full_load_value", "Y")
    spec_market_pattern: str = config.get("spec_market_pattern", r'^[A-Z]{5}$')
    ptvl_match_model_prefix: bool = config.get("ptvl_match_model_prefix", True)
    numerical_mmy_pattern: str = config.get("numerical_mmy_pattern", r'^\d+$')

    model_re = re.compile(model_pattern)
    year_re = re.compile(year_pattern)
    spec_market_re = re.compile(spec_market_pattern)
    numerical_mmy_re = re.compile(numerical_mmy_pattern)

    errors: List[Dict[str, str]] = []
    header = data.get("header", {})

    def add_error(field: str, value: Any, error: str) -> None:
        errors.append({"field": field, "value": str(value), "error": error})

    # --- market ---
    market = data.get("market")
    if market is None:
        add_error("market", "", "Field missing from linkingrecord.json")
    elif not isinstance(market, str):
        add_error("market", market, "Must be a string")
    elif market_whitelist is not None:
        if market not in market_whitelist:
            add_error("market", market, f"Market code not in allowed list: {', '.join(sorted(market_whitelist))}")
    elif not re.match(r'^[A-Z]{3}$', market):
        add_error("market", market, "Must be exactly 3 uppercase alphabetic characters (A-Z)")

    # --- model ---
    model = data.get("model")
    model_valid = False
    if model is None:
        add_error("model", "", "Field missing from linkingrecord.json")
    elif not isinstance(model, str):
        add_error("model", model, "Must be a string")
    elif not model_re.match(model):
        add_error("model", model, f"Must match pattern '{model_pattern}' (exactly 5 alphanumeric uppercase characters)")
    else:
        model_valid = True

    # --- year ---
    year = data.get("year")
    if year is None:
        add_error("year", "", "Field missing from linkingrecord.json")
    elif not isinstance(year, str):
        add_error("year", year, "Must be a string")
    elif not year_re.match(year):
        add_error("year", year, f"Must match pattern '{year_pattern}'")

    # --- fullLoad (top-level, NOT inside header) ---
    full_load = data.get("fullLoad")
    if full_load is None:
        add_error("fullLoad", "", "Field missing from linkingrecord.json")
    elif not isinstance(full_load, str):
        add_error("fullLoad", full_load, "Must be a string")
    elif full_load.upper() != full_load_value.upper():
        add_error("fullLoad", full_load, f"Must be exactly '{full_load_value}' (full load packs only)")

    # --- header.specMarket ---
    spec_market = header.get("specMarket")
    if spec_market is None:
        add_error("header.specMarket", "", "Field missing from header section")
    elif not isinstance(spec_market, str):
        add_error("header.specMarket", spec_market, "Must be a string")
    elif not spec_market_re.match(spec_market):
        add_error("header.specMarket", spec_market, f"Must match pattern '{spec_market_pattern}' (5 uppercase alphabetic characters)")

    # --- header.ptvl ---
    ptvl = header.get("ptvl")
    if ptvl is None:
        add_error("header.ptvl", "", "Field missing from header section")
    elif not isinstance(ptvl, str):
        add_error("header.ptvl", ptvl, "Must be a string")
    elif ptvl_match_model_prefix:
        if model_valid:
            expected_ptvl = model[:3]
            if ptvl != expected_ptvl:
                add_error("header.ptvl", ptvl, f"Must match first 3 characters of model '{model}' (expected: '{expected_ptvl}')")
        # If model is invalid, skip ptvl prefix check — model error is already reported

    # --- header.numericalMmy ---
    numerical_mmy = header.get("numericalMmy")
    if numerical_mmy is None:
        add_error("header.numericalMmy", "", "Field missing from header section")
    elif not isinstance(numerical_mmy, str):
        add_error("header.numericalMmy", numerical_mmy, "Must be a string")
    elif not numerical_mmy_re.match(numerical_mmy):
        add_error("header.numericalMmy", numerical_mmy, f"Must match pattern '{numerical_mmy_pattern}' (digits only, no dots or other characters)")

    if errors:
        return {
            "status": "failed",
            "details": {
                "message": f"Linkingrecord header validation failed - {len(errors)} header field error{'s' if len(errors) != 1 else ''} found",
                "header_validation_errors": errors
            }
        }

    return {
        "status": "passed",
        "details": {
            "message": "Linkingrecord header fields are valid",
            "fields_checked": 7
        }
    }