Jesse_Chen
872 lines
42 KiB
Python
872 lines
42 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""ReportDocument v1/v2 contract validator (stdlib only).
|
|
|
|
ReportDocument v2 is the canonical current contract. Stored v1 documents remain
|
|
readable while producers and renderers migrate. The v2 JSON Schema, the Zod
|
|
contract, and this validator share structural and semantic rules. Evidence
|
|
hashes are always recomputed here; they are never trusted as model self-report.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
|
|
|
|
V1_SCHEMA_VERSION = "report_document.v1"
|
|
V2_SCHEMA_VERSION = "report_document.v2"
|
|
CURRENT_SCHEMA_VERSION = V2_SCHEMA_VERSION
|
|
# Deprecated producer compatibility: existing Python callers and tests used
|
|
# SCHEMA_VERSION for v1 document assembly. New code must use CURRENT_SCHEMA_VERSION.
|
|
SCHEMA_VERSION = V1_SCHEMA_VERSION
|
|
REPORT_CONTRACT_VERSION = "1"
|
|
CURRENT_REPORT_CONTRACT_VERSION = "2"
|
|
MAX_SERIALIZED_BYTES = 1_572_864
|
|
|
|
CLAIM_STATUSES = (
|
|
"multi_system_consensus",
|
|
"single_system_inference",
|
|
"parameter_sensitive",
|
|
"unclosed_divisional_chart",
|
|
"user_history_verification_required",
|
|
"blocked",
|
|
)
|
|
REPORT_TYPES = ("personal_full", "personal_thematic")
|
|
PRESENTATION_MODES = ("default", "research")
|
|
REPORT_DEPTHS = ("concise", "standard", "deep", "research")
|
|
BIRTH_TIME_STATUSES = ("reported", "candidate", "accepted", "confirmed")
|
|
TECHNIQUE_STATUSES = ("verified", "partial", "blocked")
|
|
CONFLICT_STATUSES = ("unresolved", "partial", "resolved")
|
|
REPORT_ACTION_PRIORITIES = ("now", "next", "watch")
|
|
V1_CHART_IDS = ("D1", "D9", "D10")
|
|
CHART_IDS = ("D1", "D2", "D9", "D10", "D11", "D24")
|
|
HOUSE_NUMBERS = tuple(range(1, 13))
|
|
|
|
FAILURE_CODES = (
|
|
"profile_incomplete",
|
|
"birth_time_not_usable",
|
|
"report_generation_in_progress",
|
|
"report_rate_limited",
|
|
"calculation_unavailable",
|
|
"model_unavailable",
|
|
"report_schema_invalid",
|
|
"report_guard_rejected",
|
|
"report_not_found",
|
|
)
|
|
|
|
UUID_PATTERN = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")
|
|
SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
|
|
SHA1_PATTERN = re.compile(r"^[0-9a-f]{40}$")
|
|
SKILL_NAME_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9._-]*[a-z0-9])?$")
|
|
SKILL_VERSION_PATTERN = re.compile(
|
|
r"^(?:0|[1-9]\d*)\.(?:0|[1-9]\d*)\.(?:0|[1-9]\d*)"
|
|
r"(?:-[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?"
|
|
r"(?:\+[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?$"
|
|
)
|
|
ISO8601_PATTERN = re.compile(
|
|
r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{1,9})?(Z|[+-]\d{2}:\d{2})$"
|
|
)
|
|
EVIDENCE_ID_PATTERN = re.compile(r"^ev-[a-z0-9_-]{1,63}$")
|
|
V1_SECTION_ID_PATTERN = re.compile(r"^[a-z][a-z0-9_-]{0,63}$")
|
|
SECTION_ID_PATTERN = re.compile(r"^[a-z][a-z0-9_-]{0,95}$")
|
|
THEME_ID_PATTERN = re.compile(r"^[a-z][a-z0-9_.-]{0,95}$")
|
|
TECHNIQUE_ID_PATTERN = re.compile(r"^[a-z0-9_.-]{1,80}$")
|
|
|
|
# Keep names and regular-expression meaning aligned with the TS side.
|
|
FORBIDDEN_PATTERNS: Tuple[Tuple[str, str], ...] = (
|
|
("html_tag", r"<\s*/?\s*[a-z][^>]*>"),
|
|
("event_handler", r"\bon(?:load|error|click|mouseover|mouseout|submit|focus|blur|change|dblclick|keydown|keyup|pointerdown|pointerup)\s*="),
|
|
("style_attribute", r"\bstyle\s*="),
|
|
("css_at_rule", r"@(?:import|media|supports|font-face|keyframes)\b"),
|
|
("css_rule", r"(?:^|[}\s])(?:[.#]?[a-z][a-z0-9_-]*)(?:\s+[.#]?[a-z][a-z0-9_-]*)*\s*\{[^{}]*\}"),
|
|
("css_declaration", r"(?:^|[;{\s])(?:color|background(?:-color)?|font(?:-family|-size|-weight)?|display|position|margin|padding|width|height|grid|flex|border|transform|animation)\s*:\s*[^;\n{}]+[;}]?"),
|
|
("executable_url", r"\b(?:javascript|vbscript|data:text/html|data:text/javascript|file):"),
|
|
("processing_instruction", r"<\?"),
|
|
("template_literal", r"\$\{"),
|
|
("stack_trace", r"(?:Traceback \(most recent call last\)|node:internal/| at (?:Object|async|node)\.)"),
|
|
("dunder_path", r"__(?:dirname|filename)(?![A-Za-z0-9_])|__proto__"),
|
|
("process_env", r"\bprocess\.env\b"),
|
|
("unix_home_path", r"(?:^|[\\/:])(?:Users|home|opt|var|tmp|root|srv)[\\/]"),
|
|
("windows_drive_path", r"^[a-zA-Z]:[\\/]"),
|
|
("jwt_token", r"\beyJ[A-Za-z0-9_-]{20,}\b"),
|
|
("secret_marker", r"\b(?:SUPABASE_SERVICE_ROLE_KEY|AUTH_SECRET|BEGIN RSA PRIVATE KEY|BEGIN EC PRIVATE KEY|BEGIN OPENSSH PRIVATE KEY)\b"),
|
|
("tool_trace", r"\b(?:tool_call_id|tool_result|assistant_tool_calls|system_prompt)\b"),
|
|
("chain_of_thought", r"\bchain[\s_-]?of[\s_-]?thought\b"),
|
|
("medical_diagnosis", r"(?:你|命主)(?:已经|已|必将|一定会|确定)?(?:患有|罹患|得了)(?:癌症|糖尿病|抑郁症|双相情感障碍|心脏病|精神疾病)|(?:确诊为|诊断为)(?:癌症|糖尿病|抑郁症|双相情感障碍|心脏病|精神疾病)|\b(?:you|the native)\s+(?:definitely\s+)?(?:have|has|will develop)\s+(?:cancer|diabetes|depression|bipolar disorder|heart disease)\b|\bdiagnosed with\s+(?:cancer|diabetes|depression|bipolar disorder|heart disease)\b"),
|
|
("deterministic_financial_promise", r"(?:保证收益|保本保收益|稳赚不赔|稳赚|必赚|确定(?:盈利|获利|回报)|一定会(?:赚钱|盈利|获利))|\b(?:guaranteed|certain|risk-free)\s+(?:profit|return|gain)s?\b|\bwill definitely\s+(?:profit|earn|make money)\b"),
|
|
)
|
|
FORBIDDEN_COMPILED = tuple((name, re.compile(pattern, re.IGNORECASE)) for name, pattern in FORBIDDEN_PATTERNS)
|
|
|
|
DETERMINISTIC_PHRASES: Tuple[str, ...] = (
|
|
"必然", "必定", "一定会", "肯定会", "绝对会", "保证会", "无疑将", "百分之百", "确定无疑",
|
|
"guaranteed", "definitely will", "certainly will", "will certainly", "is certain to",
|
|
)
|
|
DETERMINISTIC_COMPILED = tuple(re.compile(re.escape(phrase), re.IGNORECASE) for phrase in DETERMINISTIC_PHRASES)
|
|
DATE_CLAIM_PATTERN = re.compile(
|
|
r"(?:^|[^0-9])(?:19|20)\d{2}(?:年(?:0?[1-9]|1[0-2])月(?:(?:0?[1-9]|[12]\d|3[01])日)?|[-/.](?:0?[1-9]|1[0-2])(?:[-/.](?:0?[1-9]|[12]\d|3[01]))?)(?:[^0-9]|$)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
TEXT_LIMITS: Dict[str, int] = {
|
|
"displayName": 120,
|
|
"birthPlaceLabel": 200,
|
|
"headline": 200,
|
|
"summary": 2000,
|
|
"priority": 200,
|
|
"chartTitle": 120,
|
|
"sign": 40,
|
|
"occupant": 40,
|
|
"planetName": 40,
|
|
"sectionTitle": 160,
|
|
"phaseLabel": 200,
|
|
"narrative": 4000,
|
|
"action": 400,
|
|
"actionNote": 1000,
|
|
"caveat": 400,
|
|
"keyFactor": 400,
|
|
"timingNote": 400,
|
|
"reason": 2000,
|
|
"missingEvidence": 400,
|
|
"conflictNote": 500,
|
|
"techniqueName": 160,
|
|
"notes": 500,
|
|
"conflictDescription": 1000,
|
|
"conflictImpact": 500,
|
|
"evidenceLabel": 160,
|
|
"evidenceValue": 500,
|
|
"evidenceSource": 200,
|
|
"blockedTechnique": 120,
|
|
"disclaimer": 2000,
|
|
}
|
|
|
|
REQUIRED_THEME_CHARTS: Dict[str, Tuple[str, ...]] = {
|
|
"career": ("D10",),
|
|
"wealth": ("D2", "D11"),
|
|
"marriage": ("D9",),
|
|
"education": ("D24",),
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class ValidationResult:
|
|
valid: bool
|
|
errors: List[str] = field(default_factory=list)
|
|
|
|
def add(self, path: str, message: str) -> None:
|
|
self.errors.append(f"{path}: {message}")
|
|
|
|
|
|
def _canonical_evidence(document: Dict[str, Any]) -> Dict[str, Any]:
|
|
appendix = document.get("evidenceAppendix")
|
|
if not isinstance(appendix, dict):
|
|
appendix = {}
|
|
|
|
def rows(key: str) -> List[Any]:
|
|
value = appendix.get(key)
|
|
return value if isinstance(value, list) else []
|
|
|
|
technique_rows: List[Dict[str, Any]] = []
|
|
for raw in rows("techniqueAudit"):
|
|
row = raw if isinstance(raw, dict) else {}
|
|
item = {
|
|
"id": row.get("id"),
|
|
"techniqueId": row.get("techniqueId"),
|
|
"techniqueName": row.get("techniqueName"),
|
|
"status": row.get("status"),
|
|
"used": row.get("used"),
|
|
}
|
|
if "notes" in row:
|
|
item["notes"] = row.get("notes")
|
|
technique_rows.append(item)
|
|
return {
|
|
"techniqueAudit": technique_rows,
|
|
"conflicts": [
|
|
{key: (raw if isinstance(raw, dict) else {}).get(key) for key in ("id", "description", "impact", "status")}
|
|
for raw in rows("conflicts")
|
|
],
|
|
"calculationEvidence": [
|
|
{key: (raw if isinstance(raw, dict) else {}).get(key) for key in ("id", "label", "value", "source")}
|
|
for raw in rows("calculationEvidence")
|
|
],
|
|
}
|
|
|
|
|
|
def compute_evidence_hash(document: Dict[str, Any]) -> str:
|
|
payload = json.dumps(_canonical_evidence(document), ensure_ascii=False, separators=(",", ":"))
|
|
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def serialized_bytes(document: Any) -> int:
|
|
return len(json.dumps(document, ensure_ascii=False, separators=(",", ":")).encode("utf-8"))
|
|
|
|
|
|
def _expect_object(result: ValidationResult, path: str, value: Any) -> Optional[Dict[str, Any]]:
|
|
if not isinstance(value, dict):
|
|
result.add(path, f"must be object, got {type(value).__name__}")
|
|
return None
|
|
return value
|
|
|
|
|
|
def _check_keys(
|
|
result: ValidationResult,
|
|
path: str,
|
|
value: Dict[str, Any],
|
|
required: Sequence[str],
|
|
allowed: Optional[Sequence[str]] = None,
|
|
) -> None:
|
|
required_set = set(required)
|
|
allowed_set = set(allowed or required)
|
|
missing = sorted(required_set - set(value))
|
|
extra = sorted(set(value) - allowed_set)
|
|
if missing:
|
|
result.add(path, f"missing required keys {missing}")
|
|
if extra:
|
|
result.add(path, f"unexpected keys {extra}")
|
|
|
|
|
|
def _check_text(
|
|
result: ValidationResult,
|
|
path: str,
|
|
value: Any,
|
|
max_length: int,
|
|
min_length: int = 1,
|
|
) -> Optional[str]:
|
|
if not isinstance(value, str):
|
|
result.add(path, "must be string")
|
|
return None
|
|
if len(value) < min_length or len(value) > max_length:
|
|
result.add(path, f"length must be {min_length}..{max_length}")
|
|
return value
|
|
|
|
|
|
def _check_pattern(result: ValidationResult, path: str, value: Any, pattern: re.Pattern[str], label: str) -> None:
|
|
if not isinstance(value, str) or pattern.fullmatch(value) is None:
|
|
result.add(path, f"invalid {label} {value!r}")
|
|
|
|
|
|
def _check_enum(result: ValidationResult, path: str, value: Any, allowed: Sequence[str]) -> None:
|
|
if value not in allowed:
|
|
result.add(path, f"must be one of {tuple(allowed)!r}, got {value!r}")
|
|
|
|
|
|
def _check_array(
|
|
result: ValidationResult,
|
|
path: str,
|
|
value: Any,
|
|
max_items: int,
|
|
min_items: int = 0,
|
|
) -> Optional[List[Any]]:
|
|
if not isinstance(value, list):
|
|
result.add(path, "must be array")
|
|
return None
|
|
if len(value) < min_items or len(value) > max_items:
|
|
result.add(path, f"item count must be {min_items}..{max_items}")
|
|
return value
|
|
|
|
|
|
def _check_text_array(
|
|
result: ValidationResult,
|
|
path: str,
|
|
value: Any,
|
|
max_items: int,
|
|
max_length: int,
|
|
min_items: int = 0,
|
|
) -> Optional[List[Any]]:
|
|
values = _check_array(result, path, value, max_items, min_items)
|
|
if values is not None:
|
|
for index, item in enumerate(values):
|
|
_check_text(result, f"{path}[{index}]", item, max_length)
|
|
return values
|
|
|
|
|
|
def _check_evidence_refs(
|
|
result: ValidationResult,
|
|
path: str,
|
|
value: Any,
|
|
min_items: int = 0,
|
|
) -> Optional[List[Any]]:
|
|
refs = _check_array(result, path, value, 24, min_items)
|
|
if refs is not None:
|
|
for index, ref in enumerate(refs):
|
|
_check_pattern(result, f"{path}[{index}]", ref, EVIDENCE_ID_PATTERN, "evidence id")
|
|
return refs
|
|
|
|
|
|
def _validate_subject(result: ValidationResult, value: Any) -> None:
|
|
subject = _expect_object(result, "subject", value)
|
|
if subject is None:
|
|
return
|
|
required = ("displayName", "birthTimeStatus", "birthPlaceLabel")
|
|
_check_keys(result, "subject", subject, required)
|
|
_check_text(result, "subject.displayName", subject.get("displayName"), TEXT_LIMITS["displayName"])
|
|
_check_enum(result, "subject.birthTimeStatus", subject.get("birthTimeStatus"), BIRTH_TIME_STATUSES)
|
|
_check_text(result, "subject.birthPlaceLabel", subject.get("birthPlaceLabel"), TEXT_LIMITS["birthPlaceLabel"])
|
|
|
|
|
|
def _validate_provenance(result: ValidationResult, value: Any, version: str) -> None:
|
|
provenance = _expect_object(result, "provenance", value)
|
|
if provenance is None:
|
|
return
|
|
common = ("skillSourceCommit", "skillSnapshotSha256", "calculationHash", "evidenceHash", "reportContractVersion")
|
|
required = common if version == V1_SCHEMA_VERSION else ("skillName", "skillVersion", *common)
|
|
allowed = ("skillName", "skillVersion", *common)
|
|
_check_keys(result, "provenance", provenance, required, allowed)
|
|
if "skillName" in provenance:
|
|
_check_pattern(result, "provenance.skillName", provenance.get("skillName"), SKILL_NAME_PATTERN, "skill name")
|
|
if isinstance(provenance.get("skillName"), str) and len(provenance["skillName"]) > 120:
|
|
result.add("provenance.skillName", "length must be 1..120")
|
|
if "skillVersion" in provenance:
|
|
_check_pattern(result, "provenance.skillVersion", provenance.get("skillVersion"), SKILL_VERSION_PATTERN, "skill version")
|
|
if isinstance(provenance.get("skillVersion"), str) and len(provenance["skillVersion"]) > 80:
|
|
result.add("provenance.skillVersion", "length must be 1..80")
|
|
commit = provenance.get("skillSourceCommit")
|
|
if commit is not None:
|
|
_check_pattern(result, "provenance.skillSourceCommit", commit, SHA1_PATTERN, "commit sha")
|
|
for key in ("skillSnapshotSha256", "calculationHash", "evidenceHash"):
|
|
_check_pattern(result, f"provenance.{key}", provenance.get(key), SHA256_PATTERN, "sha256 hex")
|
|
expected_contract = REPORT_CONTRACT_VERSION if version == V1_SCHEMA_VERSION else CURRENT_REPORT_CONTRACT_VERSION
|
|
if provenance.get("reportContractVersion") != expected_contract:
|
|
result.add("provenance.reportContractVersion", f"must equal {expected_contract!r}")
|
|
|
|
|
|
def _validate_executive_summary(result: ValidationResult, value: Any, version: str) -> None:
|
|
summary = _expect_object(result, "executiveSummary", value)
|
|
if summary is None:
|
|
return
|
|
required = ["headline", "summary", "priorities", "overallClaimStatus"]
|
|
if version == V2_SCHEMA_VERSION:
|
|
required.append("evidenceRefs")
|
|
_check_keys(result, "executiveSummary", summary, required)
|
|
_check_text(result, "executiveSummary.headline", summary.get("headline"), TEXT_LIMITS["headline"])
|
|
_check_text(result, "executiveSummary.summary", summary.get("summary"), TEXT_LIMITS["summary"])
|
|
_check_text_array(result, "executiveSummary.priorities", summary.get("priorities"), 8, TEXT_LIMITS["priority"])
|
|
_check_enum(result, "executiveSummary.overallClaimStatus", summary.get("overallClaimStatus"), CLAIM_STATUSES)
|
|
if version == V2_SCHEMA_VERSION:
|
|
_check_evidence_refs(result, "executiveSummary.evidenceRefs", summary.get("evidenceRefs"), 1)
|
|
|
|
|
|
def _validate_house(result: ValidationResult, path: str, value: Any) -> None:
|
|
house = _expect_object(result, path, value)
|
|
if house is None:
|
|
return
|
|
required = ("houseNumber", "sign", "occupants")
|
|
_check_keys(result, path, house, required)
|
|
number = house.get("houseNumber")
|
|
if not isinstance(number, int) or isinstance(number, bool) or number not in HOUSE_NUMBERS:
|
|
result.add(f"{path}.houseNumber", f"must be integer 1..12, got {number!r}")
|
|
_check_text(result, f"{path}.sign", house.get("sign"), TEXT_LIMITS["sign"])
|
|
_check_text_array(result, f"{path}.occupants", house.get("occupants"), 12, TEXT_LIMITS["occupant"])
|
|
|
|
|
|
def _validate_planet(result: ValidationResult, path: str, value: Any) -> None:
|
|
planet = _expect_object(result, path, value)
|
|
if planet is None:
|
|
return
|
|
required = ("name", "sign", "longitudeDegrees", "houseNumber", "retrograde")
|
|
_check_keys(result, path, planet, required)
|
|
_check_text(result, f"{path}.name", planet.get("name"), TEXT_LIMITS["planetName"])
|
|
_check_text(result, f"{path}.sign", planet.get("sign"), TEXT_LIMITS["sign"])
|
|
longitude = planet.get("longitudeDegrees")
|
|
if not isinstance(longitude, (int, float)) or isinstance(longitude, bool) or not (0 <= longitude < 360):
|
|
result.add(f"{path}.longitudeDegrees", f"must be number 0..360 (360 excluded), got {longitude!r}")
|
|
house_number = planet.get("houseNumber")
|
|
if not isinstance(house_number, int) or isinstance(house_number, bool) or house_number not in HOUSE_NUMBERS:
|
|
result.add(f"{path}.houseNumber", f"must be integer 1..12, got {house_number!r}")
|
|
if not isinstance(planet.get("retrograde"), bool):
|
|
result.add(f"{path}.retrograde", "must be boolean")
|
|
|
|
|
|
def _validate_chart(result: ValidationResult, path: str, value: Any, version: str) -> None:
|
|
chart = _expect_object(result, path, value)
|
|
if chart is None:
|
|
return
|
|
required = ["id", "title", "houses", "claimStatus"]
|
|
allowed = [*required, "planets"]
|
|
if version == V2_SCHEMA_VERSION:
|
|
required.append("evidenceRefs")
|
|
allowed.append("evidenceRefs")
|
|
_check_keys(result, path, chart, required, allowed)
|
|
_check_enum(result, f"{path}.id", chart.get("id"), V1_CHART_IDS if version == V1_SCHEMA_VERSION else CHART_IDS)
|
|
_check_text(result, f"{path}.title", chart.get("title"), TEXT_LIMITS["chartTitle"])
|
|
_check_enum(result, f"{path}.claimStatus", chart.get("claimStatus"), CLAIM_STATUSES)
|
|
houses = _check_array(result, f"{path}.houses", chart.get("houses"), 12)
|
|
seen_houses: List[int] = []
|
|
if houses is not None:
|
|
for index, house in enumerate(houses):
|
|
_validate_house(result, f"{path}.houses[{index}]", house)
|
|
number = house.get("houseNumber") if isinstance(house, dict) else None
|
|
if isinstance(number, int) and not isinstance(number, bool):
|
|
if number in seen_houses:
|
|
result.add(f"{path}.houses[{index}]", f"duplicate houseNumber {number!r}")
|
|
seen_houses.append(number)
|
|
if "planets" in chart:
|
|
planets = _check_array(result, f"{path}.planets", chart.get("planets"), 12)
|
|
if planets is not None:
|
|
for index, planet in enumerate(planets):
|
|
_validate_planet(result, f"{path}.planets[{index}]", planet)
|
|
if version == V2_SCHEMA_VERSION:
|
|
_check_evidence_refs(result, f"{path}.evidenceRefs", chart.get("evidenceRefs"), 1)
|
|
|
|
|
|
def _validate_thematic_section(result: ValidationResult, path: str, value: Any, version: str) -> None:
|
|
section = _expect_object(result, path, value)
|
|
if section is None:
|
|
return
|
|
required = ["id", "title", "narrative", "actions", "caveats", "claimStatus", "evidenceRefs"]
|
|
if version == V2_SCHEMA_VERSION:
|
|
required.insert(1, "theme")
|
|
_check_keys(result, path, section, required)
|
|
_check_pattern(
|
|
result,
|
|
f"{path}.id",
|
|
section.get("id"),
|
|
V1_SECTION_ID_PATTERN if version == V1_SCHEMA_VERSION else SECTION_ID_PATTERN,
|
|
"section id",
|
|
)
|
|
if version == V2_SCHEMA_VERSION:
|
|
_check_pattern(result, f"{path}.theme", section.get("theme"), THEME_ID_PATTERN, "theme id")
|
|
_check_text(result, f"{path}.title", section.get("title"), TEXT_LIMITS["sectionTitle"])
|
|
_check_text(result, f"{path}.narrative", section.get("narrative"), TEXT_LIMITS["narrative"])
|
|
_check_text_array(result, f"{path}.actions", section.get("actions"), 12, TEXT_LIMITS["action"])
|
|
_check_text_array(result, f"{path}.caveats", section.get("caveats"), 12, TEXT_LIMITS["caveat"])
|
|
_check_enum(result, f"{path}.claimStatus", section.get("claimStatus"), CLAIM_STATUSES)
|
|
_check_evidence_refs(result, f"{path}.evidenceRefs", section.get("evidenceRefs"), 1 if version == V2_SCHEMA_VERSION else 0)
|
|
|
|
|
|
def _validate_evidence_appendix(result: ValidationResult, value: Any) -> None:
|
|
appendix = _expect_object(result, "evidenceAppendix", value)
|
|
if appendix is None:
|
|
return
|
|
required = ("expandedByDefault", "techniqueAudit", "conflicts", "calculationEvidence", "blockedTechniques")
|
|
_check_keys(result, "evidenceAppendix", appendix, required)
|
|
if not isinstance(appendix.get("expandedByDefault"), bool):
|
|
result.add("evidenceAppendix.expandedByDefault", "must be boolean")
|
|
|
|
audit = _check_array(result, "evidenceAppendix.techniqueAudit", appendix.get("techniqueAudit"), 100)
|
|
if audit is not None:
|
|
for index, raw in enumerate(audit):
|
|
path = f"evidenceAppendix.techniqueAudit[{index}]"
|
|
row = _expect_object(result, path, raw)
|
|
if row is None:
|
|
continue
|
|
required_row = ("id", "techniqueId", "techniqueName", "status", "used")
|
|
_check_keys(result, path, row, required_row, (*required_row, "notes"))
|
|
_check_pattern(result, f"{path}.id", row.get("id"), EVIDENCE_ID_PATTERN, "evidence id")
|
|
_check_pattern(result, f"{path}.techniqueId", row.get("techniqueId"), TECHNIQUE_ID_PATTERN, "technique id")
|
|
_check_text(result, f"{path}.techniqueName", row.get("techniqueName"), TEXT_LIMITS["techniqueName"])
|
|
_check_enum(result, f"{path}.status", row.get("status"), TECHNIQUE_STATUSES)
|
|
if not isinstance(row.get("used"), bool):
|
|
result.add(f"{path}.used", "must be boolean")
|
|
if "notes" in row:
|
|
_check_text(result, f"{path}.notes", row.get("notes"), TEXT_LIMITS["notes"], 0)
|
|
|
|
conflicts = _check_array(result, "evidenceAppendix.conflicts", appendix.get("conflicts"), 50)
|
|
if conflicts is not None:
|
|
for index, raw in enumerate(conflicts):
|
|
path = f"evidenceAppendix.conflicts[{index}]"
|
|
row = _expect_object(result, path, raw)
|
|
if row is None:
|
|
continue
|
|
required_row = ("id", "description", "impact", "status")
|
|
_check_keys(result, path, row, required_row)
|
|
_check_pattern(result, f"{path}.id", row.get("id"), EVIDENCE_ID_PATTERN, "evidence id")
|
|
_check_text(result, f"{path}.description", row.get("description"), TEXT_LIMITS["conflictDescription"])
|
|
_check_text(result, f"{path}.impact", row.get("impact"), TEXT_LIMITS["conflictImpact"])
|
|
_check_enum(result, f"{path}.status", row.get("status"), CONFLICT_STATUSES)
|
|
|
|
calculations = _check_array(result, "evidenceAppendix.calculationEvidence", appendix.get("calculationEvidence"), 100)
|
|
if calculations is not None:
|
|
for index, raw in enumerate(calculations):
|
|
path = f"evidenceAppendix.calculationEvidence[{index}]"
|
|
row = _expect_object(result, path, raw)
|
|
if row is None:
|
|
continue
|
|
required_row = ("id", "label", "value", "source")
|
|
_check_keys(result, path, row, required_row)
|
|
_check_pattern(result, f"{path}.id", row.get("id"), EVIDENCE_ID_PATTERN, "evidence id")
|
|
_check_text(result, f"{path}.label", row.get("label"), TEXT_LIMITS["evidenceLabel"])
|
|
_check_text(result, f"{path}.value", row.get("value"), TEXT_LIMITS["evidenceValue"])
|
|
_check_text(result, f"{path}.source", row.get("source"), TEXT_LIMITS["evidenceSource"])
|
|
|
|
_check_text_array(result, "evidenceAppendix.blockedTechniques", appendix.get("blockedTechniques"), 100, TEXT_LIMITS["blockedTechnique"])
|
|
|
|
|
|
def _validate_natal_foundation(result: ValidationResult, value: Any) -> None:
|
|
foundation = _expect_object(result, "natalFoundation", value)
|
|
if foundation is None:
|
|
return
|
|
required = ("title", "narrative", "keyFactors", "caveats", "claimStatus", "evidenceRefs")
|
|
_check_keys(result, "natalFoundation", foundation, required)
|
|
_check_text(result, "natalFoundation.title", foundation.get("title"), TEXT_LIMITS["sectionTitle"])
|
|
_check_text(result, "natalFoundation.narrative", foundation.get("narrative"), TEXT_LIMITS["narrative"])
|
|
_check_text_array(result, "natalFoundation.keyFactors", foundation.get("keyFactors"), 12, TEXT_LIMITS["keyFactor"])
|
|
_check_text_array(result, "natalFoundation.caveats", foundation.get("caveats"), 12, TEXT_LIMITS["caveat"])
|
|
_check_enum(result, "natalFoundation.claimStatus", foundation.get("claimStatus"), CLAIM_STATUSES)
|
|
_check_evidence_refs(result, "natalFoundation.evidenceRefs", foundation.get("evidenceRefs"), 1)
|
|
|
|
|
|
def _validate_current_phase(result: ValidationResult, value: Any) -> None:
|
|
if value is None:
|
|
return
|
|
phase = _expect_object(result, "currentPhase", value)
|
|
if phase is None:
|
|
return
|
|
required = ("title", "phaseLabel", "narrative", "timingNotes", "caveats", "claimStatus", "evidenceRefs")
|
|
_check_keys(result, "currentPhase", phase, required)
|
|
_check_text(result, "currentPhase.title", phase.get("title"), TEXT_LIMITS["sectionTitle"])
|
|
_check_text(result, "currentPhase.phaseLabel", phase.get("phaseLabel"), TEXT_LIMITS["phaseLabel"])
|
|
_check_text(result, "currentPhase.narrative", phase.get("narrative"), TEXT_LIMITS["narrative"])
|
|
_check_text_array(result, "currentPhase.timingNotes", phase.get("timingNotes"), 12, TEXT_LIMITS["timingNote"])
|
|
_check_text_array(result, "currentPhase.caveats", phase.get("caveats"), 12, TEXT_LIMITS["caveat"])
|
|
_check_enum(result, "currentPhase.claimStatus", phase.get("claimStatus"), CLAIM_STATUSES)
|
|
_check_evidence_refs(result, "currentPhase.evidenceRefs", phase.get("evidenceRefs"), 1)
|
|
|
|
|
|
def _validate_action_notes(result: ValidationResult, value: Any) -> None:
|
|
notes = _check_array(result, "actionNotes", value, 24, 1)
|
|
if notes is None:
|
|
return
|
|
for index, raw in enumerate(notes):
|
|
path = f"actionNotes[{index}]"
|
|
note = _expect_object(result, path, raw)
|
|
if note is None:
|
|
continue
|
|
required = ("id", "title", "note", "priority", "evidenceRefs")
|
|
_check_keys(result, path, note, required)
|
|
_check_pattern(result, f"{path}.id", note.get("id"), SECTION_ID_PATTERN, "section id")
|
|
_check_text(result, f"{path}.title", note.get("title"), TEXT_LIMITS["sectionTitle"])
|
|
_check_text(result, f"{path}.note", note.get("note"), TEXT_LIMITS["actionNote"])
|
|
_check_enum(result, f"{path}.priority", note.get("priority"), REPORT_ACTION_PRIORITIES)
|
|
_check_evidence_refs(result, f"{path}.evidenceRefs", note.get("evidenceRefs"), 1)
|
|
|
|
|
|
def _validate_blocked_disclosures(result: ValidationResult, value: Any) -> None:
|
|
sections = _check_array(result, "blockedConflictDisclosure", value, 12)
|
|
if sections is None:
|
|
return
|
|
for index, raw in enumerate(sections):
|
|
path = f"blockedConflictDisclosure[{index}]"
|
|
section = _expect_object(result, path, raw)
|
|
if section is None:
|
|
continue
|
|
required = ("theme", "title", "reason", "missingEvidence", "conflictNotes", "evidenceRefs", "claimStatus")
|
|
_check_keys(result, path, section, required)
|
|
_check_pattern(result, f"{path}.theme", section.get("theme"), THEME_ID_PATTERN, "theme id")
|
|
_check_text(result, f"{path}.title", section.get("title"), TEXT_LIMITS["sectionTitle"])
|
|
_check_text(result, f"{path}.reason", section.get("reason"), TEXT_LIMITS["reason"])
|
|
_check_text_array(result, f"{path}.missingEvidence", section.get("missingEvidence"), 24, TEXT_LIMITS["missingEvidence"], 1)
|
|
_check_text_array(result, f"{path}.conflictNotes", section.get("conflictNotes"), 24, TEXT_LIMITS["conflictNote"])
|
|
_check_evidence_refs(result, f"{path}.evidenceRefs", section.get("evidenceRefs"))
|
|
if section.get("claimStatus") != "blocked":
|
|
result.add(f"{path}.claimStatus", "must equal 'blocked'")
|
|
|
|
|
|
def _text_leaves(value: Any, path: str = "") -> Iterable[Tuple[str, str]]:
|
|
if isinstance(value, str):
|
|
yield path or "(root)", value
|
|
elif isinstance(value, list):
|
|
for index, entry in enumerate(value):
|
|
yield from _text_leaves(entry, f"{path}[{index}]")
|
|
elif isinstance(value, dict):
|
|
for key, entry in value.items():
|
|
yield from _text_leaves(entry, f"{path}.{key}" if path else key)
|
|
|
|
|
|
def _known_evidence_ids(document: Dict[str, Any]) -> Tuple[set[str], List[Tuple[str, str]]]:
|
|
appendix = document.get("evidenceAppendix")
|
|
known: set[str] = set()
|
|
duplicates: List[Tuple[str, str]] = []
|
|
locations: Dict[str, str] = {}
|
|
if not isinstance(appendix, dict):
|
|
return known, duplicates
|
|
for key in ("techniqueAudit", "conflicts", "calculationEvidence"):
|
|
rows = appendix.get(key)
|
|
if not isinstance(rows, list):
|
|
continue
|
|
for index, row in enumerate(rows):
|
|
evidence_id = row.get("id") if isinstance(row, dict) else None
|
|
if not isinstance(evidence_id, str):
|
|
continue
|
|
location = f"{key}[{index}]"
|
|
if evidence_id in locations:
|
|
duplicates.append((evidence_id, f"{locations[evidence_id]} and {location}"))
|
|
else:
|
|
locations[evidence_id] = location
|
|
known.add(evidence_id)
|
|
return known, duplicates
|
|
|
|
|
|
def _evidence_ref_entries(document: Dict[str, Any], version: str) -> Iterable[Tuple[str, List[Any]]]:
|
|
sections = document.get("thematicNarrative")
|
|
if isinstance(sections, list):
|
|
for index, section in enumerate(sections):
|
|
if isinstance(section, dict) and isinstance(section.get("evidenceRefs"), list):
|
|
path = str(section.get("id")) if version == V1_SCHEMA_VERSION else f"thematicNarrative[{index}].evidenceRefs"
|
|
yield path, section["evidenceRefs"]
|
|
if version != V2_SCHEMA_VERSION:
|
|
return
|
|
for path, value in (
|
|
("executiveSummary.evidenceRefs", document.get("executiveSummary")),
|
|
("natalFoundation.evidenceRefs", document.get("natalFoundation")),
|
|
("currentPhase.evidenceRefs", document.get("currentPhase")),
|
|
):
|
|
if isinstance(value, dict) and isinstance(value.get("evidenceRefs"), list):
|
|
yield path, value["evidenceRefs"]
|
|
for key in ("actionNotes", "charts", "blockedConflictDisclosure"):
|
|
rows = document.get(key)
|
|
if isinstance(rows, list):
|
|
for index, row in enumerate(rows):
|
|
if isinstance(row, dict) and isinstance(row.get("evidenceRefs"), list):
|
|
yield f"{key}[{index}].evidenceRefs", row["evidenceRefs"]
|
|
|
|
|
|
def _semantic_guards(result: ValidationResult, document: Dict[str, Any], version: str) -> None:
|
|
charts = document.get("charts")
|
|
if isinstance(charts, list):
|
|
ids = [chart.get("id") for chart in charts if isinstance(chart, dict)]
|
|
d1_count = sum(chart_id == "D1" for chart_id in ids)
|
|
if d1_count != 1:
|
|
result.add("charts", f"must contain exactly one D1 chart, found {d1_count}")
|
|
seen: set[Any] = set()
|
|
for chart_id in ids:
|
|
if chart_id in seen:
|
|
result.add("charts", f"duplicate chart id {chart_id}")
|
|
seen.add(chart_id)
|
|
d1 = next((chart for chart in charts if isinstance(chart, dict) and chart.get("id") == "D1"), None)
|
|
if isinstance(d1, dict) and isinstance(d1.get("houses"), list):
|
|
numbers = [house.get("houseNumber") for house in d1["houses"] if isinstance(house, dict)]
|
|
if len(numbers) != 12 or set(numbers) != set(HOUSE_NUMBERS):
|
|
result.add("charts.D1.houses", "D1 chart must contain all twelve house numbers 1..12 exactly once")
|
|
|
|
known_ids, duplicates = _known_evidence_ids(document)
|
|
for evidence_id, locations in duplicates:
|
|
result.add("evidenceAppendix", f"duplicate evidence id {evidence_id} used in {locations}")
|
|
for path, refs in _evidence_ref_entries(document, version):
|
|
for ref in refs:
|
|
if isinstance(ref, str) and ref not in known_ids:
|
|
result.add(path, f"unknown evidence id {ref}")
|
|
|
|
blocked_values: List[Tuple[str, Any]] = []
|
|
summary = document.get("executiveSummary")
|
|
if isinstance(summary, dict) and summary.get("overallClaimStatus") == "blocked":
|
|
blocked_values.append(("executiveSummary", summary))
|
|
if isinstance(charts, list):
|
|
blocked_values.extend(
|
|
(f"charts[{index}]", chart)
|
|
for index, chart in enumerate(charts)
|
|
if isinstance(chart, dict) and chart.get("claimStatus") == "blocked"
|
|
)
|
|
sections = document.get("thematicNarrative")
|
|
if isinstance(sections, list):
|
|
blocked_values.extend(
|
|
(f"thematicNarrative[{index}]", section)
|
|
for index, section in enumerate(sections)
|
|
if isinstance(section, dict) and section.get("claimStatus") == "blocked"
|
|
)
|
|
if version == V2_SCHEMA_VERSION:
|
|
for path, value in (("natalFoundation", document.get("natalFoundation")), ("currentPhase", document.get("currentPhase"))):
|
|
if isinstance(value, dict) and value.get("claimStatus") == "blocked":
|
|
blocked_values.append((path, value))
|
|
disclosures = document.get("blockedConflictDisclosure")
|
|
if isinstance(disclosures, list):
|
|
blocked_values.extend((f"blockedConflictDisclosure[{index}]", row) for index, row in enumerate(disclosures) if isinstance(row, dict))
|
|
for base_path, value in blocked_values:
|
|
for path, text in _text_leaves(value, base_path):
|
|
if any(pattern.search(text) for pattern in DETERMINISTIC_COMPILED):
|
|
result.add(path, "blocked section contains deterministic prediction")
|
|
|
|
if version == V2_SCHEMA_VERSION:
|
|
requested = document.get("requestedThemes")
|
|
thematic = document.get("thematicNarrative")
|
|
disclosures = document.get("blockedConflictDisclosure")
|
|
if isinstance(requested, list) and isinstance(thematic, list) and isinstance(disclosures, list):
|
|
requested_counts: Dict[str, int] = {}
|
|
for theme in requested:
|
|
if isinstance(theme, str):
|
|
requested_counts[theme] = requested_counts.get(theme, 0) + 1
|
|
for theme, count in requested_counts.items():
|
|
if count != 1:
|
|
result.add("requestedThemes", f"must contain {theme} exactly once, found {count}")
|
|
coverage: Dict[str, int] = {}
|
|
requested_set = set(theme for theme in requested if isinstance(theme, str))
|
|
for index, section in enumerate(thematic):
|
|
theme = section.get("theme") if isinstance(section, dict) else None
|
|
if isinstance(theme, str):
|
|
coverage[theme] = coverage.get(theme, 0) + 1
|
|
if theme not in requested_set:
|
|
result.add(f"thematicNarrative[{index}].theme", f"covers unrequested theme {theme}")
|
|
for index, section in enumerate(disclosures):
|
|
theme = section.get("theme") if isinstance(section, dict) else None
|
|
if isinstance(theme, str):
|
|
coverage[theme] = coverage.get(theme, 0) + 1
|
|
if theme not in requested_set:
|
|
result.add(f"blockedConflictDisclosure[{index}].theme", f"covers unrequested theme {theme}")
|
|
for theme in requested_set:
|
|
count = coverage.get(theme, 0)
|
|
if count != 1:
|
|
result.add("requestedThemes", f"requested theme {theme} must have exactly one thematic section or blocked disclosure, found {count}")
|
|
chart_ids = set(chart.get("id") for chart in charts if isinstance(chart, dict)) if isinstance(charts, list) else set()
|
|
for index, section in enumerate(thematic):
|
|
theme = section.get("theme") if isinstance(section, dict) else None
|
|
if not isinstance(theme, str):
|
|
continue
|
|
for required_chart in REQUIRED_THEME_CHARTS.get(theme, ()):
|
|
if required_chart not in chart_ids:
|
|
result.add(f"thematicNarrative[{index}]", f"theme {theme} requires structured {required_chart} chart data or a blocked disclosure")
|
|
|
|
date_sections: List[Tuple[str, Any, List[Any]]] = []
|
|
for path, value in (
|
|
("executiveSummary", document.get("executiveSummary")),
|
|
("natalFoundation", document.get("natalFoundation")),
|
|
("currentPhase", document.get("currentPhase")),
|
|
):
|
|
if isinstance(value, dict) and isinstance(value.get("evidenceRefs"), list):
|
|
date_sections.append((path, value, value["evidenceRefs"]))
|
|
for key in ("thematicNarrative", "actionNotes", "blockedConflictDisclosure"):
|
|
rows = document.get(key)
|
|
if isinstance(rows, list):
|
|
for index, row in enumerate(rows):
|
|
if isinstance(row, dict) and isinstance(row.get("evidenceRefs"), list):
|
|
date_sections.append((f"{key}[{index}]", row, row["evidenceRefs"]))
|
|
for path, value, refs in date_sections:
|
|
if not refs and any(DATE_CLAIM_PATTERN.search(text) for _, text in _text_leaves(value)):
|
|
result.add(path, "date claim requires evidenceRefs")
|
|
|
|
for path, text in _text_leaves(document):
|
|
hits = [name for name, pattern in FORBIDDEN_COMPILED if pattern.search(text)]
|
|
if hits:
|
|
result.add(path, f"forbidden content {','.join(hits)}")
|
|
|
|
provenance = document.get("provenance")
|
|
appendix = document.get("evidenceAppendix")
|
|
if isinstance(provenance, dict) and isinstance(appendix, dict):
|
|
expected_hash = compute_evidence_hash(document)
|
|
if provenance.get("evidenceHash") != expected_hash:
|
|
result.add("provenance.evidenceHash", f"does not match computed evidence hash {expected_hash}")
|
|
|
|
try:
|
|
size = serialized_bytes(document)
|
|
except (TypeError, ValueError, OverflowError) as error:
|
|
result.add("(size)", f"unable to serialize document: {error}")
|
|
else:
|
|
if size > MAX_SERIALIZED_BYTES:
|
|
result.add("(size)", f"serialized document is {size} bytes, exceeding {MAX_SERIALIZED_BYTES}")
|
|
|
|
|
|
def validate_report_document(document: Any) -> ValidationResult:
|
|
result = ValidationResult(valid=False)
|
|
if not isinstance(document, dict):
|
|
result.add("(root)", f"must be object, got {type(document).__name__}")
|
|
return result
|
|
|
|
version = document.get("schemaVersion")
|
|
if version not in (V1_SCHEMA_VERSION, V2_SCHEMA_VERSION):
|
|
result.add("schemaVersion", f"must be one of {(V1_SCHEMA_VERSION, V2_SCHEMA_VERSION)!r}, got {version!r}")
|
|
return result
|
|
|
|
v1_required = (
|
|
"schemaVersion", "reportId", "reportType", "presentationMode", "generatedAt", "subject", "provenance",
|
|
"executiveSummary", "charts", "thematicNarrative", "evidenceAppendix", "disclaimer",
|
|
)
|
|
v2_required = (
|
|
"schemaVersion", "reportId", "reportType", "presentationMode", "depth", "requestedThemes", "generatedAt",
|
|
"subject", "provenance", "executiveSummary", "natalFoundation", "currentPhase", "actionNotes", "charts",
|
|
"thematicNarrative", "blockedConflictDisclosure", "evidenceAppendix", "disclaimer",
|
|
)
|
|
required = v1_required if version == V1_SCHEMA_VERSION else v2_required
|
|
_check_keys(result, "(root)", document, required)
|
|
_check_pattern(result, "reportId", document.get("reportId"), UUID_PATTERN, "uuid")
|
|
_check_enum(result, "reportType", document.get("reportType"), REPORT_TYPES)
|
|
_check_enum(result, "presentationMode", document.get("presentationMode"), PRESENTATION_MODES)
|
|
if version == V2_SCHEMA_VERSION:
|
|
_check_enum(result, "depth", document.get("depth"), REPORT_DEPTHS)
|
|
themes = _check_array(result, "requestedThemes", document.get("requestedThemes"), 12, 1)
|
|
if themes is not None:
|
|
for index, theme in enumerate(themes):
|
|
_check_pattern(result, f"requestedThemes[{index}]", theme, THEME_ID_PATTERN, "theme id")
|
|
_check_pattern(result, "generatedAt", document.get("generatedAt"), ISO8601_PATTERN, "ISO-8601 timestamp")
|
|
_validate_subject(result, document.get("subject"))
|
|
_validate_provenance(result, document.get("provenance"), version)
|
|
_validate_executive_summary(result, document.get("executiveSummary"), version)
|
|
if version == V2_SCHEMA_VERSION:
|
|
_validate_natal_foundation(result, document.get("natalFoundation"))
|
|
_validate_current_phase(result, document.get("currentPhase"))
|
|
_validate_action_notes(result, document.get("actionNotes"))
|
|
|
|
charts = _check_array(result, "charts", document.get("charts"), 3 if version == V1_SCHEMA_VERSION else 6, 1)
|
|
if charts is not None:
|
|
for index, chart in enumerate(charts):
|
|
_validate_chart(result, f"charts[{index}]", chart, version)
|
|
|
|
sections = _check_array(result, "thematicNarrative", document.get("thematicNarrative"), 12)
|
|
seen_section_ids: set[str] = set()
|
|
if sections is not None:
|
|
for index, section in enumerate(sections):
|
|
_validate_thematic_section(result, f"thematicNarrative[{index}]", section, version)
|
|
section_id = section.get("id") if isinstance(section, dict) else None
|
|
if isinstance(section_id, str):
|
|
if section_id in seen_section_ids:
|
|
result.add(f"thematicNarrative[{index}].id", f"duplicate section id {section_id!r}")
|
|
seen_section_ids.add(section_id)
|
|
|
|
if version == V2_SCHEMA_VERSION:
|
|
_validate_blocked_disclosures(result, document.get("blockedConflictDisclosure"))
|
|
_validate_evidence_appendix(result, document.get("evidenceAppendix"))
|
|
_check_text(result, "disclaimer", document.get("disclaimer"), TEXT_LIMITS["disclaimer"])
|
|
_semantic_guards(result, document, version)
|
|
result.valid = not result.errors
|
|
return result
|
|
|
|
|
|
def is_valid_report_document(document: Any) -> bool:
|
|
return validate_report_document(document).valid
|
|
|
|
|
|
def load_report_document(path: str) -> Dict[str, Any]:
|
|
with open(path, "r", encoding="utf-8") as handle:
|
|
return json.load(handle)
|
|
|
|
|
|
def parse_report_document_json(text: str) -> ValidationResult:
|
|
try:
|
|
document = json.loads(text)
|
|
except json.JSONDecodeError as error:
|
|
result = ValidationResult(valid=False)
|
|
result.add("(json)", f"invalid JSON: {error}")
|
|
return result
|
|
return validate_report_document(document)
|
|
|
|
|
|
def main(argv: Optional[List[str]] = None) -> int:
|
|
args = list(sys.argv[1:] if argv is None else argv)
|
|
if not args:
|
|
print("usage: python3 scripts/personal_report_contract.py <report-document.json>", file=sys.stderr)
|
|
return 2
|
|
path = args[0]
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as handle:
|
|
text = handle.read()
|
|
except OSError as error:
|
|
print(f"unable to read {path}: {error}", file=sys.stderr)
|
|
return 2
|
|
result = parse_report_document_json(text)
|
|
if result.valid:
|
|
try:
|
|
size = serialized_bytes(json.loads(text))
|
|
except (TypeError, ValueError):
|
|
size = 0
|
|
print(f"valid: {path} ({size} bytes)")
|
|
return 0
|
|
print(f"invalid: {path}", file=sys.stderr)
|
|
for error in result.errors:
|
|
print(f" - {error}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|