agent/parse.py

"""Robust JSON parsing helpers for claim / audit outputs."""

from __future__ import annotations

import json
import re
from typing import Any


_FENCE_RE = re.compile(r"```(?:json)?\s*([\s\S]*?)```", re.IGNORECASE)


def extract_json_value(text: str) -> Any | None:
    """Parse JSON from raw model text; tolerate markdown fences."""
    raw = text.strip()
    if not raw:
        return None
    fence = _FENCE_RE.search(raw)
    if fence:
        raw = fence.group(1).strip()
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        pass
    start_candidates = [i for i in (raw.find("["), raw.find("{")) if i >= 0]
    if not start_candidates:
        return None
    start = min(start_candidates)
    for end in range(len(raw), start, -1):
        try:
            return json.loads(raw[start:end])
        except json.JSONDecodeError:
            continue
    return None


def as_list_of_dicts(value: Any) -> list[dict[str, Any]]:
    if isinstance(value, list):
        return [item for item in value if isinstance(item, dict)]
    if isinstance(value, dict):
        for key in ("blocks", "claims", "items"):
            inner = value.get(key)
            if isinstance(inner, list):
                return [item for item in inner if isinstance(item, dict)]
    return []