Jonathan KirkwoodandClaude Fable 5
b1d7aed9f4
- Deterministic scoring.py (quant 60 / qual 40 / flags -15, profitability heaviest) - Per-company JSON ledger with forecast-target chaining deck N-1 -> N - Single-shot sandbox agent with guided-JSON fallback ladder (no tool loop) - Portfolio dashboard with sparklines, KPI hit rates, BDEF category bars - 48 unit tests green; endpoints smoke-tested; npm check+build green Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
"""JSON parsing + schema validation for extractor/grader outputs.
|
|
|
|
Local models occasionally wrap their JSON in prose or fences; parse_json_text
|
|
salvages the first brace-balanced top-level object before we give up. Schemas
|
|
live in orchestrator/schemas/ and are the single contract between the sandbox
|
|
agents and the deterministic scorer.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
|
|
SCHEMAS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "schemas")
|
|
_cache: dict[str, dict] = {}
|
|
|
|
|
|
def load_schema(name: str) -> dict:
|
|
"""Load "extraction" or "grades" schema (cached)."""
|
|
if name not in _cache:
|
|
path = os.path.join(SCHEMAS_DIR, f"{name}.schema.json")
|
|
with open(path, encoding="utf-8") as f:
|
|
_cache[name] = json.load(f)
|
|
return _cache[name]
|
|
|
|
|
|
def parse_json_text(text: str) -> dict | None:
|
|
"""Parse `text` as a JSON object; salvage the first balanced {...} block."""
|
|
if not text:
|
|
return None
|
|
try:
|
|
obj = json.loads(text)
|
|
return obj if isinstance(obj, dict) else None
|
|
except Exception:
|
|
pass
|
|
start = text.find("{")
|
|
while start != -1:
|
|
depth = 0
|
|
in_str = False
|
|
esc = False
|
|
for i in range(start, len(text)):
|
|
c = text[i]
|
|
if esc:
|
|
esc = False
|
|
elif in_str:
|
|
if c == "\\":
|
|
esc = True
|
|
elif c == '"':
|
|
in_str = False
|
|
elif c == '"':
|
|
in_str = True
|
|
elif c == "{":
|
|
depth += 1
|
|
elif c == "}":
|
|
depth -= 1
|
|
if depth == 0:
|
|
try:
|
|
obj = json.loads(text[start:i + 1])
|
|
if isinstance(obj, dict):
|
|
return obj
|
|
except Exception:
|
|
pass
|
|
break
|
|
start = text.find("{", start + 1)
|
|
return None
|
|
|
|
|
|
def validate_obj(obj, schema_name: str) -> str | None:
|
|
"""Validate against the named schema; error message or None if valid."""
|
|
import jsonschema
|
|
|
|
try:
|
|
jsonschema.validate(obj, load_schema(schema_name))
|
|
return None
|
|
except jsonschema.ValidationError as e:
|
|
path = ".".join(str(p) for p in e.absolute_path) or "(root)"
|
|
return f"{path}: {e.message}"[:500]
|
|
|
|
|
|
def validate_file(path: str, schema_name: str) -> tuple[dict | None, str | None]:
|
|
"""Read + parse (with salvage) + validate a JSON file -> (obj, error)."""
|
|
try:
|
|
with open(path, encoding="utf-8", errors="replace") as f:
|
|
text = f.read()
|
|
except Exception as e:
|
|
return None, f"read failed: {e}"
|
|
obj = parse_json_text(text)
|
|
if obj is None:
|
|
return None, "no parseable JSON object found"
|
|
return obj, validate_obj(obj, schema_name)
|