File size: 3,947 Bytes
58c2da3 4af38c8 58c2da3 4af38c8 58c2da3 4af38c8 58c2da3 4af38c8 58c2da3 4af38c8 58c2da3 4af38c8 58c2da3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | """JSON Schema helpers for dataset-only output contracts."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from pydantic import BaseModel
from gcmd_classifier.datasets.application_service import AwaitingReadmeSelection
from gcmd_classifier.datasets.comparison_models import (
DatasetComparisonResult,
HumanReviewRecord,
)
from gcmd_classifier.datasets.deterministic_evidence import (
DeterministicEvidencePacket,
DeterministicEvidenceSelection,
)
from gcmd_classifier.datasets.models import (
CoverageResolutionFailure,
DatasetClassificationResult,
DatasetCoverageResolution,
DatasetEvidencePacket,
DatasetExtractionManifest,
)
from gcmd_classifier.datasets.runtime_models import (
ExpertSourceBinding,
SealedExpertArtifactEnvelope,
VerifiedPersistedBlindResultReference,
)
from gcmd_classifier.datasets.structural_scope import (
DeterministicScopeResult,
StructuralPassageIndex,
)
DATASET_SCHEMA_MODELS: dict[str, type[BaseModel]] = {
"dataset_structural_passage_index.schema.json": StructuralPassageIndex,
"dataset_deterministic_product_scope.schema.json": DeterministicScopeResult,
"dataset_deterministic_evidence_selection.schema.json": DeterministicEvidenceSelection,
"dataset_deterministic_evidence_packet.schema.json": DeterministicEvidencePacket,
"dataset_awaiting_readme_selection.schema.json": AwaitingReadmeSelection,
"dataset_classification_result.schema.json": DatasetClassificationResult,
"dataset_coverage_resolution.schema.json": DatasetCoverageResolution,
"dataset_coverage_resolution_failure.schema.json": CoverageResolutionFailure,
"dataset_extraction_manifest.schema.json": DatasetExtractionManifest,
"dataset_evidence_packet.schema.json": DatasetEvidencePacket,
"dataset_comparison.schema.json": DatasetComparisonResult,
"dataset_human_review.schema.json": HumanReviewRecord,
"dataset_sealed_expert_artifact.schema.json": SealedExpertArtifactEnvelope,
"dataset_expert_source_binding.schema.json": ExpertSourceBinding,
"dataset_verified_blind_reference.schema.json": VerifiedPersistedBlindResultReference,
}
def schema_for(model_type: type[BaseModel]) -> dict[str, Any]:
"""Generate and validate a Draft 2020-12-compatible Pydantic schema."""
schema = model_type.model_json_schema()
if model_type is DatasetClassificationResult:
schema.setdefault("allOf", []).append(
{
"if": {
"properties": {"classification_attempted": {"const": False}},
"required": ["classification_attempted"],
},
"then": {
"properties": {
"classification_outcome": {"type": "null"},
"classifications": {"maxItems": 0},
"blind_result": {"type": "null"},
"comparison_reference": {"type": "null"},
"processing_metadata": {
"properties": {"model_calls": {"const": 0}},
"required": ["model_calls"],
},
}
},
}
)
return schema
def validate_serialized(model_type: type[BaseModel], payload: dict[str, Any]) -> BaseModel:
"""Validate a serialized dataset payload through its authoritative typed contract."""
return model_type.model_validate(payload)
def write_dataset_schemas(directory: Path) -> tuple[Path, ...]:
"""Write formal dataset schemas deterministically."""
directory.mkdir(parents=True, exist_ok=True)
written: list[Path] = []
for filename, model_type in DATASET_SCHEMA_MODELS.items():
path = directory / filename
path.write_text(json.dumps(schema_for(model_type), indent=2) + "\n")
written.append(path)
return tuple(written)
|