```python
#!/usr/bin/env python3
"""Adversarial ML Input Guard and Monitoring Script.

Purpose:
    Provide a practical, vendor-neutral guardrail layer for machine learning
    inference pipelines. The script performs lightweight validation,
    normalization, heuristic anomaly checks, confidence gating, and basic
    monitoring output to help reduce attacker leverage before model outputs
    are used downstream.

What this script is / is not:
    - It is a defensive utility, not a full adversarial defense solution.
    - It is designed to sit in front of a model or to be used in batch review.
    - It does not call external services, require secrets, or hard-code endpoints.
    - It does not mutate production systems or make irreversible decisions.

Typical uses:
    - Validate incoming records before inference.
    - Flag suspicious requests for review or fallback handling.
    - Apply conservative confidence thresholds.
    - Log summary statistics for drift / probing detection.

Example:
    python adversarial_ml_guard.py --input sample.json --schema schema.json --min-confidence 0.80

Input formats:
    - A JSON file containing either a single record (object) or a list of records.
    - Optional schema file describing expected keys and basic constraints.

Schema format example:
    {
      "required": ["age", "country", "score"],
      "types": {
        "age": "number",
        "country": "string",
        "score": "number"
      },
      "ranges": {
        "age": [0, 120],
        "score": [0, 1]
      },
      "max_lengths": {
        "country": 2
      }
    }

Output:
    Prints a per-record assessment and a summary suitable for logs or manual review.
"""

from __future__ import annotations

import argparse
import json
import math
import re
import sys
from collections import Counter
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple


SUSPICIOUS_TEXT_PATTERNS = [
    r"\x00",
    r"[\u202e\u202d]",  # bidi control characters
    r"[\r\n]{3,}",
    r"[\t]{4,}",
    r"(?:%[0-9A-Fa-f]{2}){4,}",
    r"(?:\.{2,}/){2,}",
]


@dataclass
class RecordAssessment:
    index: int
    valid: bool
    accepted: bool
    reasons: List[str] = field(default_factory=list)
    warnings: List[str] = field(default_factory=list)
    score: Optional[float] = None


@dataclass
class GuardConfig:
    schema: Dict[str, Any] = field(default_factory=dict)
    min_confidence: float = 0.80
    max_text_length: int = 4096
    max_keys: int = 100
    near_duplicate_threshold: float = 0.92
    enable_normalization: bool = True


def load_json_file(path: Path) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        raise SystemExit(f"Input file not found: {path}")
    except json.JSONDecodeError as exc:
        raise SystemExit(f"Invalid JSON in {path}: {exc}")


def load_records(path: Path) -> List[Dict[str, Any]]:
    data = load_json_file(path)
    if isinstance(data, dict):
        return [data]
    if isinstance(data, list):
        if not all(isinstance(item, dict) for item in data):
            raise SystemExit("Input JSON list must contain only objects/records.")
        return data
    raise SystemExit("Input JSON must be an object or a list of objects.")


def load_schema(path: Optional[Path]) -> Dict[str, Any]:
    if path is None:
        return {}
    schema = load_json_file(path)
    if not isinstance(schema, dict):
        raise SystemExit("Schema JSON must be an object.")
    return schema


def normalize_value(value: Any) -> Any:
    if isinstance(value, str):
        value = value.strip()
        value = re.sub(r"\s+", " ", value)
        return value
    return value


def normalize_record(record: Dict[str, Any]) -> Dict[str, Any]:
    return {str(k): normalize_value(v) for k, v in record.items()}


def is_number(value: Any) -> bool:
    return isinstance(value, (int, float)) and not isinstance(value, bool) and not math.isnan(float(value))


def infer_type(value: Any) -> str:
    if value is None:
        return "null"
    if isinstance(value, bool):
        return "boolean"
    if is_number(value):
        return "number"
    if isinstance(value, str):
        return "string"
    if isinstance(value, list):
        return "array"
    if isinstance(value, dict):
        return "object"
    return type(value).__name__


def check_text_suspicion(text: str) -> List[str]:
    findings: List[str] = []
    for pattern in SUSPICIOUS_TEXT_PATTERNS:
        if re.search(pattern, text):
            findings.append(f"suspicious text pattern matched: {pattern}")
    if len(text) > 0:
        alpha = sum(ch.isalpha() for ch in text)
        nonspace = sum(not ch.isspace() for ch in text)
        if nonspace > 0:
            ratio = alpha / nonspace
            if ratio < 0.15 and len(text) > 24:
                findings.append("low alphabetic-to-nonspace ratio")
    return findings


def schema_value_type_check(value: Any, expected: str) -> bool:
    if expected == "string":
        return isinstance(value, str)
    if expected == "number":
        return is_number(value)
    if expected == "boolean":
        return isinstance(value, bool)
    if expected == "array":
        return isinstance(value, list)
    if expected == "object":
        return isinstance(value, dict)
    if expected == "null":
        return value is None
    return True


def validate_record(record: Dict[str, Any], cfg: GuardConfig) -> Tuple[bool, List[str], List[str]]:
    reasons: List[str] = []
    warnings: List[str] = []

    if not isinstance(record, dict):
        return False, ["record is not an object"], []

    if len(record) > cfg.max_keys:
        reasons.append(f"too many keys: {len(record)} > {cfg.max_keys}")

    required = cfg.schema.get("required", [])
    if isinstance(required, list):
        for key in required:
            if key not in record:
                reasons.append(f"missing required key: {key}")
    else:
        warnings.append("schema.required should be a list")

    types = cfg.schema.get("types", {})
    if isinstance(types, dict):
        for key, expected in types.items():
            if key in record and isinstance(expected, str):
                if not schema_value_type_check(record[key], expected):
                    reasons.append(f"type mismatch for {key}: expected {expected}, got {infer_type(record[key])}")
    else:
        warnings.append("schema.types should be an object")

    ranges = cfg.schema.get("ranges", {})
    if isinstance(ranges, dict):
        for key, bounds in ranges.items():
            if key in record and isinstance(bounds, list) and len(bounds) == 2 and is_number(record[key]):
                low, high = bounds
                if is_number(low) and is_number(high) and not (low <= float(record[key]) <= high):
                    reasons.append(f"range violation for {key}: {record[key]} not in [{low}, {high}]")
    else:
        warnings.append("schema.ranges should be an object")

    max_lengths = cfg.schema.get("max_lengths", {})
    if isinstance(max_lengths, dict):
        for key, limit in max_lengths.items():
            if key in record and isinstance(record[key], str) and isinstance(limit, int):
                if len(record[key]) > limit:
                    reasons.append(f"length violation for {key}: {len(record[key])} > {limit}")
    else:
        warnings.append("schema.max_lengths should be an object")

    for key, value in record.items():
        if isinstance(value, str):
            if len(value) > cfg.max_text_length:
                reasons.append(f"text too long for {key}: {len(value)} > {cfg.max_text_length}")
            warnings.extend(check_text_suspicion(value))
        elif isinstance(value, list) and len(value) > cfg.max_keys:
            warnings.append(f"large array in field {key}: {len(value)} items")

    valid = not reasons
    return valid, reasons, warnings


def estimate_risk_score(record: Dict[str, Any], warnings: List[str]) -> float:
    score = 1.0
    penalty = 0.0
    penalty += min(len(record) / 200.0, 0.10)
    penalty += min(len(warnings) * 0.04, 0.35)

    text_fields = [v for v in record.values() if isinstance(v, str)]
    if text_fields:
        avg_len = sum(len(v) for v in text_fields) / len(text_fields)
        if avg_len > 256:
            penalty += 0.05

    score = max(0.0, min(1.0, score - penalty))
    return score


def near_duplicate_signature(record: Dict[str, Any]) -> str:
    normalized = json.dumps(record, sort_keys=True, ensure_ascii=False)
    normalized = re.sub(r"\s+", "", normalized)
    return normalized


def assess_records(records: List[Dict[str, Any]], cfg: GuardConfig) -> Tuple[List[RecordAssessment], Dict[str, Any]]:
    assessments: List[RecordAssessment] = []
    signatures: Counter[str] = Counter()

    for idx, raw in enumerate(records):
        record = normalize_record(raw) if cfg.enable_normalization else raw
        valid, reasons, warnings = validate_record(record, cfg)
        signature = near_duplicate_signature(record)
        signatures[signature] += 1
        if signatures[signature] > 1:
            warnings.append("duplicate or near-duplicate request observed")

        risk_score = estimate_risk_score(record, warnings)
        accepted = valid and risk_score >= cfg.min_confidence
        if not accepted and valid and risk_score < cfg.min_confidence:
            reasons.append(f"confidence below threshold: {risk_score:.2f} < {cfg.min_confidence:.2f}")

        assessments.append(
            RecordAssessment(
                index=idx,
                valid=valid,
                accepted=accepted,
                reasons=reasons,
                warnings=warnings,
                score=risk_score,
            )
        )

    summary = {
        "total": len(assessments),
        "accepted": sum(1 for a in assessments if a.accepted),
        "rejected": sum(1 for a in assessments if not a.accepted),
        "valid": sum(1 for a in assessments if a.valid),
        "invalid": sum(1 for a in assessments if not a.valid),
        "avg_score": round(sum(a.score or 0.0 for a in assessments) / max(len(assessments), 1), 4),
    }
    return assessments, summary


def print_assessment(assessment: RecordAssessment) -> None:
    status = "ACCEPT" if assessment.accepted else "REVIEW/REJECT"
    print(f"Record {assessment.index}: {status}")
    print(f"  valid: {assessment.valid}")
    print(f"  score: {assessment.score:.2f}" if assessment.score is not None else "  score: n/a")
    if assessment.reasons:
        print("  reasons:")
        for reason in assessment.reasons:
            print(f"    - {reason}")
    if assessment.warnings:
        print("  warnings:")
        for warning in sorted(set(assessment.warnings)):
            print(f"    - {warning}")


def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Validate and score ML inference records for adversarial-risk signals.")
    parser.add_argument("--input", required=True, type=Path, help="Path to a JSON file containing one record or a list of records.")
    parser.add_argument("--schema", type=Path, default=None, help="Optional JSON schema file with required fields, types, ranges, and max lengths.")
    parser.add_argument("--min-confidence", type=float, default=0.80, help="Minimum acceptable confidence/risk score before routing to review.")
    parser.add_argument("--max-text-length", type=int, default=4096, help="Maximum permitted string length before rejection.")
    parser.add_argument("--max-keys", type=int, default=100, help="Maximum number of keys per record before rejection.")
    parser.add_argument("--no-normalization", action="store_true", help="Disable basic whitespace normalization of string fields.")
    return parser.parse_args(argv)


def main(argv: Optional[List[str]] = None) -> int:
    args = parse_args(argv)
    if not 0.0 <= args.min_confidence <= 1.0:
        raise SystemExit("--min-confidence must be between 0.0 and 1.0")
    if args.max_text_length <= 0 or args.max_keys <= 0:
        raise SystemExit("--max-text-length and --max-keys must be positive integers")

    schema = load_schema(args.schema)
    cfg = GuardConfig(
        schema=schema,
        min_confidence=args.min_confidence,
        max_text_length=args.max_text_length,
        max_keys=args.max_keys,
        enable_normalization=not args.no_normalization,
    )

    records = load_records(args.input)
    assessments, summary = assess_records(records, cfg)

    for assessment in assessments:
        print_assessment(assessment)
        print()

    print("Summary:")
    for key, value in summary.items():
        print(f"  {key}: {value}")

    # Exit code convention:
    #   0 = no critical validation failures
    #   1 = one or more records were rejected or routed to review
    return 0 if summary["rejected"] == 0 else 1


if __name__ == "__main__":
    sys.exit(main())
```