"""Process Data Catalog models — Phase 2 (EXT / OUT Items). Models that capture the standard inputs and outputs of organisational processes and the operational steps that produce or consume them. Model inventory --------------- ProcessDataItem — catalogued input/output item (EXT-xxx or OUT-xxx). ProcessIO — many-to-many link: process ↔ data item with direction. ProcessOperationalStep — ordered execution steps within a process. Design decisions ---------------- * ``ProcessDataItem`` belongs to a tenant (company-wide catalog) but is not org-node scoped — a data item can be referenced by multiple processes. * ``direction`` on ``ProcessDataItem`` marks the *default* direction the item travels (input into the process, or output from it). ``ProcessIO.direction`` lets a single item serve as input to one process and output of another. * ``pcf_activities`` on ``ProcessOperationalStep`` stores a JSON list of PCFElement hierarchy_ids (e.g. ["1.1.1.1", "1.1.1.2"]). This avoids a heavyweight M2M table while keeping the data human-readable. SQLite and PostgreSQL both support JSONField natively in Django 3.1+. * ``is_external`` on ``ProcessDataItem`` mirrors the EXT/OUT naming convention in APQC sample documents: EXT items originate from external sources (is_external=True); OUT items are internally produced (is_external=False). * ``code`` has a UniqueConstraint per tenant to allow the same code pattern ("EXT-001") across different tenants without collision. """ from __future__ import annotations from typing import ClassVar from django.db import models from django.utils.translation import gettext_lazy as _ from simorgh.core.models import TimeStampedModel, UUIDModel from simorgh.apps.bpm.models.process import ProcessDefinition # --------------------------------------------------------------------------- # Choices # --------------------------------------------------------------------------- class DataDirection(models.TextChoices): INPUT = "input", _("Input") OUTPUT = "output", _("Output") class IOLinkDirection(models.TextChoices): INPUT = "input", _("Input") OUTPUT = "output", _("Output") # --------------------------------------------------------------------------- # ProcessDataItem # --------------------------------------------------------------------------- class ProcessDataItem(UUIDModel, TimeStampedModel): """A catalogued data item that flows into or out of organisational processes. Codes follow APQC convention: * EXT-001, EXT-002, … — external inputs. * OUT-111-1, OUT-111-2, … — outputs of process 1.1.1. """ tenant = models.ForeignKey( "tenants.Tenant", on_delete=models.CASCADE, related_name="process_data_items", verbose_name=_("tenant"), ) # ------------------------------------------------------------------ # Identity # ------------------------------------------------------------------ code = models.CharField( _("code"), max_length=64, help_text=_('Short identifier, e.g. "EXT-001" or "OUT-111-1".'), ) name = models.CharField(_("name (EN)"), max_length=512) name_fa = models.CharField(_("name (FA)"), max_length=512, blank=True) description = models.TextField(_("description"), blank=True) # ------------------------------------------------------------------ # Classification # ------------------------------------------------------------------ direction = models.CharField( _("direction"), max_length=8, choices=DataDirection.choices, default=DataDirection.INPUT, db_index=True, help_text=_("Default direction: input (EXT) or output (OUT)."), ) is_external = models.BooleanField( _("is external"), default=True, help_text=_("True for EXT items (external sources); False for OUT items (internally produced)."), ) # ------------------------------------------------------------------ # Format / reference # ------------------------------------------------------------------ format = models.CharField( _("format"), max_length=128, blank=True, help_text=_('File/data format, e.g. "CSV/Excel", "PPT", "Excel + Word".'), ) key_fields = models.JSONField( _("key fields"), default=list, blank=True, help_text=_('List of key data fields, e.g. ["GDP", "نرخ بهره"].'), ) standard_reference = models.CharField( _("standard reference"), max_length=256, blank=True, help_text=_('Data source or standard, e.g. "IMF / بانک مرکزی".'), ) class Meta: verbose_name = _("process data item") verbose_name_plural = _("process data items") ordering = ["direction", "code"] constraints: ClassVar = [ models.UniqueConstraint( fields=["tenant", "code"], name="bpm_processdataitem_unique_tenant_code", ), ] indexes = [ models.Index(fields=["tenant", "direction"]), models.Index(fields=["tenant", "is_external"]), ] def __str__(self) -> str: return f"{self.code} — {self.name}" # --------------------------------------------------------------------------- # ProcessIO # --------------------------------------------------------------------------- class ProcessIO(TimeStampedModel): """Links a ProcessDefinition to a ProcessDataItem with a flow direction. A single data item can be an *input* to one process and an *output* of another; ``direction`` here overrides the default on ``ProcessDataItem``. """ process = models.ForeignKey( ProcessDefinition, on_delete=models.CASCADE, related_name="io_links", verbose_name=_("process"), ) data_item = models.ForeignKey( ProcessDataItem, on_delete=models.CASCADE, related_name="io_links", verbose_name=_("data item"), ) direction = models.CharField( _("direction"), max_length=8, choices=IOLinkDirection.choices, default=IOLinkDirection.INPUT, db_index=True, ) is_required = models.BooleanField( _("is required"), default=True, help_text=_("Whether this data item is mandatory for the process to start/complete."), ) notes = models.TextField(_("notes"), blank=True) class Meta: verbose_name = _("process I/O link") verbose_name_plural = _("process I/O links") ordering = ["direction", "data_item__code"] constraints: ClassVar = [ models.UniqueConstraint( fields=["process", "data_item", "direction"], name="bpm_processio_unique_process_item_direction", ), ] indexes = [ models.Index(fields=["process", "direction"]), ] def __str__(self) -> str: return f"{self.process.hierarchy_id} ← {self.direction} → {self.data_item.code}" # --------------------------------------------------------------------------- # ProcessOperationalStep # --------------------------------------------------------------------------- class ProcessOperationalStep(TimeStampedModel): """An ordered execution step within a ProcessDefinition. Corresponds to the rows in the "Operational Sequence" or "جدول الف" tables found in APQC sample process documents. ``pcf_activities`` stores a JSON list of PCFElement hierarchy_ids that are performed during this step (e.g. ``["1.1.1.1", "1.1.1.2"]``). """ process = models.ForeignKey( ProcessDefinition, on_delete=models.CASCADE, related_name="operational_steps", verbose_name=_("process"), ) step_number = models.PositiveSmallIntegerField( _("step number"), help_text=_("Sequential number within the process (1-based)."), ) title = models.CharField(_("title"), max_length=512) title_fa = models.CharField(_("title (FA)"), max_length=512, blank=True) description = models.TextField(_("description"), blank=True) output_description = models.TextField( _("output description"), blank=True, help_text=_("What this step produces or delivers."), ) pcf_activities = models.JSONField( _("PCF activities"), default=list, blank=True, help_text=_( "List of PCFElement hierarchy_ids mapped to this step, " 'e.g. ["1.1.1.1", "1.1.1.2"].' ), ) order = models.PositiveSmallIntegerField( _("order"), default=0, db_index=True, help_text=_("Display order (may differ from step_number for re-ordering without renumbering)."), ) class Meta: verbose_name = _("process operational step") verbose_name_plural = _("process operational steps") ordering = ["process", "order", "step_number"] constraints: ClassVar = [ models.UniqueConstraint( fields=["process", "step_number"], name="bpm_processoperationalstep_unique_process_step", ), ] indexes = [ models.Index(fields=["process", "order"]), ] def __str__(self) -> str: return f"{self.process.hierarchy_id} / Step {self.step_number}: {self.title}"