validate.py

"""
AIO 20002 Log v1.1 validator (formerly PRISM).

Checks that an AIO 20002 log code conforms to the v1.1 specification.
v1.1 grammar: '-' allowed in scope/reversibility/time; V/E/S layers optional;
single-code layers (no '<') allowed. Legacy <prism_log> tags accepted.

Usage:
    python validate.py                           # runs built-in test suite
    python validate.py "C:MED/IXi | V:Bec<Sda | E:Exp<Gui | S:Usr<Pro"
    python validate.py --file responses.txt      # validates all logs in a file
"""
import re
import sys
from typing import Optional

# ============================================================
# Vocabulary
# ============================================================

# 22 domains: CAP (Comparative Agendas Project) 21 major topics + GEN.
# Legacy v1.0 2-letter codes (MD ED LW DF FN TC GN) parse but are not valid
# v1.1 vocabulary — map at ingestion: MD>MED ED>EDU LW>LAW DF>DEF FN>COM TC>TEC GN>GEN.
DOMAINS = {"MAC", "CIV", "MED", "AGR", "LAB", "EDU", "ENV", "ENE", "IMM", "TRA",
           "LAW", "WEL", "HOU", "COM", "DEF", "TEC", "TRD", "INT", "GOV", "LND",
           "CUL", "GEN"}
SCOPES = set("IGCPS")
REVERSIBILITY = set("RPX")
TIME = set("isl")

# 19-value refined vocabulary (Schwartz et al., 2012) — the single value profile
# since v1.1. Legacy v1.0 10-value codes (Pow, Sel, Uni, Ben, Con, Sec) are NOT
# accepted here; validate legacy logs with a v1.0 validator.
SCHWARTZ = {"Sdt", "Sda", "Sti", "Hed", "Ach", "Pod", "Por", "Fac", "Sep", "Ses",
            "Tra", "Cor", "Coi", "Hum", "Bed", "Bec", "Unc", "Unn", "Unt"}
EVIDENCE = {"Rev", "Dat", "Cas", "Gui", "Exp", "Log", "Tri", "Pop", "Emo", "Ane"}
SOURCE = {"Pee", "Gov", "Pro", "Ind", "New", "Sta", "Tes", "Usr", "Alt", "Ano"}

# ============================================================
# Regex
# ============================================================

AIO20002_LOG_PATTERN = re.compile(
    r"C:(?P<domain>\w{2,3})/(?P<scope>[A-Z-])(?P<rev>[A-Z-])(?P<time>[a-z-])"
    r"(?:\s*\|\s*V:(?:(?P<v_lo>\w{3})<)?(?P<v_hi>\w{3}))?"
    r"(?:\s*\|\s*E:(?:(?P<e_lo>\w{3})<)?(?P<e_hi>\w{3}))?"
    r"(?:\s*\|\s*S:(?:(?P<s_lo>\w{3})<)?(?P<s_hi>\w{3}))?\s*$"
)

TAG_PATTERN = re.compile(r"<(?:aio20002|prism)_log>\s*(.*?)\s*</(?:aio20002|prism)_log>", re.DOTALL)  # legacy tag accepted


# ============================================================
# Validation
# ============================================================

def validate(code: str) -> list[str]:
    """Return a list of validation errors. Empty list = valid."""
    errors = []
    code = code.strip()

    m = AIO20002_LOG_PATTERN.match(code)
    if not m:
        return [f"Code does not match AIO 20002 v1.1 format: {code!r}"]

    g = m.groupdict()

    if g["domain"] not in DOMAINS:
        errors.append(f"Unknown domain: {g['domain']!r}. Valid: {sorted(DOMAINS)}")
    if g["scope"] not in SCOPES and g["scope"] != "-":
        errors.append(f"Unknown scope: {g['scope']!r}. Valid: {sorted(SCOPES)} or '-'")
    if g["rev"] not in REVERSIBILITY and g["rev"] != "-":
        errors.append(f"Unknown reversibility: {g['rev']!r}. Valid: {sorted(REVERSIBILITY)} or '-'")
    if g["time"] not in TIME and g["time"] != "-":
        errors.append(f"Unknown time: {g['time']!r}. Valid: {sorted(TIME)} or '-'")

    for field, vocab in [("v_lo", SCHWARTZ), ("v_hi", SCHWARTZ),
                         ("e_lo", EVIDENCE), ("e_hi", EVIDENCE),
                         ("s_lo", SOURCE), ("s_hi", SOURCE)]:
        if g[field] is not None and g[field] not in vocab:
            errors.append(f"Unknown code for {field}: {g[field]!r}")

    if g["v_hi"] is None and g["e_hi"] is None and g["s_hi"] is None:
        errors.append("No V/E/S layer present — at least one layer is required (v1.1)")

    if not errors:
        if g["v_lo"] is not None and g["v_lo"] == g["v_hi"]:
            errors.append(f"V layer: same code on both sides ({g['v_lo']})")
        if g["e_lo"] is not None and g["e_lo"] == g["e_hi"]:
            errors.append(f"E layer: same code on both sides ({g['e_lo']})")
        if g["s_lo"] is not None and g["s_lo"] == g["s_hi"]:
            errors.append(f"S layer: same code on both sides ({g['s_lo']})")

    return errors


