"""DMS AI bounded context — models. Design ------ Phase 11 introduces three entities that form the *AI/OCR integration layer* for the DMS. All entities are linked to a specific ``DocumentVersion`` because OCR and AI classification operate on a concrete file snapshot, not on the abstract logical document. OCRResult Tracks a single OCR processing run against a document version. Status lifecycle: PENDING → PROCESSING → COMPLETED / FAILED. ``provider`` is a free-text slug (e.g. ``"tesseract"``, ``"azure_ocr"``, ``"google_vision"``). The DMS layer is provider-agnostic; the value is stored for observability and multi-provider comparison. ``full_text`` holds the concatenated raw OCR output across all pages. ``page_count`` and ``confidence_score`` are aggregate metrics. Multiple OCR runs against the same version are allowed (e.g. retry after failure, re-run with a different provider). Soft-delete is supported so stale/failed runs can be hidden without losing audit history. AIClassification A single AI classification result attached to a document version. ``classification_type`` identifies the nature of the result (document type prediction, topic, sentiment, summary, custom). This allows multiple independent classifications (type + topic + summary) on the same version without conflating them. ``label`` is the primary output (e.g. ``"contract"``, ``"finance"``). ``tags`` is a JSON list of secondary labels (e.g. auto-tags). ``summary`` is a free-text synopsis when ``classification_type=summary``. Confidence is stored as a decimal in [0, 1]. ExtractedEntity A named entity extracted from a document version, typically produced as a by-product of OCR or an AI NER pass. ``source`` optionally links back to the ``OCRResult`` that produced the entity so provenance is preserved. A NULL ``source`` means the entity was produced by a standalone NER step. ``entity_type`` uses a controlled vocabulary. ``position_metadata`` stores provider-specific location hints (page, bbox, char offsets) in a JSON blob so the schema stays thin while location data is preserved for future document highlighting features. Soft-deletable. Re-extraction overwrites via a service that soft-deletes old entities then bulk-creates new ones. Provider abstraction No provider logic lives in this bounded context. The ``provider`` field is a mere label written by whichever adapter submits the job. Future provider adapters must live outside DMS (e.g. in the AI app layer) and call the DMS service API. Async readiness The services accept and return model instances. Celery tasks will wrap them; the DMS layer itself does not import Celery directly. """ from __future__ import annotations from typing import ClassVar from django.conf import settings from django.db import models from django.utils.translation import gettext_lazy as _ from simorgh.core.db.scoped import ScopedSoftDeleteManager from simorgh.core.models import AuditedModel, SoftDeleteModel, TenantScopedModel, UUIDModel # --------------------------------------------------------------------------- # Choices # --------------------------------------------------------------------------- class OCRStatus(models.TextChoices): PENDING = "pending", _("Pending") PROCESSING = "processing", _("Processing") COMPLETED = "completed", _("Completed") FAILED = "failed", _("Failed") class AIClassificationType(models.TextChoices): DOCUMENT_TYPE = "document_type", _("Document Type") TOPIC = "topic", _("Topic") SENTIMENT = "sentiment", _("Sentiment") SUMMARY = "summary", _("Summary") CUSTOM = "custom", _("Custom") class EntityType(models.TextChoices): PERSON = "person", _("Person") ORGANIZATION = "organization", _("Organization") DATE = "date", _("Date") AMOUNT = "amount", _("Amount") LOCATION = "location", _("Location") REFERENCE = "reference", _("Reference Number") CUSTOM = "custom", _("Custom") # --------------------------------------------------------------------------- # OCRResult # --------------------------------------------------------------------------- class OCRResult(UUIDModel, TenantScopedModel, AuditedModel, SoftDeleteModel): """Tracks an OCR processing run on a specific document version. One version may have many OCR runs (different providers, retries). Use ``queries.get_latest_ocr_result`` to get the most recent completed run. """ document_version = models.ForeignKey( "dms.DocumentVersion", on_delete=models.CASCADE, related_name="ocr_results", verbose_name=_("document version"), ) provider = models.CharField( _("provider"), max_length=100, default="default", help_text=_("Identifier of the OCR provider (e.g. 'tesseract', 'azure_ocr')."), ) status = models.CharField( _("status"), max_length=20, choices=OCRStatus.choices, default=OCRStatus.PENDING, db_index=True, ) language_code = models.CharField( _("language code"), max_length=20, blank=True, default="", help_text=_("BCP-47 language tag detected or requested (e.g. 'en', 'fa')."), ) full_text = models.TextField( _("full text"), blank=True, default="", help_text=_("Concatenated raw OCR output across all pages."), ) confidence_score = models.DecimalField( _("confidence score"), max_digits=5, decimal_places=4, null=True, blank=True, help_text=_("Aggregate confidence score in the range [0.0000, 1.0000]."), ) page_count = models.PositiveIntegerField( _("page count"), default=0, ) processing_metadata = models.JSONField( _("processing metadata"), default=dict, blank=True, help_text=_("Provider-specific metadata (timing, warnings, per-page results)."), ) error_message = models.TextField( _("error message"), blank=True, default="", ) submitted_at = models.DateTimeField( _("submitted at"), auto_now_add=True, ) completed_at = models.DateTimeField( _("completed at"), null=True, blank=True, ) objects: ClassVar[ScopedSoftDeleteManager] = ScopedSoftDeleteManager() class Meta: verbose_name = _("OCR result") verbose_name_plural = _("OCR results") ordering = ("-submitted_at",) indexes = [ models.Index( fields=["document_version_id", "status"], name="dms_ocr_version_status_idx", ), models.Index( fields=["tenant_id", "provider"], name="dms_ocr_tenant_provider_idx", ), ] def __str__(self) -> str: return f"OCR({self.provider}) version={self.document_version_id} [{self.status}]" # --------------------------------------------------------------------------- # AIClassification # --------------------------------------------------------------------------- class AIClassification(UUIDModel, TenantScopedModel, AuditedModel, SoftDeleteModel): """AI classification result for a specific document version. Multiple classification records of different ``classification_type`` values may exist for the same version (e.g. one for document type, one for topic, one for summary). """ document_version = models.ForeignKey( "dms.DocumentVersion", on_delete=models.CASCADE, related_name="ai_classifications", verbose_name=_("document version"), ) provider = models.CharField( _("provider"), max_length=100, default="default", help_text=_("Identifier of the AI provider (e.g. 'openai', 'azure_ai')."), ) classification_type = models.CharField( _("classification type"), max_length=50, choices=AIClassificationType.choices, default=AIClassificationType.DOCUMENT_TYPE, db_index=True, ) label = models.CharField( _("label"), max_length=255, blank=True, default="", help_text=_("Primary classification output (e.g. 'contract', 'finance')."), ) confidence_score = models.DecimalField( _("confidence score"), max_digits=5, decimal_places=4, null=True, blank=True, help_text=_("Confidence score in the range [0.0000, 1.0000]."), ) summary = models.TextField( _("summary"), blank=True, default="", help_text=_("Free-text synopsis generated by the AI (used for summary type)."), ) tags = models.JSONField( _("tags"), default=list, blank=True, help_text=_("Auto-generated tag list from AI classification."), ) processing_metadata = models.JSONField( _("processing metadata"), default=dict, blank=True, ) objects: ClassVar[ScopedSoftDeleteManager] = ScopedSoftDeleteManager() class Meta: verbose_name = _("AI classification") verbose_name_plural = _("AI classifications") ordering = ("-created_at",) indexes = [ models.Index( fields=["document_version_id", "classification_type"], name="dms_ai_version_type_idx", ), models.Index( fields=["tenant_id", "provider"], name="dms_ai_tenant_prov_idx", ), ] def __str__(self) -> str: return ( f"AIClass({self.classification_type}) " f"label={self.label!r} version={self.document_version_id}" ) # --------------------------------------------------------------------------- # ExtractedEntity # --------------------------------------------------------------------------- class ExtractedEntity(UUIDModel, TenantScopedModel, AuditedModel, SoftDeleteModel): """A named entity extracted from a document version. Entities may come from an OCR pass (``source`` FK is set) or from a standalone NER step (``source`` is NULL). A single document version may have many extracted entities of different types. """ document_version = models.ForeignKey( "dms.DocumentVersion", on_delete=models.CASCADE, related_name="extracted_entities", verbose_name=_("document version"), ) source = models.ForeignKey( OCRResult, on_delete=models.SET_NULL, null=True, blank=True, related_name="entities", verbose_name=_("OCR source"), help_text=_("OCR run that produced this entity, if any."), ) entity_type = models.CharField( _("entity type"), max_length=50, choices=EntityType.choices, default=EntityType.CUSTOM, db_index=True, ) value = models.CharField( _("value"), max_length=500, help_text=_("The raw extracted entity value."), ) normalized_value = models.CharField( _("normalized value"), max_length=500, blank=True, default="", help_text=_("Standardized form of the value (e.g. ISO date, canonical name)."), ) confidence_score = models.DecimalField( _("confidence score"), max_digits=5, decimal_places=4, null=True, blank=True, ) provider = models.CharField( _("provider"), max_length=100, blank=True, default="", help_text=_("Provider that extracted this entity."), ) position_metadata = models.JSONField( _("position metadata"), default=dict, blank=True, help_text=_( "Location hints in the source document: page number, bounding box, " "character offsets. Format is provider-specific." ), ) objects: ClassVar[ScopedSoftDeleteManager] = ScopedSoftDeleteManager() class Meta: verbose_name = _("extracted entity") verbose_name_plural = _("extracted entities") ordering = ["entity_type", "value"] indexes = [ models.Index( fields=["document_version_id", "entity_type"], name="dms_entity_version_type_idx", ), models.Index( fields=["tenant_id", "entity_type"], name="dms_entity_tenant_type_idx", ), ] def __str__(self) -> str: return f"{self.entity_type}:{self.value!r} version={self.document_version_id}"