Spaces:
Sleeping
Sleeping
File size: 3,684 Bytes
efff1ff | 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 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | """Typed request/response contracts and the error hierarchy."""
from __future__ import annotations
import enum
from typing import Any
from pydantic import BaseModel, Field, field_validator
class FieldStatus(str, enum.Enum):
"""Outcome of a single key extraction.
`EXTRACTED` is only assigned above the confidence threshold; anything else
is surfaced to the caller rather than being silently returned as fact.
This distinction is the whole point of the service: the underlying model
will always emit *some* span, so absence must be modelled explicitly.
"""
EXTRACTED = "extracted"
LOW_CONFIDENCE = "low_confidence"
NOT_FOUND = "not_found"
ERROR = "error"
class SourceType(str, enum.Enum):
PDF_TEXT_LAYER = "pdf_text_layer"
PDF_OCR = "pdf_ocr"
IMAGE_OCR = "image_ocr"
class ExtractedField(BaseModel):
key: str
value: str | None = None
status: FieldStatus = FieldStatus.NOT_FOUND
confidence: float = Field(default=0.0, ge=0.0, le=1.0)
page: int | None = None
source: SourceType | None = None
error: str | None = None
@property
def is_usable(self) -> bool:
return self.status is FieldStatus.EXTRACTED
class DocumentInfo(BaseModel):
document_id: str
page_count: int
pages_processed: int
word_count: int
source_type: SourceType
truncated: bool = False
class ExtractRequest(BaseModel):
"""Request body. `file` arrives separately as an UploadFile."""
keys: list[str] = Field(..., min_length=1)
confidence_threshold: float | None = Field(default=None, ge=0.0, le=1.0)
max_pages: int | None = Field(default=None, ge=1, le=50)
@field_validator("keys")
@classmethod
def _validate_keys(cls, v: list[str]) -> list[str]:
cleaned = [k.strip() for k in v if k and k.strip()]
if not cleaned:
raise ValueError("at least one non-empty key is required")
if any(len(k) > 200 for k in cleaned):
raise ValueError("each key must be 200 characters or fewer")
return cleaned
@field_validator("keys")
@classmethod
def _dedupe(cls, v: list[str]) -> list[str]:
"""Preserve order but drop exact duplicates.
Repeating a key costs a full forward pass and returns an identical
answer, so duplicates are collapsed rather than rejected.
"""
seen: set[str] = set()
out: list[str] = []
for k in v:
if k not in seen:
seen.add(k)
out.append(k)
return out
class ExtractResponse(BaseModel):
document: DocumentInfo
fields: list[ExtractedField]
latency_ms: float
confidence_threshold: float
@property
def values(self) -> dict[str, str | None]:
return {f.key: f.value for f in self.fields}
# --------------------------------------------------------------------------
# errors
# --------------------------------------------------------------------------
class DocExtractError(Exception):
"""Base class. `status_code` is what the HTTP layer will return."""
status_code = 500
def __init__(self, message: str, detail: Any = None):
super().__init__(message)
self.message = message
self.detail = detail
class ValidationError(DocExtractError):
status_code = 422
class UnsupportedMediaError(DocExtractError):
status_code = 415
class PayloadTooLargeError(DocExtractError):
status_code = 413
class CorruptDocumentError(DocExtractError):
status_code = 422
class ExtractionTimeoutError(DocExtractError):
status_code = 504
class ModelUnavailableError(DocExtractError):
status_code = 503
|