validator.py

# Copyright 2025 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.
"""Module for the Validator class."""

import duckdb
import pandas as pd
import os
import sys

_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.append(_SCRIPT_DIR)
_DATA_DIR = os.path.dirname(os.path.dirname(_SCRIPT_DIR))
sys.path.append(os.path.join(_DATA_DIR, 'util'))

from result import ValidationResult, ValidationStatus
from counters import Counters
import validator_goldens


class Validator:
    """
  Contains the core logic for all validation rules.
  This class is stateless and does not interact with the filesystem.
  """

    def validate_sql(self, stats_df: pd.DataFrame, differ_df: pd.DataFrame,
                     params: dict) -> ValidationResult:
        """Runs a SQL query to validate the data.

    Args:
        stats_df: A DataFrame containing the summary statistics.
        differ_df: A DataFrame containing the differ output.
        params: A dictionary containing the validation parameters, which must
          have 'query' and 'condition' keys.

    Returns:
        A ValidationResult object.
    """
        if 'query' not in params or 'condition' not in params:
            return ValidationResult(
                ValidationStatus.CONFIG_ERROR,
                'SQL_VALIDATOR',
                message=
                "Configuration error: 'query' and 'condition' must be specified for SQL_VALIDATOR."
            )

        try:
            con = duckdb.connect(database=':memory:', read_only=False)
            con.register('stats', stats_df)
            con.register('differ', differ_df)

            final_query = f"""
            WITH data_to_validate AS (
                {params['query']}
            )
            SELECT *
            FROM data_to_validate
            WHERE NOT ({params['condition']})
            """
            failing_df = con.execute(final_query).fetchdf()

            if failing_df.empty:
                return ValidationResult(ValidationStatus.PASSED,
                                        'SQL_VALIDATOR')

            return ValidationResult(
                ValidationStatus.FAILED,
                'SQL_VALIDATOR',
                message=f"{len(failing_df)} rows failed the SQL validation.",
                details={'failing_rows': failing_df.to_dict('records')})

        except duckdb.Error as e:
            return ValidationResult(ValidationStatus.CONFIG_ERROR,
                                    'SQL_VALIDATOR',
                                    message=f"SQL Error: {e}")

    def validate_max_date_latest(self, stats_df: pd.DataFrame,
                                 params: dict) -> ValidationResult:
        """Checks that the MaxDate in the stats summary is from the current year.

    Args:
      stats_df: A DataFrame containing the summary statistics, expected to have
        a 'MaxDate' column.
      params: A dictionary containing the validation parameters.

    Returns:
      A ValidationResult object.
    """
        if 'MaxDate' not in stats_df.columns:
            return ValidationResult(
                ValidationStatus.DATA_ERROR,
                'MAX_DATE_LATEST',
                message="Input data is missing required column: 'MaxDate'.")

        rows_processed = len(stats_df)
        if stats_df.empty:
            return ValidationResult(ValidationStatus.PASSED,
                                    'MAX_DATE_LATEST',
                                    details={
                                        'rows_processed': 0,
                                        'rows_succeeded': 0,
                                        'rows_failed': 0
                                    })

        stats_df['MaxDate'] = pd.to_datetime(stats_df['MaxDate'])
        max_date_year = stats_df['MaxDate'].dt.year.max()
        current_year = pd.to_datetime('today').year

        if max_date_year < current_year:
            return ValidationResult(
                ValidationStatus.FAILED,
                'MAX_DATE_LATEST',
                message=
                f"Latest date found was {max_date_year}, expected {current_year}.",
                details={
                    'latest_date_found': int(max_date_year),
                    'expected_latest_date': int(current_year),
                    'rows_processed': rows_processed,
                    'rows_succeeded': 0,
                    'rows_failed': rows_processed
                })
        return ValidationResult(ValidationStatus.PASSED,
                                'MAX_DATE_LATEST',
                                details={
                                    'rows_processed': rows_processed,
                                    'rows_succeeded': rows_processed,
                                    'rows_failed': 0
                                })

    def validate_deleted_records_count(self, differ_df: pd.DataFrame,
                                       params: dict) -> ValidationResult:
        """Checks if the total number of deleted points is within a threshold.

    Args:
      differ_df: A DataFrame containing the differ output, expected to have a
        'DELETED' column.
      params: A dictionary containing the validation parameters, which may
        have a 'threshold' key.

    Returns:
      A ValidationResult object.
    """
        if differ_df.empty:
            deleted_records_count = 0
            threshold = params.get('threshold', 0)
            if deleted_records_count > threshold:
                return ValidationResult(
                    ValidationStatus.FAILED,
                    'DELETED_RECORDS_COUNT',
                    message=
                    f"Found {deleted_records_count} deleted points, which is over the threshold of {threshold}.",
                    details={
                        'deleted_records_count': int(deleted_records_count),
                        'threshold': threshold,
                        'rows_processed': 0,
                        'rows_succeeded': 0,
                        'rows_failed': 0
                    })
            return ValidationResult(ValidationStatus.PASSED,
                                    'DELETED_RECORDS_COUNT',
                                    details={
                                        'rows_processed': 0,
                                        'rows_succeeded': 0,
                                        'rows_failed': 0
                                    })

        if 'DELETED' not in differ_df.columns:
            return ValidationResult(
                ValidationStatus.DATA_ERROR,
                'DELETED_RECORDS_COUNT',
                message="Input data is missing required column: 'DELETED'.")

        rows_processed = len(differ_df)
        threshold = params.get('threshold', 0)
        deleted_records_count = differ_df['DELETED'].sum()

        if deleted_records_count > threshold:
            return ValidationResult(
                ValidationStatus.FAILED,
                'DELETED_RECORDS_COUNT',
                message=
                f"Found {deleted_records_count} deleted points, which is over the threshold of {threshold}.",
                details={
                    'deleted_records_count': int(deleted_records_count),
                    'threshold': threshold,
                    'rows_processed': rows_processed,
                    'rows_succeeded': 0,
                    'rows_failed': rows_processed
                })
        return ValidationResult(ValidationStatus.PASSED,
                                'DELETED_RECORDS_COUNT',
                                details={
                                    'rows_processed': rows_processed,
                                    'rows_succeeded': rows_processed,
                                    'rows_failed': 0
                                })

    def validate_deleted_records_percent(self, differ_df: pd.DataFrame,
                                         summary: dict,
                                         params: dict) -> ValidationResult:
        """Checks if the percentage of deleted records is within a threshold.

    Args:
      differ_df: A DataFrame containing the differ output.
      summary: A dictionary containing the differ summary.
      params: A dictionary containing the validation parameters, which may
        have a 'threshold' key.

    Returns:
      A ValidationResult object.
    """
        if differ_df is None:
            return ValidationResult(ValidationStatus.DATA_ERROR,
                                    'DELETED_RECORDS_PERCENT',
                                    message="Differ DataFrame is missing.")

        if summary is None:
            return ValidationResult(ValidationStatus.DATA_ERROR,
                                    'DELETED_RECORDS_PERCENT',
                                    message="Differ summary is missing.")

        if 'previous_obs_size' not in summary:
            return ValidationResult(
                ValidationStatus.DATA_ERROR,
                'DELETED_RECORDS_PERCENT',
                message=
                "Differ summary is missing required field: 'previous_obs_size'."
            )

        previous_obs_size = summary['previous_obs_size']

        if differ_df.empty:
            deleted_records_count = 0
        elif 'DELETED' not in differ_df.columns:
            return ValidationResult(
                ValidationStatus.DATA_ERROR,
                'DELETED_RECORDS_PERCENT',
                message="Input data is missing required column: 'DELETED'.")
        else:
            deleted_records_count = differ_df['DELETED'].sum()

        if previous_obs_size == 0:
            if deleted_records_count > 0:
                percent = 100.0
            else:
                percent = 0.0
        else:
            percent = (deleted_records_count / previous_obs_size) * 100

        threshold = params.get('threshold', 0)

        if percent > threshold:
            return ValidationResult(
                ValidationStatus.FAILED,
                'DELETED_RECORDS_PERCENT',
                message=
                f"Found {percent:.2f}% deleted records, which is over the threshold of {threshold}%.",
                details={
                    'deleted_records_count': int(deleted_records_count),
                    'previous_obs_size': int(previous_obs_size),
                    'percent': percent,
                    'threshold': threshold
                })

        return ValidationResult(
            ValidationStatus.PASSED,
            'DELETED_RECORDS_PERCENT',
            details={
                'deleted_records_count': int(deleted_records_count),
                'previous_obs_size': int(previous_obs_size),
                'percent': percent,
                'threshold': threshold
            })

    def validate_empty_import(self, differ_df: pd.DataFrame, summary: dict,
                              params: dict) -> ValidationResult:
        """Checks if the import is empty (no observations and no schema).

    Args:
      differ_df: A DataFrame containing the differ output (unused but passed for consistency).
      summary: A dictionary containing the differ summary.
      params: A dictionary containing the validation parameters.

    Returns:
      A ValidationResult object.
    """
        if summary is None:
            return ValidationResult(ValidationStatus.DATA_ERROR,
                                    'EMPTY_IMPORT_CHECK',
                                    message="Differ summary is missing.")

        current_obs_size = summary['current_obs_size']
        current_schema_size = summary['current_schema_size']

        if current_obs_size == 0 and current_schema_size == 0:
            return ValidationResult(
                ValidationStatus.FAILED,
                'EMPTY_IMPORT_CHECK',
                message=
                "The import is empty: both current_obs_size and current_schema_size are 0.",
                details={
                    'current_obs_size': int(current_obs_size),
                    'current_schema_size': int(current_schema_size)
                })

        return ValidationResult(
            ValidationStatus.PASSED,
            'EMPTY_IMPORT_CHECK',
            details={
                'current_obs_size': int(current_obs_size),
                'current_schema_size': int(current_schema_size)
            })

    def validate_missing_refs_count(self, report: dict,
                                    params: dict) -> ValidationResult:
        """Checks if the total number of missing references is within a threshold.

    Args:
      report: A json object containing the lint report
      params: A dictionary containing the validation parameters, which may
        have a 'threshold' key.

    Returns:
      A ValidationResult object.
    """
        missing_refs_count = 0
        if report:
            counters = report.get('levelSummary',
                                  {}).get('LEVEL_WARNING',
                                          {}).get('counters', {})
            missing_refs_count = sum(
                int(value)
                for key, value in counters.items()
                if key.startswith('Existence_MissingReference'))
            threshold = params.get('threshold', 0)
            if missing_refs_count > threshold:
                return ValidationResult(
                    ValidationStatus.FAILED,
                    'MISSING_REFS_COUNT',
                    message=
                    f"Found {missing_refs_count} missing references, which is over the threshold of {threshold}.",
                    details={'missing_refs_count': missing_refs_count})
        return ValidationResult(
            ValidationStatus.PASSED,
            'MISSING_REFS_COUNT',
            details={'missing_refs_count': missing_refs_count})

    def validate_lint_error_count(self, report: dict,
                                  params: dict) -> ValidationResult:
        """Checks if the total number of lint errors is within a threshold.

    Args:
      report: A json object containing the lint report
      params: A dictionary containing the validation parameters, which may
        have a 'threshold' key.

    Returns:
      A ValidationResult object.
    """
        lint_error_count = 0
        if report:
            counters = report.get('levelSummary',
                                  {}).get('LEVEL_ERROR',
                                          {}).get('counters', {})
            lint_error_count = sum(int(value) for value in counters.values())
            threshold = params.get('threshold', 0)
            if lint_error_count > threshold:
                return ValidationResult(
                    ValidationStatus.FAILED,
                    'LINT_ERROR_COUNT',
                    message=
                    f"Found {lint_error_count} lint errors, which is over the threshold of {threshold}.",
                    details={'lint_error_count': lint_error_count})
        return ValidationResult(ValidationStatus.PASSED,
                                'LINT_ERROR_COUNT',
                                details={'lint_error_count': lint_error_count})

    def validate_modified_records_count(self, differ_df: pd.DataFrame,
                                        params: dict) -> ValidationResult:
        """Checks if the number of modified points is the same for all StatVars.

    Args:
      differ_df: A DataFrame containing the differ output, expected to have a
        'MODIFIED' column.
      params: A dictionary containing the validation parameters.

    Returns:
      A ValidationResult object.
    """
        if differ_df.empty:
            return ValidationResult(ValidationStatus.PASSED,
                                    'MODIFIED_RECORDS_COUNT',
                                    details={
                                        'rows_processed': 0,
                                        'rows_succeeded': 0,
                                        'rows_failed': 0
                                    })

        if 'MODIFIED' not in differ_df.columns:
            return ValidationResult(
                ValidationStatus.DATA_ERROR,
                'MODIFIED_RECORDS_COUNT',
                message="Input data is missing required column: 'MODIFIED'.")

        rows_processed = len(differ_df)
        unique_counts = differ_df['MODIFIED'].nunique()

        if unique_counts > 1:
            return ValidationResult(
                ValidationStatus.FAILED,
                'MODIFIED_RECORDS_COUNT',
                message=
                "The number of modified data points is not consistent across all StatVars",
                details={
                    'distinct_statvar_count': differ_df['StatVar'].nunique(),
                    'distinct_modified_records_count': unique_counts,
                    'rows_processed': rows_processed,
                    'rows_succeeded': 0,
                    'rows_failed': rows_processed
                })
        return ValidationResult(ValidationStatus.PASSED,
                                'MODIFIED_RECORDS_COUNT',
                                details={
                                    'rows_processed': rows_processed,
                                    'rows_succeeded': rows_processed,
                                    'rows_failed': 0
                                })

    def validate_added_records_count(self, differ_df: pd.DataFrame,
                                     params: dict) -> ValidationResult:
        """Checks if the number of added points is the same for all StatVars.

    Args:
      differ_df: A DataFrame containing the differ output, expected to have an
        'ADDED' column.
      params: A dictionary containing the validation parameters.

    Returns:
      A ValidationResult object.
    """
        if differ_df.empty:
            return ValidationResult(ValidationStatus.PASSED,
                                    'ADDED_RECORDS_COUNT',
                                    details={
                                        'rows_processed': 0,
                                        'rows_succeeded': 0,
                                        'rows_failed': 0
                                    })

        if 'ADDED' not in differ_df.columns:
            return ValidationResult(
                ValidationStatus.DATA_ERROR,
                'ADDED_RECORDS_COUNT',
                message="Input data is missing required column: 'ADDED'.")

        rows_processed = len(differ_df)
        unique_counts = differ_df['ADDED'].nunique()

        if unique_counts > 1:
            return ValidationResult(
                ValidationStatus.FAILED,
                'ADDED_RECORDS_COUNT',
                message=
                "The number of added data points is not consistent across all StatVars.",
                details={
                    'distinct_statvar_count': differ_df['StatVar'].nunique(),
                    'distinct_added_records_count': unique_counts,
                    'rows_processed': rows_processed,
                    'rows_succeeded': 0,
                    'rows_failed': rows_processed
                })
        return ValidationResult(ValidationStatus.PASSED,
                                'ADDED_RECORDS_COUNT',
                                details={
                                    'rows_processed': rows_processed,
                                    'rows_succeeded': rows_processed,
                                    'rows_failed': 0
                                })

    def validate_num_places_consistent(self, stats_df: pd.DataFrame,
                                       params: dict) -> ValidationResult:
        """Checks if the number of places is the same for all StatVars.

    Args:
      stats_df: A DataFrame containing the summary statistics, expected to have
        a 'NumPlaces' column.
      params: A dictionary containing the validation parameters.

    Returns:
      A ValidationResult object.
    """
        if 'NumPlaces' not in stats_df.columns:
            return ValidationResult(
                ValidationStatus.DATA_ERROR,
                'NUM_PLACES_CONSISTENT',
                message="Input data is missing required column: 'NumPlaces'.")

        rows_processed = len(stats_df)
        if stats_df.empty:
            return ValidationResult(ValidationStatus.PASSED,
                                    'NUM_PLACES_CONSISTENT',
                                    details={
                                        'rows_processed': 0,
                                        'rows_succeeded': 0,
                                        'rows_failed': 0
                                    })

        unique_counts = stats_df['NumPlaces'].nunique()

        if unique_counts > 1:
            return ValidationResult(
                ValidationStatus.FAILED,
                'NUM_PLACES_CONSISTENT',
                message=
                "The number of places is not consistent across all StatVars.",
                details={
                    'distinct_statvar_count': stats_df['StatVar'].nunique(),
                    'distinct_place_count': unique_counts,
                    'rows_processed': rows_processed,
                    'rows_succeeded': 0,
                    'rows_failed': rows_processed
                })
        return ValidationResult(ValidationStatus.PASSED,
                                'NUM_PLACES_CONSISTENT',
                                details={
                                    'rows_processed': rows_processed,
                                    'rows_succeeded': rows_processed,
                                    'rows_failed': 0
                                })

    def validate_num_places_count(self, stats_df: pd.DataFrame,
                                  params: dict) -> ValidationResult:
        """Checks if the number of places for each StatVar is within a defined range.

    The range can be specified using 'minimum', 'maximum', or an exact 'value'
    in the params.

    Args:
      stats_df: A DataFrame containing the summary statistics, expected to have
        'NumPlaces' and 'StatVar' columns.
      params: A dictionary containing the validation parameters.

    Returns:
      A ValidationResult object.
    """
        if 'NumPlaces' not in stats_df.columns:
            return ValidationResult(
                ValidationStatus.DATA_ERROR,
                'NUM_PLACES_COUNT',
                message="Input data is missing required column: 'NumPlaces'.")
        if stats_df.empty:
            return ValidationResult(ValidationStatus.PASSED,
                                    'NUM_PLACES_COUNT',
                                    details={
                                        'rows_processed': 0,
                                        'rows_succeeded': 0,
                                        'rows_failed': 0
                                    })

        min_val = params.get('minimum')
        max_val = params.get('maximum')
        exact_val = params.get('value')

        rows_processed = len(stats_df)
        rows_failed = 0
        failed_rows_details = []

        for _, row in stats_df.iterrows():
            value = row['NumPlaces']
            stat_var = row.get('StatVar', 'Unknown')
            failed = False

            if exact_val is not None and value != exact_val:
                failed = True
                failed_rows_details.append({
                    'stat_var': stat_var,
                    'actual_value': value,
                    'expected_value': exact_val,
                    'reason': f"Expected exactly {exact_val}"
                })
            elif min_val is not None and value < min_val:
                failed = True
                failed_rows_details.append({
                    'stat_var': stat_var,
                    'actual_value': value,
                    'minimum': min_val,
                    'reason': f"Below minimum of {min_val}"
                })
            elif max_val is not None and value > max_val:
                failed = True
                failed_rows_details.append({
                    'stat_var': stat_var,
                    'actual_value': value,
                    'maximum': max_val,
                    'reason': f"Above maximum of {max_val}"
                })

            if failed:
                rows_failed += 1

        rows_succeeded = rows_processed - rows_failed

        if rows_failed > 0:
            return ValidationResult(
                ValidationStatus.FAILED,
                'NUM_PLACES_COUNT',
                message=
                f"{rows_failed} out of {rows_processed} rows failed the range check.",
                details={
                    'failed_rows': failed_rows_details,
                    'rows_processed': rows_processed,
                    'rows_succeeded': rows_succeeded,
                    'rows_failed': rows_failed
                })

        return ValidationResult(ValidationStatus.PASSED,
                                'NUM_PLACES_COUNT',
                                details={
                                    'rows_processed': rows_processed,
                                    'rows_succeeded': rows_succeeded,
                                    'rows_failed': rows_failed
                                })

    def validate_min_value_check(self, stats_df: pd.DataFrame,
                                 params: dict) -> ValidationResult:
        """Checks if the MinValue for each StatVar is not below a defined minimum.

    Args:
      stats_df: A DataFrame containing the summary statistics, expected to have
        'MinValue' and 'StatVar' columns.
      params: A dictionary containing the validation parameters, which must
        have a 'minimum' key.

    Returns:
      A ValidationResult object.
    """
        if 'minimum' not in params:
            return ValidationResult(
                ValidationStatus.CONFIG_ERROR,
                'MIN_VALUE_CHECK',
                message="Configuration error: 'minimum' key not specified.")
        if 'MinValue' not in stats_df.columns:
            return ValidationResult(
                ValidationStatus.DATA_ERROR,
                'MIN_VALUE_CHECK',
                message="Input data is missing required column: 'MinValue'.")
        if stats_df.empty:
            return ValidationResult(ValidationStatus.PASSED,
                                    'MIN_VALUE_CHECK',
                                    details={
                                        'rows_processed': 0,
                                        'rows_succeeded': 0,
                                        'rows_failed': 0
                                    })

        min_val = params['minimum']
        rows_processed = len(stats_df)
        rows_failed = 0
        failed_rows_details = []

        for _, row in stats_df.iterrows():
            min_value = row['MinValue']
            stat_var = row.get('StatVar', 'Unknown')

            if min_value < min_val:
                rows_failed += 1
                failed_rows_details.append({
                    'stat_var': stat_var,
                    'actual_min_value': min_value,
                    'minimum': min_val
                })

        rows_succeeded = rows_processed - rows_failed

        if rows_failed > 0:
            return ValidationResult(
                ValidationStatus.FAILED,
                'MIN_VALUE_CHECK',
                message=
                f"{rows_failed} out of {rows_processed} StatVars failed the minimum value check.",
                details={
                    'failed_rows': failed_rows_details,
                    'rows_processed': rows_processed,
                    'rows_succeeded': rows_succeeded,
                    'rows_failed': rows_failed
                })

        return ValidationResult(ValidationStatus.PASSED,
                                'MIN_VALUE_CHECK',
                                details={
                                    'rows_processed': rows_processed,
                                    'rows_succeeded': rows_succeeded,
                                    'rows_failed': rows_failed
                                })

    def validate_max_date_consistent(self, stats_df: pd.DataFrame,
                                     params: dict) -> ValidationResult:
        """Checks if the MaxDate is the same for all StatVars.

    Args:
      stats_df: A DataFrame containing the summary statistics, expected to have
        a 'MaxDate' column.
      params: A dictionary containing the validation parameters.

    Returns:
      A ValidationResult object.
    """
        if 'MaxDate' not in stats_df.columns:
            return ValidationResult(
                ValidationStatus.DATA_ERROR,
                'MAX_DATE_CONSISTENT',
                message="Input data is missing required column: 'MaxDate'.")

        rows_processed = len(stats_df)
        if stats_df.empty:
            return ValidationResult(ValidationStatus.PASSED,
                                    'MAX_DATE_CONSISTENT',
                                    details={
                                        'rows_processed': 0,
                                        'rows_succeeded': 0,
                                        'rows_failed': 0
                                    })

        unique_dates = stats_df['MaxDate'].nunique()

        if unique_dates > 1:
            return ValidationResult(
                ValidationStatus.FAILED,
                'MAX_DATE_CONSISTENT',
                message="The MaxDate is not consistent across all StatVars.",
                details={
                    'distinct_statvar_count': stats_df['StatVar'].nunique(),
                    'distinct_max_date_count': unique_dates,
                    'rows_processed': rows_processed,
                    'rows_succeeded': 0,
                    'rows_failed': rows_processed
                })
        return ValidationResult(ValidationStatus.PASSED,
                                'MAX_DATE_CONSISTENT',
                                details={
                                    'rows_processed': rows_processed,
                                    'rows_succeeded': rows_processed,
                                    'rows_failed': 0
                                })

    def validate_num_observations_check(self, stats_df: pd.DataFrame,
                                        params: dict) -> ValidationResult:
        """Checks if the number of observations for each StatVar is within a defined range.

    The range can be specified using 'minimum', 'maximum', or an exact 'value'
    in the params.

    Args:
      stats_df: A DataFrame containing the summary statistics, expected to have
        'NumObservations' and 'StatVar' columns.
      params: A dictionary containing the validation parameters.

    Returns:
      A ValidationResult object.
    """
        if 'NumObservations' not in stats_df.columns:
            return ValidationResult(
                ValidationStatus.DATA_ERROR,
                'NUM_OBSERVATIONS_CHECK',
                message=
                "Input data is missing required column: 'NumObservations'.")
        if stats_df.empty:
            return ValidationResult(ValidationStatus.PASSED,
                                    'NUM_OBSERVATIONS_CHECK',
                                    details={
                                        'rows_processed': 0,
                                        'rows_succeeded': 0,
                                        'rows_failed': 0
                                    })

        min_val = params.get('minimum')
        max_val = params.get('maximum')
        exact_val = params.get('value')

        rows_processed = len(stats_df)
        rows_failed = 0
        failed_rows_details = []

        for _, row in stats_df.iterrows():
            value = row['NumObservations']
            stat_var = row.get('StatVar', 'Unknown')
            failed = False

            if exact_val is not None and value != exact_val:
                failed = True
                failed_rows_details.append({
                    'stat_var': stat_var,
                    'actual_value': value,
                    'expected_value': exact_val,
                    'reason': f"Expected exactly {exact_val}"
                })
            elif min_val is not None and value < min_val:
                failed = True
                failed_rows_details.append({
                    'stat_var': stat_var,
                    'actual_value': value,
                    'minimum': min_val,
                    'reason': f"Below minimum of {min_val}"
                })
            elif max_val is not None and value > max_val:
                failed = True
                failed_rows_details.append({
                    'stat_var': stat_var,
                    'actual_value': value,
                    'maximum': max_val,
                    'reason': f"Above maximum of {max_val}"
                })

            if failed:
                rows_failed += 1

        rows_succeeded = rows_processed - rows_failed

        if rows_failed > 0:
            return ValidationResult(
                ValidationStatus.FAILED,
                'NUM_OBSERVATIONS_CHECK',
                message=
                f"{rows_failed} out of {rows_processed} rows failed the range check.",
                details={
                    'failed_rows': failed_rows_details,
                    'rows_processed': rows_processed,
                    'rows_succeeded': rows_succeeded,
                    'rows_failed': rows_failed
                })

        return ValidationResult(ValidationStatus.PASSED,
                                'NUM_OBSERVATIONS_CHECK',
                                details={
                                    'rows_processed': rows_processed,
                                    'rows_succeeded': rows_succeeded,
                                    'rows_failed': rows_failed
                                })

    def validate_unit_consistency(self, stats_df: pd.DataFrame,
                                  params: dict) -> ValidationResult:
        """Checks if the unit is the same for all StatVars.

    Args:
      stats_df: A DataFrame containing the summary statistics, expected to have
        a 'Units' column.
      params: A dictionary containing the validation parameters.

    Returns:
      A ValidationResult object.
    """
        if 'Units' not in stats_df.columns:
            return ValidationResult(
                ValidationStatus.DATA_ERROR,
                'UNIT_CONSISTENCY_CHECK',
                message="Input data is missing required column: 'Units'.")

        rows_processed = len(stats_df)
        if stats_df.empty:
            return ValidationResult(ValidationStatus.PASSED,
                                    'UNIT_CONSISTENCY_CHECK',
                                    details={
                                        'rows_processed': 0,
                                        'rows_succeeded': 0,
                                        'rows_failed': 0
                                    })

        unique_units = stats_df['Units'].nunique()

        if unique_units > 1:
            return ValidationResult(
                ValidationStatus.FAILED,
                'UNIT_CONSISTENCY_CHECK',
                message="The unit is not consistent across all StatVars.",
                details={
                    'distinct_statvar_count': stats_df['StatVar'].nunique(),
                    'distinct_unit_count': unique_units,
                    'rows_processed': rows_processed,
                    'rows_succeeded': 0,
                    'rows_failed': rows_processed
                })
        return ValidationResult(ValidationStatus.PASSED,
                                'UNIT_CONSISTENCY_CHECK',
                                details={
                                    'rows_processed': rows_processed,
                                    'rows_succeeded': rows_processed,
                                    'rows_failed': 0
                                })

    def validate_max_value_check(self, stats_df: pd.DataFrame,
                                 params: dict) -> ValidationResult:
        """Checks if the MaxValue for each StatVar is not above a defined maximum.

    Args:
      stats_df: A DataFrame containing the summary statistics, expected to have
        'MaxValue' and 'StatVar' columns.
      params: A dictionary containing the validation parameters, which must
        have a 'maximum' key.

    Returns:
      A ValidationResult object.
    """
        if 'maximum' not in params:
            return ValidationResult(
                ValidationStatus.CONFIG_ERROR,
                'MAX_VALUE_CHECK',
                message="Configuration error: 'maximum' key not specified.")
        if 'MaxValue' not in stats_df.columns:
            return ValidationResult(
                ValidationStatus.DATA_ERROR,
                'MAX_VALUE_CHECK',
                message="Input data is missing required column: 'MaxValue'.")
        if stats_df.empty:
            return ValidationResult(ValidationStatus.PASSED,
                                    'MAX_VALUE_CHECK',
                                    details={
                                        'rows_processed': 0,
                                        'rows_succeeded': 0,
                                        'rows_failed': 0
                                    })

        max_val = params['maximum']
        rows_processed = len(stats_df)
        rows_failed = 0
        failed_rows_details = []

        for _, row in stats_df.iterrows():
            max_value = row['MaxValue']
            stat_var = row.get('StatVar', 'Unknown')

            if max_value > max_val:
                rows_failed += 1
                failed_rows_details.append({
                    'stat_var': stat_var,
                    'actual_max_value': max_value,
                    'maximum': max_val
                })

        rows_succeeded = rows_processed - rows_failed

        if rows_failed > 0:
            return ValidationResult(
                ValidationStatus.FAILED,
                'MAX_VALUE_CHECK',
                message=
                f"{rows_failed} out of {rows_processed} StatVars failed the maximum value check.",
                details={
                    'failed_rows': failed_rows_details,
                    'rows_processed': rows_processed,
                    'rows_succeeded': rows_succeeded,
                    'rows_failed': rows_failed
                })

        return ValidationResult(ValidationStatus.PASSED,
                                'MAX_VALUE_CHECK',
                                details={
                                    'rows_processed': rows_processed,
                                    'rows_succeeded': rows_succeeded,
                                    'rows_failed': rows_failed
                                })

    def validate_goldens(self, df: pd.DataFrame,
                         params: dict) -> ValidationResult:
        """Validates records against a golden set.

        Args:
          df: A DataFrame containing the data to validate (used if input_files
            is not provided in params).
          params: A dictionary containing:
            'golden_files': Path(s) to golden MCF/CSV files.
            'input_files': (Optional) Path(s) to input files. If not provided,
                           the 'df' will be used.
            'output_path': (Optional) folder or output filename to save missing goldens.
            And other optional validator_goldens config (e.g., goldens_key_property).

        Returns:
          A ValidationResult object.
        """
        golden_files = params.get('golden_files')
        if not golden_files:
            return ValidationResult(
                ValidationStatus.CONFIG_ERROR,
                'GOLDENS_CHECK',
                message=
                "Configuration error: 'golden_files' must be specified for GOLDENS_CHECK validator."
            )

        try:
            inputs = params.get('input_files')
            if not inputs:
                inputs = df.to_dict('index')
            output_path = params.get('output_path')
            # Compare nodes
            counters = Counters()
            missing_goldens = validator_goldens.validate_goldens(
                inputs,
                golden_files,
                output_path,
                config=params,
                counters=counters)
            details = {
                name: value
                for name, value in counters.get_counters().items()
                if 'golden' in name
            }
            if not missing_goldens:
                return ValidationResult(ValidationStatus.PASSED,
                                        'GOLDENS_CHECK',
                                        details=details)
            details['missing_goldens'] = missing_goldens

            return ValidationResult(
                ValidationStatus.FAILED,
                'GOLDENS_CHECK',
                message=f"Found {len(missing_goldens)} missing golden records.",
                details=details)

        except IOError as e:
            return ValidationResult(
                ValidationStatus.DATA_ERROR,
                'GOLDENS_CHECK',
                message=f"Error during golden validation: {e}")

    def validate_counter_zero(self, counters: dict,
                              params: dict) -> ValidationResult:
        """Checks that a specific counter is strictly zero.

    Args:
      counters: A dictionary containing the counter values.
      params: A dictionary containing the validation parameters, which must
        have a 'counter_name' key.

    Returns:
      A ValidationResult object.
    """
        if 'counter_name' not in params:
            return ValidationResult(
                ValidationStatus.CONFIG_ERROR,
                'COUNTER_ZERO_CHECK',
                message="Configuration error: 'counter_name' key not specified."
            )

        counter_name = params['counter_name']
        value = counters.get(counter_name, 0)

        if value > 0:
            return ValidationResult(
                ValidationStatus.FAILED,
                'COUNTER_ZERO_CHECK',
                message=f"Counter '{counter_name}' is {value}, expected 0.",
                details={
                    'counter_name': counter_name,
                    'actual_value': value,
                    'expected_value': 0
                })

        return ValidationResult(ValidationStatus.PASSED,
                                'COUNTER_ZERO_CHECK',
                                details={
                                    'counter_name': counter_name,
                                    'actual_value': value,
                                    'expected_value': 0
                                })

    def validate_counter_max_threshold(self, counters: dict,
                                       params: dict) -> ValidationResult:
        """Checks that a specific counter value does not exceed a maximum threshold.

    Args:
      counters: A dictionary containing the counter values.
      params: A dictionary containing the validation parameters, which must
        have 'counter_name' and 'threshold' keys.

    Returns:
      A ValidationResult object.
    """
        if 'counter_name' not in params or 'threshold' not in params:
            return ValidationResult(
                ValidationStatus.CONFIG_ERROR,
                'COUNTER_MAX_THRESHOLD',
                message=
                "Configuration error: 'counter_name' and 'threshold' must be specified."
            )

        counter_name = params['counter_name']
        threshold = params['threshold']
        value = counters.get(counter_name, 0)

        if value > threshold:
            return ValidationResult(
                ValidationStatus.FAILED,
                'COUNTER_MAX_THRESHOLD',
                message=
                f"Counter '{counter_name}' is {value}, which exceeds the threshold of {threshold}.",
                details={
                    'counter_name': counter_name,
                    'actual_value': value,
                    'threshold': threshold
                })

        return ValidationResult(ValidationStatus.PASSED,
                                'COUNTER_MAX_THRESHOLD',
                                details={
                                    'counter_name': counter_name,
                                    'actual_value': value,
                                    'threshold': threshold
                                })

    def validate_counter_ratio_threshold(self, counters: dict,
                                         params: dict) -> ValidationResult:
        """Checks that the ratio of two counters (percentage) does not exceed a threshold.

    Args:
      counters: A dictionary containing the counter values.
      params: A dictionary containing the validation parameters, which must
        have 'subset_counter', 'total_counter', and 'threshold_percent' keys.

    Returns:
      A ValidationResult object.
    """
        if 'subset_counter' not in params or 'total_counter' not in params or 'threshold_percent' not in params:
            return ValidationResult(
                ValidationStatus.CONFIG_ERROR,
                'COUNTER_RATIO_THRESHOLD',
                message=
                "Configuration error: 'subset_counter', 'total_counter', and 'threshold_percent' must be specified."
            )

        subset_counter = params['subset_counter']
        total_counter = params['total_counter']
        threshold_percent = params['threshold_percent']

        subset_value = counters.get(subset_counter, 0)
        total_value = counters.get(total_counter, 0)

        if total_value == 0:
            if subset_value > 0:
                percent = 100.0
            else:
                percent = 0.0
        else:
            percent = (subset_value / total_value) * 100

        if percent > threshold_percent:
            return ValidationResult(
                ValidationStatus.FAILED,
                'COUNTER_RATIO_THRESHOLD',
                message=
                f"Ratio of '{subset_counter}' to '{total_counter}' is {percent:.2f}%, which exceeds the threshold of {threshold_percent}%.",
                details={
                    'subset_counter': subset_counter,
                    'subset_value': subset_value,
                    'total_counter': total_counter,
                    'total_value': total_value,
                    'percent': percent,
                    'threshold_percent': threshold_percent
                })

        return ValidationResult(ValidationStatus.PASSED,
                                'COUNTER_RATIO_THRESHOLD',
                                details={
                                    'subset_counter': subset_counter,
                                    'subset_value': subset_value,
                                    'total_counter': total_counter,
                                    'total_value': total_value,
                                    'percent': percent,
                                    'threshold_percent': threshold_percent
                                })

    def validate_counter_sum_integrity(self, counters: dict,
                                       params: dict) -> ValidationResult:
        """Checks that the sum of constituent counters equals a total counter.

    Args:
      counters: A dictionary containing the counter values.
      params: A dictionary containing the validation parameters, which must
        have 'total_counter' and 'constituent_counters' keys.

    Returns:
      A ValidationResult object.
    """
        if 'total_counter' not in params or 'constituent_counters' not in params:
            return ValidationResult(
                ValidationStatus.CONFIG_ERROR,
                'COUNTER_SUM_INTEGRITY',
                message=
                "Configuration error: 'total_counter' and 'constituent_counters' must be specified."
            )

        total_counter = params['total_counter']
        constituent_counters = params['constituent_counters']

        total_value = counters.get(total_counter, 0)
        constituent_sum = sum(
            counters.get(c, 0) for c in constituent_counters)

        if total_value != constituent_sum:
            return ValidationResult(
                ValidationStatus.FAILED,
                'COUNTER_SUM_INTEGRITY',
                message=
                f"Sum of constituent counters ({constituent_sum}) does not equal total counter '{total_counter}' ({total_value}).",
                details={
                    'total_counter': total_counter,
                    'total_value': total_value,
                    'constituent_counters': constituent_counters,
                    'constituent_sum': constituent_sum
                })

        return ValidationResult(ValidationStatus.PASSED,
                                'COUNTER_SUM_INTEGRITY',
                                details={
                                    'total_counter': total_counter,
                                    'total_value': total_value,
                                    'constituent_counters': constituent_counters,
                                    'constituent_sum': constituent_sum
                                })

    def validate_counter_min_yield(self, counters: dict,
                                   params: dict) -> ValidationResult:
        """Checks that a specific counter is above a minimum yield.

    Args:
      counters: A dictionary containing the counter values.
      params: A dictionary containing the validation parameters, which must
        have 'counter_name' and 'min_yield' keys.

    Returns:
      A ValidationResult object.
    """
        if 'counter_name' not in params or 'min_yield' not in params:
            return ValidationResult(
                ValidationStatus.CONFIG_ERROR,
                'COUNTER_MIN_YIELD',
                message=
                "Configuration error: 'counter_name' and 'min_yield' must be specified."
            )

        counter_name = params['counter_name']
        min_yield = params['min_yield']
        value = counters.get(counter_name, 0)

        if value < min_yield:
            return ValidationResult(
                ValidationStatus.FAILED,
                'COUNTER_MIN_YIELD',
                message=
                f"Counter '{counter_name}' is {value}, which is below the minimum yield of {min_yield}.",
                details={
                    'counter_name': counter_name,
                    'actual_value': value,
                    'min_yield': min_yield
                })

        return ValidationResult(ValidationStatus.PASSED,
                                'COUNTER_MIN_YIELD',
                                details={
                                    'counter_name': counter_name,
                                    'actual_value': value,
                                    'min_yield': min_yield
                                })