def extract_logs(text: str) -> list[str]:
    """Extract AIO 20002 code strings from input text in any of three modes.

    Mode A: inline <aio20002_log>...</aio20002_log> tags
    Mode B: JSON with 'aio20002_log' key containing {'code': '...'} or bare 'aio20002_log': '...'
    Mode C: tool call JSON with 'input' or 'arguments' containing {'code': '...'}

    Returns list of code strings found. Empty if none.
    """
    import json as _json

    codes = []

    # Mode A: inline tags
    codes.extend(m.group(1).strip() for m in TAG_PATTERN.finditer(text))

    # Mode B / Mode C: scan for JSON objects that contain a AIO 20002 code
    # Try the whole input as one JSON object first
    stripped = text.strip()
    if stripped.startswith("{") or stripped.startswith("["):
        try:
            obj = _json.loads(stripped)
            codes.extend(_walk_for_codes(obj))
        except _json.JSONDecodeError:
            pass

    # Also scan for embedded JSON-like substrings (conservative: only if no Mode A matches)
    # This catches JSON snippets that appear alongside other text.
    if not codes:
        # Find things that look like JSON objects containing "code"
        for match in re.finditer(r'\{[^{}]*"code"\s*:\s*"([^"]+)"[^{}]*\}', text):
            candidate = match.group(1).strip()
            if AIO20002_LOG_PATTERN.match(candidate):
                codes.append(candidate)

    # Deduplicate while preserving order
    seen = set()
    unique = []
    for c in codes:
        if c not in seen:
            seen.add(c)
            unique.append(c)
    return unique


def _walk_for_codes(obj):
    """Recursively search a JSON-parsed object for AIO 20002 code strings."""
    results = []
    if isinstance(obj, dict):
        # Direct 'code' key with a valid-looking string
        if "code" in obj and isinstance(obj["code"], str):
            if AIO20002_LOG_PATTERN.match(obj["code"].strip()):
                results.append(obj["code"].strip())
        # 'aio20002_log' as a string (legacy/flat form)
        if "aio20002_log" in obj and isinstance(obj["aio20002_log"], str):
            if AIO20002_LOG_PATTERN.match(obj["aio20002_log"].strip()):
                results.append(obj["aio20002_log"].strip())
        # Recurse into all values
        for v in obj.values():
            results.extend(_walk_for_codes(v))
    elif isinstance(obj, list):
        for item in obj:
            results.extend(_walk_for_codes(item))
    return results


def parse(code: str) -> Optional[dict]:
    """Parse a AIO 20002 log code into a dict. Returns None if invalid."""
    m = AIO20002_LOG_PATTERN.match(code.strip())
    return m.groupdict() if m else None


# ============================================================
# Built-in tests
# ============================================================

VALID_SAMPLES = [
    "C:MED/IXi | V:Bec<Sda | E:Exp<Gui | S:Usr<Pro",
    "C:EDU/IPl | V:Sep<Sda | E:Pop<Exp | S:New<Pro",
    "C:DEF/SXi | V:Ses<Unc | E:Exp<Gui | S:Ind<Gov",
    "C:COM/IRs | V:Ach<Sep | E:Pop<Gui | S:Alt<Pro",
    "C:LAW/CXs | V:Sda<Unc | E:Ane<Gui | S:Usr<Pro",
    "C:TEC/GRl | V:Sti<Ses | E:Ane<Cas | S:Alt<Pro",
    # v1.1 partial output
    "C:MED/I-s | V:Hed<Sep | E:Gui",
    "C:GEN/I-l | V:Ach<Bec | E:Log",
    "C:LAW/CXs | V:Sda<Unc",
    "C:GEN/--l | V:Sep<Unc",
    "C:IMM/PPl | V:Unt<Ses | E:Cas<Gui | S:New<Gov",
    "C:ENV/SXl | V:Por<Unn | E:Rev | S:Pee",
]

