| """Strict JSON Schema sanitization for extractions.""" |
|
|
| from __future__ import annotations |
|
|
| import logging |
| from typing import Any |
|
|
| from jsonschema import Draft7Validator |
|
|
| from app.schemas import DocType |
| from app.services.doctypes import as_doc_type |
| from app.services.extraction_schemas import empty_payload, schema_for |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| def _type_ok(value: Any, allowed: list[str] | str) -> bool: |
| """Check value against JSON Schema type token(s).""" |
| types = allowed if isinstance(allowed, list) else [allowed] |
| for t in types: |
| if t == "null" and value is None: |
| return True |
| if t == "string" and isinstance(value, str): |
| return True |
| if t == "number" and isinstance(value, (int, float)) and ( |
| not isinstance(value, bool) |
| ): |
| return True |
| if t == "integer" and isinstance(value, int) and ( |
| not isinstance(value, bool) |
| ): |
| return True |
| if t == "boolean" and isinstance(value, bool): |
| return True |
| if t == "array" and isinstance(value, list): |
| return True |
| if t == "object" and isinstance(value, dict): |
| return True |
| return False |
|
|
|
|
| def _sanitize_object( |
| raw: Any, |
| props: dict[str, Any], |
| required: list[str], |
| ) -> dict[str, Any]: |
| """Keep only schema properties; invalid → null / [].""" |
| src = raw if isinstance(raw, dict) else {} |
| out: dict[str, Any] = {} |
| for key, prop_schema in props.items(): |
| value = src.get(key, None) |
| out[key] = _sanitize_value(value, prop_schema) |
| for key in required: |
| if key not in out: |
| out[key] = None |
| return out |
|
|
|
|
| def _sanitize_value(value: Any, schema: dict[str, Any]) -> Any: |
| """Sanitize one value against a property schema.""" |
| if "properties" in schema: |
| return _sanitize_object( |
| value, |
| schema.get("properties", {}), |
| schema.get("required", []), |
| ) |
|
|
| if schema.get("type") == "array" or ( |
| isinstance(schema.get("type"), list) |
| and "array" in schema["type"] |
| ): |
| if not isinstance(value, list): |
| return [] |
| item_schema = schema.get("items") or {} |
| cleaned = [] |
| for item in value: |
| cleaned.append(_sanitize_value(item, item_schema)) |
| return cleaned |
|
|
| allowed = schema.get("type", ["string", "null"]) |
| if _type_ok(value, allowed): |
| return value |
| return None |
|
|
|
|
| def sanitize_extracted( |
| doc_type: DocType | str, |
| raw: dict[str, Any] | None, |
| ) -> dict[str, Any]: |
| """ |
| Force payload to match the imposed schema. |
| |
| Missing / invalid fields become null (or [] for arrays). |
| Extra keys are dropped. Never invents values. |
| """ |
| doc_type = as_doc_type(doc_type) |
| schema = schema_for(doc_type) |
| base = empty_payload(doc_type) |
| merged = _sanitize_object( |
| raw or {}, |
| schema.get("properties", {}), |
| schema.get("required", list(base.keys())), |
| ) |
|
|
| validator = Draft7Validator(schema) |
| errors = sorted(validator.iter_errors(merged), key=lambda e: e.path) |
| if errors: |
| logger.warning( |
| "schema validation issues for %s: %s", |
| doc_type.value, |
| "; ".join(e.message for e in errors[:5]), |
| ) |
| for err in errors: |
| path = list(err.path) |
| if not path: |
| continue |
| key = path[0] |
| if key in merged and not isinstance(merged[key], list): |
| merged[key] = None |
| return merged |
|
|
|
|
| def is_schema_valid( |
| doc_type: DocType | str, |
| data: dict[str, Any], |
| ) -> bool: |
| """Return True when data fully validates.""" |
| schema = schema_for(as_doc_type(doc_type)) |
| return Draft7Validator(schema).is_valid(data) |
|
|
