95 lines
3.0 KiB
Python
95 lines
3.0 KiB
Python
"""SQLAlchemy model for OCR results."""
|
|
|
|
import enum
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy import DateTime, Enum, Float, ForeignKey, String, Text, func
|
|
from sqlalchemy.dialects.postgresql import JSONB, UUID
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
|
|
from app.database import Base
|
|
|
|
|
|
class OCRStatus(str, enum.Enum):
|
|
pending = "pending"
|
|
processing = "processing"
|
|
completed = "completed"
|
|
failed = "failed"
|
|
manual_review = "manual_review"
|
|
|
|
|
|
class OCRResult(Base):
|
|
"""OCR result entity linked to a vehicle (optional) and an uploaded scan file."""
|
|
|
|
__tablename__ = "ocr_results"
|
|
|
|
id: Mapped[uuid.UUID] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
primary_key=True,
|
|
default=uuid.uuid4,
|
|
)
|
|
vehicle_id: Mapped[uuid.UUID | None] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("vehicles.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
index=True,
|
|
)
|
|
file_path: Mapped[str] = mapped_column(String(512), nullable=False)
|
|
file_name: Mapped[str] = mapped_column(String(255), nullable=False)
|
|
mime_type: Mapped[str] = mapped_column(
|
|
String(100), nullable=False, default="image/png"
|
|
)
|
|
status: Mapped[str] = mapped_column(
|
|
Enum(OCRStatus, name="ocr_status", create_constraint=True),
|
|
nullable=False,
|
|
default=OCRStatus.pending,
|
|
index=True,
|
|
)
|
|
raw_text: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
structured_data: Mapped[dict[str, Any] | None] = mapped_column(
|
|
JSONB,
|
|
nullable=True,
|
|
)
|
|
confidence_score: Mapped[float | None] = mapped_column(
|
|
Float,
|
|
nullable=True,
|
|
)
|
|
error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
nullable=False,
|
|
server_default=func.now(),
|
|
)
|
|
updated_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
nullable=False,
|
|
server_default=func.now(),
|
|
onupdate=func.now(),
|
|
)
|
|
|
|
vehicle = relationship("Vehicle", backref="ocr_results")
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<OCRResult id={self.id} status={self.status}>"
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""Serialize OCR result for API responses."""
|
|
return {
|
|
"id": str(self.id),
|
|
"vehicle_id": str(self.vehicle_id) if self.vehicle_id else None,
|
|
"file_path": self.file_path,
|
|
"file_name": self.file_name,
|
|
"mime_type": self.mime_type,
|
|
"status": self.status.value
|
|
if isinstance(self.status, OCRStatus)
|
|
else str(self.status),
|
|
"raw_text": self.raw_text,
|
|
"structured_data": self.structured_data,
|
|
"confidence_score": self.confidence_score,
|
|
"error_message": self.error_message,
|
|
"created_at": self.created_at.isoformat() if self.created_at else None,
|
|
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
|
|
}
|