INVALID_SAMPLES = [
    ("C:ZZ/IXi | V:Bec<Sda | E:Exp<Gui | S:Usr<Pro", "domain"),
    ("C:MED/QXi | V:Bec<Sda | E:Exp<Gui | S:Usr<Pro", "scope"),
    ("C:MED/IZi | V:Bec<Sda | E:Exp<Gui | S:Usr<Pro", "reversibility"),
    ("C:MED/IXz | V:Bec<Sda | E:Exp<Gui | S:Usr<Pro", "time"),
    ("C:MED/IXi | V:Xxx<Sda | E:Exp<Gui | S:Usr<Pro", "v_lo"),
    ("C:MED/IXi | V:Bec<Sda | E:Xxx<Gui | S:Usr<Pro", "e_lo"),
    ("C:MED/IXi | V:Bec<Sda | E:Exp<Gui | S:Xxx<Pro", "s_lo"),
    ("C:MED/IXi | V:Bec<Bec | E:Exp<Gui | S:Usr<Pro", "same code"),
    ("malformed string", "match"),
    ("C:MED/I-i", "layer"),  # v1.1: context alone is not enough
]


def run_tests():
    print("Running AIO 20002 log validator self-test...\n")

    print(f"Valid samples ({len(VALID_SAMPLES)}):")
    v_passed = 0
    for code in VALID_SAMPLES:
        errors = validate(code)
        status = "✅" if not errors else "❌"
        print(f"  {status} {code}")
        if errors:
            for e in errors:
                print(f"     - {e}")
        else:
            v_passed += 1

    print(f"\nInvalid samples ({len(INVALID_SAMPLES)}):")
    i_passed = 0
    for code, expected_keyword in INVALID_SAMPLES:
        errors = validate(code)
        caught = bool(errors) and any(expected_keyword.lower() in e.lower() for e in errors)
        status = "✅" if caught else "❌"
        print(f"  {status} [expects '{expected_keyword}'] {code[:55]}...")
        if errors:
            for e in errors[:1]:
                print(f"     - {e}")
        if caught:
            i_passed += 1

    # Multi-mode extraction test
    print(f"\nMulti-mode extraction tests:")
    mode_a = '<aio20002_log>\nC:MED/IXi | V:Bec<Sda | E:Exp<Gui | S:Usr<Pro\n</aio20002_log>'
    legacy_a = '<prism_log>\nC:MED/IXi | V:Bec<Sda | E:Exp<Gui | S:Usr<Pro\n</prism_log>'
    mode_b = '{"response": "text", "aio20002_log": {"code": "C:MED/IXi | V:Bec<Sda | E:Exp<Gui | S:Usr<Pro"}}'
    mode_c = '{"tool_call": {"name": "record_aio20002_log", "input": {"code": "C:MED/IXi | V:Bec<Sda | E:Exp<Gui | S:Usr<Pro"}}}'
    expected = "C:MED/IXi | V:Bec<Sda | E:Exp<Gui | S:Usr<Pro"
    m_passed = 0
    for mode_name, raw in [("Mode A (inline)", mode_a), ("Mode A (legacy prism tag)", legacy_a), ("Mode B (JSON)", mode_b), ("Mode C (tool)", mode_c)]:
        extracted = extract_logs(raw)
        ok = len(extracted) == 1 and extracted[0] == expected
        status = "✅" if ok else "❌"
        print(f"  {status} {mode_name}: extracted {extracted}")
        if ok:
            m_passed += 1

    print(f"\nResult: {v_passed}/{len(VALID_SAMPLES)} valid samples accepted, "
          f"{i_passed}/{len(INVALID_SAMPLES)} invalid samples caught, "
          f"{m_passed}/4 modes extracted correctly")

    if v_passed != len(VALID_SAMPLES) or i_passed != len(INVALID_SAMPLES) or m_passed != 4:
        sys.exit(1)


# ============================================================
# CLI
# ============================================================

if __name__ == "__main__":
    args = sys.argv[1:]

    if not args:
        run_tests()
        sys.exit(0)

    if args[0] == "--file":
        if len(args) < 2:
            print("Usage: validate.py --file <path>")
            sys.exit(1)
        with open(args[1]) as f:
            text = f.read()
        logs = extract_logs(text)
        print(f"Found {len(logs)} AIO 20002 logs in {args[1]}")
        total_errors = 0
        for i, log in enumerate(logs, 1):
            errors = validate(log)
            if errors:
                total_errors += 1
                print(f"\n[{i}] ❌ {log}")
                for e in errors:
                    print(f"     {e}")
            else:
                print(f"[{i}] ✅ {log}")
        print(f"\nSummary: {len(logs) - total_errors}/{len(logs)} logs valid")
        sys.exit(1 if total_errors else 0)

    # Single code on CLI
    code = args[0]
    errors = validate(code)
    if not errors:
        parsed = parse(code)
        print("✅ Valid AIO 20002 log (v1.1)")
        print()
        for k, v in parsed.items():
            print(f"  {k}: {v}")
    else:
        print("❌ Invalid AIO 20002 log")
        for e in errors:
            print(f"  {e}")
        sys.exit(1)