erp-nutzfahrzeuge/backend/app/models/ocr_result.py

"""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,
        }