osrs-codex/tools/validate.py

#!/usr/bin/env python3
"""OSRS Codex validator — the gate for every codex/**.yaml entry.

Runs:
  1. JSON-Schema validation (schemas/entry.schema.json)
  2. Cross-field consistency checks the schema can't express:
     - state_machine `next`/transition targets must be declared states
     - `initial` must be a declared state
     - duplicate action ids / step names
     - status:verified entries with `verified:` older than --stale-days
       => WARNING (entry should be re-verified, not dropped)
     - `best_practice` lines whose source starts with "researched"
       => WARNING (citation TODO — file a follow-up issue)
     - GE-margin entries (tags include flipping/merchanting): the 2% GE
       convenience fee is rounded DOWN to whole coins per item, so
       net_per_cycle MUST equal income - floor(0.02*income) - cost.
       Fractional-tax arithmetic (0.98*sell - buy) is an error.
  3. Global id uniqueness across the whole codex tree.

Exit codes: 0 = clean (warnings ok), 1 = validation errors.

Usage: python3 tools/validate.py [--stale-days 30]
"""
import datetime
import json
import math
import re
import sys
from pathlib import Path

import jsonschema
import yaml

ROOT = Path(__file__).resolve().parent.parent
SCHEMA = json.loads((ROOT / "schemas" / "entry.schema.json").read_text())
CODEX = ROOT / "codex"
STALE_DAYS = 30


def load_entries():
    out = []
    for p in CODEX.rglob("*.yaml"):
        if p.name.startswith("_"):  # dataset files (e.g. _f2p-alch-catalog.yaml)
            continue
        out.append(p)
    return sorted(out)


def main():
    global STALE_DAYS
    if "--stale-days" in sys.argv:
        STALE_DAYS = int(sys.argv[sys.argv.index("--stale-days") + 1])

    errors, warnings = [], []
    entries = load_entries()
    if not entries:
        print("ERROR: no codex entries found under codex/")
        return 1

    ids = {}
    today = datetime.date.today()

    for path in entries:
        rel = str(path.relative_to(ROOT))
        try:
            data = yaml.safe_load(path.read_text())
        except yaml.YAMLError as e:
            errors.append(f"{rel}: YAML parse error: {e}")
            continue

        # 1. schema
        v = jsonschema.Draft7Validator(SCHEMA)
        for err in sorted(v.iter_errors(data), key=lambda e: list(e.path)):
            loc = "/".join(str(p) for p in err.path) or "(root)"
            errors.append(f"{rel} [{loc}]: {err.message}")
        if errors and errors[-1].startswith(rel):
            # schema already broke; cross-checks on a broken doc are noise
            continue

        eid = data.get("id", rel)
        if eid in ids:
            errors.append(f"{rel}: duplicate entry id '{eid}' (also in {ids[eid]})")
        ids[eid] = rel

        # 2. state machine consistency
        sm = data.get("state_machine", {})
        states = sm.get("states", {})
        declared = set(states)
        terminals = set(sm.get("terminal", []))
        for sname, sdef in states.items():
            if sname not in terminals and not sdef.get("next"):
                errors.append(f"{rel}: non-terminal state '{sname}' has no 'next'")
            for nxt in sdef.get("next", []):
                if nxt not in declared:
                    errors.append(f"{rel}: state '{sname}' next->'{nxt}' not declared")
        if sm.get("initial") not in declared:
            errors.append(f"{rel}: initial state '{sm.get('initial')}' not declared")
        for t in sm.get("transitions", []):
            if t["from"] not in declared:
                errors.append(f"{rel}: transition from '{t['from']}' not declared")
            if t["to"] not in declared:
                errors.append(f"{rel}: transition to '{t['to']}' not declared")
        for term in sm.get("terminal", []):
            if term not in declared:
                errors.append(f"{rel}: terminal '{term}' not declared")

        # action id/step uniqueness
        a_ids = [a["id"] for a in data["actions"]]
        a_steps = [a["step"] for a in data["actions"]]
        if len(a_ids) != len(set(a_ids)):
            errors.append(f"{rel}: duplicate action ids")
        if len(a_steps) != len(set(a_steps)):
            errors.append(f"{rel}: duplicate action step names")

        # GE-margin floor-tax arithmetic. The wiki: "The 2% tax rounds down
        # to the nearest whole number, so items sold for below 50 coins have
        # a tax of less than one coin" — per-item net MUST use the floored
        # tax, never 0.98*sell - buy. Scoped to GE-margin tags so alchemy
        # (income is alch value, not a GE sale) is unaffected.
        tags = {str(t).lower() for t in data.get("tags", [])}
        eco = data.get("economy") or {}
        if tags & {"flipping", "merchanting", "ge"} and all(
            k in eco for k in ("income_per_cycle", "cost_per_cycle", "net_per_cycle")
        ):
            sell, buy = eco["income_per_cycle"], eco["cost_per_cycle"]
            tax = math.floor(sell * 0.02)
            expected = sell - tax - buy
            if eco["net_per_cycle"] != expected:
                errors.append(
                    f"{rel}: net_per_cycle {eco['net_per_cycle']} != floored-tax model "
                    f"{expected} (sell {sell} - floor({sell}*0.02)={tax} - buy {buy}); "
                    f"GE tax rounds down per item — fractional 0.98*sell - buy is invalid")

        # 3. staleness / research TODOs
        if data.get("status") == "verified":
            try:
                vdate = datetime.date.fromisoformat(data["verified"])
                age = (today - vdate).days
                if age > STALE_DAYS:
                    warnings.append(
                        f"{rel}: status=verified but facts are {age}d old "
                        f"(> {STALE_DAYS}d) — re-verify or drop to status:stale")
            except ValueError:
                pass
        for line in data.get("best_practice", {}).get("equipment", []) + \
                    data.get("best_practice", {}).get("techniques", []):
            if str(line.get("source", "")).startswith("researched"):
                warnings.append(
                    f"{rel}: best_practice '{line.get('item') or line.get('name')}' "
                    f"carries a 'researched' source — citation TODO, file follow-up")

    # report
    for w in warnings:
        print(f"WARN  {w}")
    for e in errors:
        print(f"ERROR {e}")
    print(f"RESULT: {len(entries)} entries, {len(errors)} errors, {len(warnings)} warnings")
    return 1 if errors else 0


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