"""Services for the platform_core generic models. These are the **only** entry points modules should call — they enforce tenant scoping, audit logging, and event dispatch. Direct ORM access is fine for reads but writes must go through here. """ from __future__ import annotations import re from typing import TYPE_CHECKING, Any from django.contrib.contenttypes.models import ContentType from django.db import transaction from django.utils import timezone from simorgh.apps.events.bus import dispatch, subscribe from simorgh.apps.platform_core.models import ( Activity, Attachment, AttachmentKind, Comment, CustomFieldDefinition, CustomFieldType, CustomFieldValue, DocumentTemplate, ExportJob, ExportJobStatus, ExportLayout, FormDefinition, FormFieldType, FormSubmission, ImportJob, ImportJobStatus, MessageChannel, MessageTemplate, Mention, Note, NoteVisibility, PageOrientation, PaperSize, PrintTemplate, ReferenceNumberSequence, Tag, TagAssignment, ) from simorgh.core.audit import record_service_event from simorgh.core.context import current_request_context from simorgh.core.exceptions import PlatformError if TYPE_CHECKING: from django.contrib.auth.models import AbstractUser from django.db.models import Model from simorgh.apps.storage.models import FileMetadata class PlatformCoreError(PlatformError): """Raised by platform_core services on contract violations.""" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _entity_ref(entity: Model) -> tuple[ContentType, str, int]: """Return (content_type, object_id_str, tenant_id) for ``entity``.""" if not hasattr(entity, "pk") or entity.pk is None: raise PlatformCoreError("entity must be a saved model instance") tenant_id = getattr(entity, "tenant_id", None) if tenant_id is None: raise PlatformCoreError( f"{entity._meta.label} is not tenant-scoped — " "cannot attach platform_core resources to it", ) ct = ContentType.objects.get_for_model(entity.__class__) return ct, str(entity.pk), tenant_id def _resolve_actor(actor: AbstractUser | None) -> AbstractUser | None: if actor is not None: return actor ctx = current_request_context() return ctx.actor def _entity_type_label(entity: Model) -> str: return f"{entity._meta.app_label}.{entity._meta.model_name}" # --------------------------------------------------------------------------- # Attachments # --------------------------------------------------------------------------- def attach_file( entity: Model, file: FileMetadata, *, kind: str = AttachmentKind.DOCUMENT, description: str = "", sort_order: int = 0, uploaded_by: AbstractUser | None = None, ) -> Attachment: ct, object_id, tenant_id = _entity_ref(entity) if file.tenant_id != tenant_id: raise PlatformCoreError("attachment file belongs to a different tenant") organization_node_id = getattr(entity, "organization_node_id", None) if organization_node_id is None: raise PlatformCoreError( f"{entity._meta.label} has no organization_node — cannot attach files", ) actor = _resolve_actor(uploaded_by) with transaction.atomic(): att = Attachment.objects.create( tenant_id=tenant_id, organization_node_id=organization_node_id, content_type=ct, object_id=object_id, file=file, kind=kind, description=description, sort_order=sort_order, uploaded_by=actor, ) record_service_event( "platform_core.attachment_added", resource=att, after={"file_id": str(file.pk), "kind": kind}, ) dispatch( "core.attachment_added", { "tenant_id": tenant_id, "entity_type": _entity_type_label(entity), "entity_id": object_id, "attachment_id": str(att.pk), }, ) record_activity( verb="attached", entity=entity, actor=actor, extra={"attachment_id": str(att.pk), "kind": kind}, ) return att def remove_attachment(attachment: Attachment, *, actor: AbstractUser | None = None) -> None: actor = _resolve_actor(actor) actor_id = actor.pk if actor else None tenant_id = attachment.tenant_id ct_label = f"{attachment.content_type.app_label}.{attachment.content_type.model}" object_id = attachment.object_id attachment.delete(actor_id=actor_id) record_service_event( "platform_core.attachment_removed", resource=attachment, before={"file_id": str(attachment.file_id)}, ) dispatch( "core.attachment_removed", { "tenant_id": tenant_id, "entity_type": ct_label, "entity_id": object_id, "attachment_id": str(attachment.pk), }, ) def list_attachments(entity: Model) -> list[Attachment]: ct, object_id, _tenant_id = _entity_ref(entity) return list( Attachment.objects.filter(content_type=ct, object_id=object_id).select_related("file"), ) # --------------------------------------------------------------------------- # Comments # --------------------------------------------------------------------------- def post_comment( entity: Model, *, body: str, author: AbstractUser | None = None, parent: Comment | None = None, mentions: list[AbstractUser] | None = None, ) -> Comment: if not body or not body.strip(): raise PlatformCoreError("comment body cannot be empty") ct, object_id, tenant_id = _entity_ref(entity) organization_node_id = getattr(entity, "organization_node_id", None) if organization_node_id is None: raise PlatformCoreError( f"{entity._meta.label} has no organization_node — cannot comment on it", ) actor = _resolve_actor(author) if parent is not None and parent.tenant_id != tenant_id: raise PlatformCoreError("parent comment belongs to a different tenant") with transaction.atomic(): comment = Comment.objects.create( tenant_id=tenant_id, organization_node_id=organization_node_id, content_type=ct, object_id=object_id, author=actor, body=body, parent=parent, ) if mentions: comment.mentions.set(mentions) _process_mentions(body, tenant_id, organization_node_id, comment=comment, author=actor) record_service_event("platform_core.comment_added", resource=comment, after={"body": body}) dispatch( "core.comment_added", { "tenant_id": tenant_id, "entity_type": _entity_type_label(entity), "entity_id": object_id, "comment_id": str(comment.pk), }, ) record_activity( verb="commented", entity=entity, actor=actor, extra={"comment_id": str(comment.pk)}, ) return comment def edit_comment(comment: Comment, *, body: str, actor: AbstractUser | None = None) -> Comment: if not body or not body.strip(): raise PlatformCoreError("comment body cannot be empty") actor = _resolve_actor(actor) before = {"body": comment.body} comment.body = body comment.edited_at = timezone.now() comment.save(update_fields=["body", "edited_at", "updated_at"]) _process_mentions( body, comment.tenant_id, comment.organization_node_id, comment=comment, author=actor, ) record_service_event( "platform_core.comment_updated", resource=comment, before=before, after={"body": body}, ) dispatch( "core.comment_updated", {"tenant_id": comment.tenant_id, "comment_id": str(comment.pk)}, ) return comment def delete_comment(comment: Comment, *, actor: AbstractUser | None = None) -> None: actor = _resolve_actor(actor) actor_id = actor.pk if actor else None comment.delete(actor_id=actor_id) record_service_event( "platform_core.comment_deleted", resource=comment, before={"body": comment.body}, ) dispatch( "core.comment_deleted", {"tenant_id": comment.tenant_id, "comment_id": str(comment.pk)}, ) def list_comments(entity: Model, *, include_deleted: bool = False) -> list[Comment]: ct, object_id, _tenant_id = _entity_ref(entity) manager = Comment.objects.with_deleted() if include_deleted else Comment.objects return list( manager.filter(content_type=ct, object_id=object_id) .select_related("author") .order_by("created_at"), ) # --------------------------------------------------------------------------- # Activity feed # --------------------------------------------------------------------------- def record_activity( *, verb: str, entity: Model | None = None, actor: AbstractUser | None = None, target: Model | None = None, extra: dict[str, Any] | None = None, occurred_at: Any = None, ) -> Activity: ctx = current_request_context() actor = _resolve_actor(actor) tenant_id: int | None = None organization_node_id: int | None = None ct = None object_id = "" if entity is not None: ct, object_id, tenant_id = _entity_ref(entity) organization_node_id = getattr(entity, "organization_node_id", None) if tenant_id is None and ctx.tenant is not None: tenant_id = ctx.tenant.pk if organization_node_id is None and ctx.org_node_ids: organization_node_id = next(iter(ctx.org_node_ids)) if tenant_id is None or organization_node_id is None: raise PlatformCoreError( "record_activity requires either an entity or an active tenant/org context", ) target_ct = None target_id = "" if target is not None: target_ct = ContentType.objects.get_for_model(target.__class__) target_id = str(target.pk) activity = Activity.objects.create( tenant_id=tenant_id, organization_node_id=organization_node_id, actor=actor, verb=verb, content_type=ct, object_id=object_id, target_content_type=target_ct, target_object_id=target_id, occurred_at=occurred_at or timezone.now(), extra=extra or {}, ) dispatch( "core.activity_recorded", { "tenant_id": tenant_id, "actor_id": actor.pk if actor else None, "verb": verb, "entity_type": _entity_type_label(entity) if entity else "", "entity_id": object_id, }, audit=False, # activity itself is the audit-visible artifact ) return activity def activity_for_entity(entity: Model, *, limit: int = 50) -> list[Activity]: ct, object_id, _tenant_id = _entity_ref(entity) return list( Activity.objects.filter(content_type=ct, object_id=object_id) .select_related("actor")[:limit], ) def activity_for_actor(actor: AbstractUser, *, limit: int = 50) -> list[Activity]: return list(Activity.objects.filter(actor=actor)[:limit]) # --------------------------------------------------------------------------- # Custom fields # --------------------------------------------------------------------------- def define_custom_field( *, tenant_id: int, organization_node_id: int, entity_type: str, key: str, label_key: str, field_type: str, is_required: bool = False, sort_order: int = 0, validation: dict[str, Any] | None = None, options: list[Any] | None = None, help_text: str = "", ) -> CustomFieldDefinition: if field_type not in CustomFieldType.values: raise PlatformCoreError(f"unknown custom field type: {field_type!r}") cfd = CustomFieldDefinition.objects.create( tenant_id=tenant_id, organization_node_id=organization_node_id, entity_type=entity_type, key=key, label_key=label_key, field_type=field_type, is_required=is_required, sort_order=sort_order, validation=validation or {}, options=options or [], help_text=help_text, ) record_service_event("platform_core.custom_field_defined", resource=cfd) dispatch( "core.custom_field_defined", {"tenant_id": tenant_id, "entity_type": entity_type, "definition_id": str(cfd.pk)}, ) return cfd def set_custom_field_value( entity: Model, *, definition: CustomFieldDefinition | str, value: Any, ) -> CustomFieldValue: ct, object_id, tenant_id = _entity_ref(entity) if isinstance(definition, str): try: definition = CustomFieldDefinition.objects.get( tenant_id=tenant_id, entity_type=_entity_type_label(entity), key=definition, ) except CustomFieldDefinition.DoesNotExist as exc: raise PlatformCoreError( f"no custom field {definition!r} defined for {_entity_type_label(entity)}", ) from exc if definition.tenant_id != tenant_id: raise PlatformCoreError("custom field belongs to a different tenant") organization_node_id = getattr(entity, "organization_node_id", None) if organization_node_id is None: raise PlatformCoreError( f"{entity._meta.label} has no organization_node — cannot set custom fields", ) cfv, _created = CustomFieldValue.objects.update_or_create( definition=definition, content_type=ct, object_id=object_id, defaults={ "tenant_id": tenant_id, "organization_node_id": organization_node_id, "value": value, }, ) dispatch( "core.custom_field_value_changed", { "tenant_id": tenant_id, "definition_id": str(definition.pk), "entity_type": _entity_type_label(entity), "entity_id": object_id, }, ) return cfv def get_custom_field_values(entity: Model) -> dict[str, Any]: ct, object_id, tenant_id = _entity_ref(entity) rows = CustomFieldValue.objects.filter( tenant_id=tenant_id, content_type=ct, object_id=object_id, ).select_related("definition") return {row.definition.key: row.value for row in rows} # --------------------------------------------------------------------------- # Activity auto-subscription to lifecycle events # --------------------------------------------------------------------------- # Subscribers ensure that key events on other modules are recorded in the # activity feed without each module having to know about Activity. @subscribe("core.comment_added") def _record_comment_added(payload: dict[str, Any]) -> None: # Already covered by post_comment() — kept as a no-op subscription so the # event surface is observable from tests and future plugins. return __all__ = [ "PlatformCoreError", "activity_for_actor", "activity_for_entity", "attach_file", "define_custom_field", "delete_comment", "edit_comment", "get_custom_field_values", "list_attachments", "list_comments", "post_comment", "record_activity", "remove_attachment", "set_custom_field_value", # Tags "create_tag", "delete_tag", "add_tags", "remove_tags", "set_tags", # Notes "create_note", "edit_note", "delete_note", # Mentions "extract_mentions", # Reference numbers "get_next_reference", "ensure_reference_sequence", # Dynamic forms "create_form_definition", "update_form_definition", "delete_form_definition", "submit_form", "get_form_definition_by_slug", "validate_form_data", # Import engine "create_import_job", "enqueue_import_job", "cancel_import_job", # Export engine "create_export_job", "cancel_export_job", # Print engine "create_print_template", "update_print_template", "delete_print_template", "render_pdf", "render_html_preview", # Template engine "create_message_template", "update_message_template", "delete_message_template", "render_template", "render_template_subject", # Document template engine "create_document_template", "update_document_template", "delete_document_template", "render_document", "render_document_pdf", # Export layout engine "create_export_layout", "update_export_layout", "delete_export_layout", ] # --------------------------------------------------------------------------- # Mentions (@user) # --------------------------------------------------------------------------- _MENTION_PATTERN = re.compile(r"@([\w][\w.\-]*)") def extract_mentions(body: str, tenant_id: int) -> list[Any]: """Parse ``@handle`` patterns from *body* and return matching active users. The handle is matched against the ``username`` (display-name) field. Only users who are active members of *tenant_id* are returned. The result may contain multiple users that share the same display name. """ from django.contrib.auth import get_user_model handles = {m.lower() for m in _MENTION_PATTERN.findall(body)} if not handles: return [] User = get_user_model() return list( User.objects.filter( memberships__tenant_id=tenant_id, memberships__status="active", is_active=True, username__in=handles, ).distinct() ) def _process_mentions( body: str, tenant_id: int, organization_node_id: int, *, comment: Comment | None = None, note: Note | None = None, author: Any = None, ) -> None: """Create :class:`Mention` records for new @-mentions and dispatch notifications.""" mentioned_users = extract_mentions(body, tenant_id) if not mentioned_users: return author_id = author.pk if author else None new_users = [u for u in mentioned_users if u.pk != author_id] if not new_users: return # Exclude users already recorded for this source if comment is not None: existing_ids = set( Mention.objects.filter(comment=comment).values_list("user_id", flat=True) ) else: existing_ids = set( Mention.objects.filter(note=note).values_list("user_id", flat=True) ) new_users = [u for u in new_users if u.pk not in existing_ids] if not new_users: return with transaction.atomic(): created = [ Mention.objects.create( tenant_id=tenant_id, organization_node_id=organization_node_id, comment=comment, note=note, user=u, notified=False, ) for u in new_users ] # Best-effort: dispatch notifications; never raise try: from simorgh.apps.notifications.services import ( dispatch as _notify, ) kind = "core.mention_in_comment" if comment else "core.mention_in_note" ctx: dict[str, Any] = {"author": str(author) if author else ""} if comment is not None: ctx["comment_id"] = str(comment.pk) else: ctx["note_id"] = str(note.pk) # type: ignore[union-attr] _notify( kind, recipients=new_users, context=ctx, tenant_id=tenant_id, organization_node_id=organization_node_id, ) Mention.objects.filter(pk__in=[m.pk for m in created]).update(notified=True) except Exception: # noqa: BLE001 pass dispatch( "core.mention_created", { "tenant_id": tenant_id, "user_ids": [str(u.pk) for u in new_users], "comment_id": str(comment.pk) if comment else None, "note_id": str(note.pk) if note else None, }, ) # --------------------------------------------------------------------------- # Tags # --------------------------------------------------------------------------- def create_tag( *, tenant_id: int, organization_node_id: int, name: str, slug: str, color: str = "#6B7280", entity_type: str = "", created_by: AbstractUser | None = None, ) -> Tag: """Create a new platform-level tag for *tenant_id*.""" from django.utils.text import slugify slug = slug or slugify(name) if Tag.objects.filter(tenant_id=tenant_id, slug=slug).exists(): raise PlatformCoreError(f"a tag with slug {slug!r} already exists for this tenant") tag = Tag.objects.create( tenant_id=tenant_id, organization_node_id=organization_node_id, name=name, slug=slug, color=color, entity_type=entity_type, created_by=created_by, ) record_service_event("platform_core.tag_created", resource=tag, after={"name": name}) dispatch( "core.tag_created", {"tenant_id": tenant_id, "tag_id": str(tag.pk), "slug": slug}, ) return tag def delete_tag(tag: Tag, *, actor: AbstractUser | None = None) -> None: """Delete a tag and all its assignments.""" actor = _resolve_actor(actor) actor_id = actor.pk if actor else None record_service_event( "platform_core.tag_deleted", resource=tag, before={"name": tag.name, "slug": tag.slug}, ) dispatch("core.tag_deleted", {"tenant_id": tag.tenant_id, "tag_id": str(tag.pk)}) tag.delete() def add_tags( entity: Model, tag_ids: list[int], *, actor: AbstractUser | None = None, ) -> list[TagAssignment]: """Attach one or more tags to *entity*. Tags that are already assigned are silently ignored (idempotent). """ if not tag_ids: return [] ct, object_id, tenant_id = _entity_ref(entity) organization_node_id = getattr(entity, "organization_node_id", None) if organization_node_id is None: raise PlatformCoreError( f"{entity._meta.label} has no organization_node — cannot assign tags", ) actor = _resolve_actor(actor) tags = list(Tag.objects.filter(pk__in=tag_ids, tenant_id=tenant_id)) if len(tags) != len(set(tag_ids)): raise PlatformCoreError("one or more tag IDs not found or belong to a different tenant") assignments: list[TagAssignment] = [] with transaction.atomic(): for tag in tags: assignment, created = TagAssignment.objects.get_or_create( tag=tag, content_type=ct, object_id=object_id, defaults={ "tenant_id": tenant_id, "organization_node_id": organization_node_id, "assigned_by": actor, }, ) assignments.append(assignment) if created: record_activity( verb="tag_added", entity=entity, actor=actor, extra={"tag_id": str(tag.pk), "tag_name": tag.name}, ) dispatch( "core.tags_added", { "tenant_id": tenant_id, "entity_type": _entity_type_label(entity), "entity_id": object_id, "tag_ids": [str(t.pk) for t in tags], }, ) return assignments def remove_tags( entity: Model, tag_ids: list[int], *, actor: AbstractUser | None = None, ) -> None: """Detach one or more tags from *entity*. Tags that are not currently assigned are silently ignored. """ if not tag_ids: return ct, object_id, tenant_id = _entity_ref(entity) actor = _resolve_actor(actor) with transaction.atomic(): deleted_count, _ = TagAssignment.objects.filter( tenant_id=tenant_id, tag_id__in=tag_ids, content_type=ct, object_id=object_id, ).delete() if deleted_count: record_activity( verb="tags_removed", entity=entity, actor=actor, extra={"tag_ids": [str(t) for t in tag_ids]}, ) dispatch( "core.tags_removed", { "tenant_id": tenant_id, "entity_type": _entity_type_label(entity), "entity_id": object_id, "tag_ids": [str(t) for t in tag_ids], }, ) def set_tags( entity: Model, tag_ids: list[int], *, actor: AbstractUser | None = None, ) -> list[TagAssignment]: """Replace all tags on *entity* with exactly *tag_ids*. This is an atomic operation: existing assignments not in *tag_ids* are removed, new assignments are created, unchanged assignments are kept. """ ct, object_id, tenant_id = _entity_ref(entity) organization_node_id = getattr(entity, "organization_node_id", None) if organization_node_id is None: raise PlatformCoreError( f"{entity._meta.label} has no organization_node — cannot set tags", ) actor = _resolve_actor(actor) if tag_ids: tags = list(Tag.objects.filter(pk__in=tag_ids, tenant_id=tenant_id)) if len(tags) != len(set(tag_ids)): raise PlatformCoreError( "one or more tag IDs not found or belong to a different tenant", ) else: tags = [] with transaction.atomic(): # Remove tags that are no longer in the set. if tags: TagAssignment.objects.filter( tenant_id=tenant_id, content_type=ct, object_id=object_id, ).exclude(tag_id__in=tag_ids).delete() else: TagAssignment.objects.filter( tenant_id=tenant_id, content_type=ct, object_id=object_id, ).delete() # Add any missing assignments. assignments: list[TagAssignment] = [] for tag in tags: assignment, _ = TagAssignment.objects.get_or_create( tag=tag, content_type=ct, object_id=object_id, defaults={ "tenant_id": tenant_id, "organization_node_id": organization_node_id, "assigned_by": actor, }, ) assignments.append(assignment) dispatch( "core.tags_set", { "tenant_id": tenant_id, "entity_type": _entity_type_label(entity), "entity_id": object_id, "tag_ids": [str(t.pk) for t in tags], }, ) return assignments # --------------------------------------------------------------------------- # Notes # --------------------------------------------------------------------------- def create_note( entity: Model, body: str, *, visibility: str = NoteVisibility.PRIVATE, author: AbstractUser | None = None, ) -> Note: """Create a note attached to *entity*. Args: entity: Any saved :class:`TenantScopedModel` instance. body: Note text (non-empty). visibility: One of ``"private"``, ``"team"``, ``"workspace"``. author: User creating the note; falls back to request context actor. """ if not body or not body.strip(): raise PlatformCoreError("note body cannot be empty") if visibility not in NoteVisibility.values: raise PlatformCoreError( f"invalid visibility {visibility!r}; must be one of {NoteVisibility.values}" ) ct, object_id, tenant_id = _entity_ref(entity) organization_node_id = getattr(entity, "organization_node_id", None) if organization_node_id is None: raise PlatformCoreError( f"{entity._meta.label} has no organization_node — cannot add notes", ) actor = _resolve_actor(author) note = Note.objects.create( tenant_id=tenant_id, organization_node_id=organization_node_id, content_type=ct, object_id=object_id, body=body, visibility=visibility, author=actor, ) _process_mentions(body, tenant_id, organization_node_id, note=note, author=actor) record_service_event( "platform_core.note_created", resource=note, after={"visibility": visibility}, ) dispatch( "core.note_created", { "tenant_id": tenant_id, "entity_type": _entity_type_label(entity), "entity_id": object_id, "note_id": str(note.pk), }, ) return note def edit_note( note: Note, *, body: str | None = None, is_pinned: bool | None = None, actor: AbstractUser | None = None, ) -> Note: """Edit the body and/or pin state of *note*. At least one of *body* or *is_pinned* must be provided. """ if body is None and is_pinned is None: raise PlatformCoreError("nothing to update — provide body or is_pinned") actor = _resolve_actor(actor) if body is not None: if not body.strip(): raise PlatformCoreError("note body cannot be empty") note.body = body note.edited_at = timezone.now() if is_pinned is not None: note.is_pinned = is_pinned note.save() if body is not None: _process_mentions( body, note.tenant_id, note.organization_node_id, note=note, author=actor, ) record_service_event( "platform_core.note_edited", resource=note, after={"body_changed": body is not None, "is_pinned": note.is_pinned}, ) dispatch( "core.note_edited", {"tenant_id": note.tenant_id, "note_id": str(note.pk)}, ) return note def delete_note(note: Note, *, actor: AbstractUser | None = None) -> None: """Hard-delete *note*.""" actor = _resolve_actor(actor) record_service_event( "platform_core.note_deleted", resource=note, before={"visibility": note.visibility}, ) dispatch( "core.note_deleted", {"tenant_id": note.tenant_id, "note_id": str(note.pk)}, ) note.delete() # --------------------------------------------------------------------------- # Reference Number Sequences # --------------------------------------------------------------------------- def _jalali_year(gregorian_year: int, gregorian_month: int, gregorian_day: int) -> int: """Return an approximate Jalali (Persian/Shamsi) year. Nowruz (Persian New Year) falls on or around March 20–21. This simple approximation is accurate for the purposes of reference-number generation: dates from March 21 onward belong to the new Jalali year. """ if gregorian_month > 3 or (gregorian_month == 3 and gregorian_day >= 21): return gregorian_year - 621 return gregorian_year - 622 def get_next_reference(tenant_id: int, entity_type: str) -> str: """Atomically increment and return the next reference number for *entity_type*. Uses ``SELECT FOR UPDATE`` so concurrent calls never produce duplicate reference numbers. Raises :class:`PlatformCoreError` if no active sequence is configured for this (tenant, entity_type) combination. Args: tenant_id: The tenant PK. entity_type: Dotted ``app_label.model_name``, e.g. ``"helpdesk.ticket"``. Returns: A formatted reference string, e.g. ``"TK-0001"`` or ``"TK-2025-0001"``. """ from django.utils import timezone as tz with transaction.atomic(): try: seq: ReferenceNumberSequence = ( ReferenceNumberSequence.objects.select_for_update() .get(tenant_id=tenant_id, entity_type=entity_type, is_active=True) ) except ReferenceNumberSequence.DoesNotExist: raise PlatformCoreError( f"no active reference-number sequence configured for " f"entity_type={entity_type!r} in tenant {tenant_id}" ) now = tz.localtime(tz.now()) if seq.use_jalali_year: current_year = _jalali_year(now.year, now.month, now.day) else: current_year = now.year # Reset counter at year boundary when requested. update_fields = ["last_number", "updated_at"] if seq.reset_yearly and seq.last_reset_year != current_year: seq.last_number = 0 seq.last_reset_year = current_year update_fields.append("last_reset_year") seq.last_number += 1 seq.save(update_fields=update_fields) try: reference = seq.format.format( prefix=seq.prefix, year=current_year, seq=seq.last_number, ) except (KeyError, ValueError) as exc: raise PlatformCoreError( f"invalid reference number format {seq.format!r}: {exc}" ) from exc dispatch( "core.reference_number_generated", { "tenant_id": tenant_id, "entity_type": entity_type, "reference": reference, }, ) return reference def ensure_reference_sequence( *, tenant_id: int, organization_node_id: int, entity_type: str, prefix: str, format: str = "{prefix}-{seq:04d}", # noqa: A002 reset_yearly: bool = False, use_jalali_year: bool = False, ) -> ReferenceNumberSequence: """Idempotently create or retrieve the sequence for *entity_type*. Safe to call multiple times (e.g. from data migrations or seed scripts). """ seq, created = ReferenceNumberSequence.objects.get_or_create( tenant_id=tenant_id, entity_type=entity_type, defaults={ "organization_node_id": organization_node_id, "prefix": prefix, "format": format, "reset_yearly": reset_yearly, "use_jalali_year": use_jalali_year, "is_active": True, }, ) if created: dispatch( "core.reference_sequence_created", { "tenant_id": tenant_id, "entity_type": entity_type, "prefix": prefix, "sequence_id": str(seq.pk), }, ) return seq # --------------------------------------------------------------------------- # Dynamic Forms # --------------------------------------------------------------------------- def validate_form_data( form: FormDefinition, data: dict[str, Any], ) -> dict[str, list[str]]: """Validate *data* against *form* field definitions. Returns a mapping of ``field_id → list[error_message]``. An empty dict means the data is valid. """ errors: dict[str, list[str]] = {} for field in form.fields: field_id = field.get("id", "") required = field.get("required", False) value = data.get(field_id) field_errors: list[str] = [] if required and (value is None or value == "" or value == []): field_errors.append("This field is required.") if value is not None and value != "": field_type = field.get("type", "") validation = field.get("validation") or {} max_length = validation.get("max_length") min_value = validation.get("min_value") max_value = validation.get("max_value") if field_type in (FormFieldType.TEXT, FormFieldType.TEXTAREA, FormFieldType.EMAIL, FormFieldType.PHONE, FormFieldType.URL): if max_length and len(str(value)) > max_length: field_errors.append(f"Value exceeds maximum length of {max_length}.") if field_type == FormFieldType.EMAIL: import re as _re if not _re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", str(value)): field_errors.append("Enter a valid email address.") elif field_type == FormFieldType.NUMBER: try: num = float(value) if min_value is not None and num < float(min_value): field_errors.append(f"Value must be at least {min_value}.") if max_value is not None and num > float(max_value): field_errors.append(f"Value must be at most {max_value}.") except (TypeError, ValueError): field_errors.append("A valid number is required.") elif field_type == FormFieldType.RATING: try: rating = int(value) if not (1 <= rating <= (validation.get("max") or 5)): field_errors.append("Rating is out of allowed range.") except (TypeError, ValueError): field_errors.append("A valid integer rating is required.") if field_errors: errors[field_id] = field_errors return errors def create_form_definition( *, tenant_id: int, organization_node_id: int, title: str, description: str = "", entity_type: str = "", fields: list[dict[str, Any]] | None = None, submit_action: str = "notify_only", submit_config: dict[str, Any] | None = None, is_active: bool = True, public_url_slug: str = "", created_by: AbstractUser | None = None, ) -> FormDefinition: """Create a new :class:`FormDefinition` for *tenant_id*.""" if not title or not title.strip(): raise PlatformCoreError("form title cannot be empty") if public_url_slug: if FormDefinition.objects.filter( tenant_id=tenant_id, public_url_slug=public_url_slug ).exists(): raise PlatformCoreError( f"a form with slug {public_url_slug!r} already exists for this tenant" ) actor = _resolve_actor(created_by) form = FormDefinition.objects.create( tenant_id=tenant_id, organization_node_id=organization_node_id, title=title, description=description, entity_type=entity_type, fields=fields or [], submit_action=submit_action, submit_config=submit_config or {}, is_active=is_active, public_url_slug=public_url_slug, created_by=actor, ) record_service_event("platform_core.form_created", resource=form, after={"title": title}) dispatch( "core.form_definition_created", { "tenant_id": tenant_id, "form_id": str(form.pk), "title": title, }, ) return form def update_form_definition( form: FormDefinition, *, title: str | None = None, description: str | None = None, fields: list[dict[str, Any]] | None = None, submit_action: str | None = None, submit_config: dict[str, Any] | None = None, is_active: bool | None = None, public_url_slug: str | None = None, actor: AbstractUser | None = None, ) -> FormDefinition: """Partial-update *form* with only the supplied keyword arguments.""" actor = _resolve_actor(actor) update_fields: list[str] = ["updated_at"] if title is not None: if not title.strip(): raise PlatformCoreError("form title cannot be empty") form.title = title update_fields.append("title") if description is not None: form.description = description update_fields.append("description") if fields is not None: form.fields = fields update_fields.append("fields") if submit_action is not None: form.submit_action = submit_action update_fields.append("submit_action") if submit_config is not None: form.submit_config = submit_config update_fields.append("submit_config") if is_active is not None: form.is_active = is_active update_fields.append("is_active") if public_url_slug is not None: if public_url_slug and public_url_slug != form.public_url_slug: if FormDefinition.objects.filter( tenant_id=form.tenant_id, public_url_slug=public_url_slug ).exclude(pk=form.pk).exists(): raise PlatformCoreError( f"a form with slug {public_url_slug!r} already exists for this tenant" ) form.public_url_slug = public_url_slug update_fields.append("public_url_slug") form.save(update_fields=update_fields) record_service_event("platform_core.form_updated", resource=form) dispatch("core.form_definition_updated", {"tenant_id": form.tenant_id, "form_id": str(form.pk)}) return form def delete_form_definition( form: FormDefinition, *, actor: AbstractUser | None = None, ) -> None: """Delete *form* (and cascade-delete its submissions via DB).""" actor = _resolve_actor(actor) record_service_event( "platform_core.form_deleted", resource=form, before={"title": form.title}, ) dispatch("core.form_definition_deleted", {"tenant_id": form.tenant_id, "form_id": str(form.pk)}) form.delete() def get_form_definition_by_slug(slug: str, *, tenant_id: int) -> FormDefinition: """Return the active :class:`FormDefinition` with *slug* for *tenant_id*. Raises :class:`PlatformCoreError` if not found or inactive. """ try: return FormDefinition.objects.get( tenant_id=tenant_id, public_url_slug=slug, is_active=True, ) except FormDefinition.DoesNotExist: raise PlatformCoreError( f"no active public form found with slug {slug!r} for this tenant" ) def submit_form( form: FormDefinition, data: dict[str, Any], *, user: AbstractUser | None = None, ip: str | None = None, organization_node_id: int | None = None, ) -> FormSubmission: """Validate *data* and create a :class:`FormSubmission`. Args: form: The :class:`FormDefinition` to submit. data: Key-value mapping of ``field_id → value``. user: The authenticated submitter; ``None`` for public. ip: IP address of the submitter (for audit). organization_node_id: Organisation node; falls back to the form's own node. Raises: :class:`PlatformCoreError` if *form* is inactive or validation fails. """ if not form.is_active: raise PlatformCoreError("this form is no longer accepting submissions") validation_errors = validate_form_data(form, data) if validation_errors: raise PlatformCoreError( f"form validation failed: {validation_errors}", ) org_node_id = organization_node_id or form.organization_node_id with transaction.atomic(): submission = FormSubmission.objects.create( tenant_id=form.tenant_id, organization_node_id=org_node_id, form=form, data=data, submitted_by=user, ip_address=ip, processed=False, ) record_service_event( "platform_core.form_submitted", resource=submission, after={"form_id": str(form.pk)}, ) dispatch( "core.form_submitted", { "tenant_id": form.tenant_id, "form_id": str(form.pk), "submission_id": str(submission.pk), "submit_action": form.submit_action, }, ) # Execute submit action best-effort so a failing action never blocks # the submission itself from being persisted. try: _execute_submit_action(form, submission, user=user) FormSubmission.objects.filter(pk=submission.pk).update(processed=True) submission.processed = True except Exception: # noqa: BLE001 pass return submission def _execute_submit_action( form: FormDefinition, submission: FormSubmission, *, user: AbstractUser | None = None, ) -> None: """Run the configured ``submit_action`` after a successful submission. ``notify_only`` is handled inline; other actions are best-effort hooks. """ action = form.submit_action cfg = form.submit_config or {} if action == "notify_only": # Event already dispatched by submit_form — nothing more to do. return if action == "webhook": url = cfg.get("url", "") if url: import json import urllib.request payload = json.dumps({ "form_id": str(form.pk), "submission_id": str(submission.pk), "data": submission.data, }).encode() req = urllib.request.Request( url, data=payload, headers={"Content-Type": "application/json"}, method="POST", ) urllib.request.urlopen(req, timeout=10) # noqa: S310 return # Future actions (create_ticket, create_lead) are registered via the # event bus; other modules can subscribe to "core.form_submitted" and # react accordingly based on submit_action value. # --------------------------------------------------------------------------- # Import Engine # --------------------------------------------------------------------------- def create_import_job( *, tenant_id: int, organization_node_id: int, entity_type: str, file_id: int, actor: AbstractUser | None = None, enqueue: bool = True, ) -> ImportJob: """Create an :class:`ImportJob` and optionally enqueue it for processing. Args: tenant_id: The owning tenant. organization_node_id: The org node for the job. entity_type: Dot-notation entity type, e.g. ``"crm.lead"``. file_id: PK of the uploaded :class:`~storage.FileMetadata`. actor: User who initiated the import. enqueue: If *True* (default), dispatch the Celery task immediately after creating the job. Raises: :class:`PlatformCoreError` if *entity_type* has no registered spec. """ from simorgh.apps.platform_core.import_registry import import_registry actor = _resolve_actor(actor) if entity_type not in import_registry: raise PlatformCoreError( f"No importer registered for entity type {entity_type!r}. " f"Available: {import_registry.all_entity_types()}" ) job = ImportJob.objects.create( tenant_id=tenant_id, organization_node_id=organization_node_id, entity_type=entity_type, file_id=file_id, status=ImportJobStatus.PENDING, created_by=actor, ) record_service_event( "platform_core.import_job_created", resource=job, after={"entity_type": entity_type}, ) dispatch( "core.import_job_created", { "tenant_id": tenant_id, "job_id": str(job.pk), "entity_type": entity_type, }, ) if enqueue: enqueue_import_job(job) return job def enqueue_import_job(job: ImportJob) -> None: """Dispatch the Celery task for *job*. Safe to call multiple times — the task itself is idempotent. Raises :class:`PlatformCoreError` if the job is already DONE or FAILED. """ if job.status in (ImportJobStatus.DONE, ImportJobStatus.FAILED): raise PlatformCoreError( f"Cannot enqueue import job {job.pk}: already in terminal state {job.status!r}." ) from simorgh.apps.platform_core.tasks import process_import_job process_import_job.delay(job.pk) def cancel_import_job( job: ImportJob, *, actor: AbstractUser | None = None, ) -> ImportJob: """Mark *job* as FAILED with a cancellation notice. Only PENDING or VALIDATING jobs can be cancelled. Processing or finished jobs raise :class:`PlatformCoreError`. """ actor = _resolve_actor(actor) if job.status not in (ImportJobStatus.PENDING, ImportJobStatus.VALIDATING): raise PlatformCoreError( f"Cannot cancel import job {job.pk}: status is {job.status!r}." ) now = timezone.now() ImportJob.objects.filter(pk=job.pk).update( status=ImportJobStatus.FAILED, finished_at=now, error_report=[{"row": 0, "errors": ["Job was cancelled by user."]}], ) job.refresh_from_db() record_service_event("platform_core.import_job_cancelled", resource=job) return job # --------------------------------------------------------------------------- # Export Engine # --------------------------------------------------------------------------- def create_export_job( *, tenant_id: int, organization_node_id: int, entity_type: str, fmt: str = "csv", filters: dict | None = None, actor: AbstractUser | None = None, enqueue: bool = True, ) -> ExportJob: """Create an :class:`ExportJob` and optionally enqueue it for processing. Args: tenant_id: The owning tenant. organization_node_id: The org node for the job. entity_type: Dot-notation entity type, e.g. ``"crm.lead"``. fmt: Export format: ``"csv"``, ``"xlsx"``, or ``"json"``. filters: Arbitrary filter dict passed to the exporter. actor: User who initiated the export. enqueue: If *True* (default), dispatch the Celery task. Raises: :class:`PlatformCoreError` if *entity_type* has no registered spec. """ from simorgh.apps.platform_core.export_registry import export_registry actor = _resolve_actor(actor) if entity_type not in export_registry: raise PlatformCoreError( f"No exporter registered for entity type {entity_type!r}. " f"Available: {export_registry.all_entity_types()}" ) job = ExportJob.objects.create( tenant_id=tenant_id, organization_node_id=organization_node_id, entity_type=entity_type, format=fmt, filters=filters or {}, status=ExportJobStatus.PENDING, created_by=actor, ) record_service_event( "platform_core.export_job_created", resource=job, after={"entity_type": entity_type, "format": fmt}, ) dispatch( "core.export_job_created", { "tenant_id": tenant_id, "job_id": str(job.pk), "entity_type": entity_type, }, ) if enqueue: from simorgh.apps.platform_core.tasks import process_export_job process_export_job.delay(job.pk) return job def cancel_export_job( job: ExportJob, *, actor: AbstractUser | None = None, ) -> ExportJob: """Mark a pending export job as FAILED (cancelled). Only PENDING jobs can be cancelled. Raises :class:`PlatformCoreError` for jobs that are already PROCESSING, DONE, or FAILED. """ actor = _resolve_actor(actor) if job.status != ExportJobStatus.PENDING: raise PlatformCoreError( f"Cannot cancel export job {job.pk}: status is {job.status!r}." ) ExportJob.objects.filter(pk=job.pk).update( status=ExportJobStatus.FAILED, finished_at=timezone.now(), error_message="Job was cancelled by user.", ) job.refresh_from_db() record_service_event("platform_core.export_job_cancelled", resource=job) return job # --------------------------------------------------------------------------- # Print Engine # --------------------------------------------------------------------------- def create_print_template( *, tenant_id: int, organization_node_id: int, entity_type: str, name: str, html_template: str, css: str = "", paper_size: str = PaperSize.A4, orientation: str = PageOrientation.PORTRAIT, actor: AbstractUser | None = None, ) -> PrintTemplate: """Create a new :class:`PrintTemplate` for *entity_type*.""" actor = _resolve_actor(actor) tmpl = PrintTemplate.objects.create( tenant_id=tenant_id, organization_node_id=organization_node_id, entity_type=entity_type, name=name, html_template=html_template, css=css, paper_size=paper_size, orientation=orientation, created_by=actor, is_active=True, ) record_service_event( "platform_core.print_template_created", resource=tmpl, after={"name": name, "entity_type": entity_type}, ) dispatch( "core.print_template_created", { "tenant_id": tenant_id, "template_id": str(tmpl.pk), "entity_type": entity_type, "name": name, }, ) return tmpl def update_print_template( tmpl: PrintTemplate, *, actor: AbstractUser | None = None, **fields: Any, ) -> PrintTemplate: """Update allowed fields of *tmpl* and return the refreshed instance.""" actor = _resolve_actor(actor) allowed = { "name", "html_template", "css", "paper_size", "orientation", "is_active", } before = {f: getattr(tmpl, f) for f in allowed if f in fields} for field, value in fields.items(): if field in allowed: setattr(tmpl, field, value) tmpl.save(update_fields=list(fields.keys() & allowed) + ["updated_at"]) record_service_event( "platform_core.print_template_updated", resource=tmpl, before=before, after={f: getattr(tmpl, f) for f in before}, ) dispatch( "core.print_template_updated", { "tenant_id": tmpl.tenant_id, "template_id": str(tmpl.pk), "entity_type": tmpl.entity_type, "name": tmpl.name, }, ) return tmpl def delete_print_template( tmpl: PrintTemplate, *, actor: AbstractUser | None = None, ) -> None: """Hard-delete *tmpl* after recording an audit event.""" actor = _resolve_actor(actor) record_service_event( "platform_core.print_template_deleted", resource=tmpl, before={"name": tmpl.name, "entity_type": tmpl.entity_type}, ) dispatch( "core.print_template_deleted", { "tenant_id": tmpl.tenant_id, "template_id": str(tmpl.pk), "entity_type": tmpl.entity_type, "name": tmpl.name, }, ) tmpl.delete() def render_html_preview( tmpl: PrintTemplate, entity: Any, *, language: str | None = None, extra_context: dict | None = None, ) -> str: """Render *tmpl* with *entity* as context and return the HTML string. The full HTML document (with ``" "\n" f"{body_html}\n" "" ) def render_pdf( tmpl: PrintTemplate, entity: Any, *, language: str | None = None, extra_context: dict | None = None, ) -> bytes: """Render *tmpl* with *entity* and return a PDF byte string. Uses **xhtml2pdf** as the PDF backend (pure-Python, no system libs). Raises :class:`PlatformCoreError` if PDF generation fails. """ html = render_html_preview( tmpl, entity, language=language, extra_context=extra_context ) try: from io import BytesIO from xhtml2pdf import pisa buf = BytesIO() result = pisa.CreatePDF(html, dest=buf) if result.err: raise PlatformCoreError( f"PDF generation failed for template {tmpl.pk}: " f"{result.err} error(s) encountered." ) return buf.getvalue() except PlatformCoreError: raise except Exception as exc: raise PlatformCoreError( f"PDF generation failed for template {tmpl.pk}: {exc}" ) from exc # --------------------------------------------------------------------------- # Template Engine (Message Templates) # --------------------------------------------------------------------------- _JINJA2_ENV = None def _get_jinja2_env(): """Return a lazily-created, sandboxed Jinja2 environment.""" global _JINJA2_ENV # noqa: PLW0603 if _JINJA2_ENV is None: try: from jinja2 import Environment from jinja2.sandbox import SandboxedEnvironment _JINJA2_ENV = SandboxedEnvironment(autoescape=False) except Exception as exc: raise PlatformCoreError(f"Jinja2 not available: {exc}") from exc return _JINJA2_ENV def render_template( template_code: str, context: dict, *, tenant_id: int, language: str = "en", ) -> str: """Render the body of a :class:`MessageTemplate` identified by *template_code*. Looks up the template for *tenant_id* and *language* (falls back to "en"). Renders the ``body`` field with *context* using a Jinja2 sandbox. Returns the rendered string. """ tmpl = ( MessageTemplate.objects.filter( tenant_id=tenant_id, code=template_code, language=language, is_active=True, ).first() or MessageTemplate.objects.filter( tenant_id=tenant_id, code=template_code, language="en", is_active=True, ).first() ) if tmpl is None: raise PlatformCoreError( f"No active MessageTemplate with code={template_code!r} found for tenant {tenant_id}." ) env = _get_jinja2_env() try: tmpl_obj = env.from_string(tmpl.body) return tmpl_obj.render(**context) except Exception as exc: raise PlatformCoreError( f"Template rendering failed for {template_code!r}: {exc}" ) from exc def render_template_subject( template_code: str, context: dict, *, tenant_id: int, language: str = "en", ) -> str: """Render the subject of a :class:`MessageTemplate`. Returns an empty string if the template has no subject or does not exist. """ tmpl = ( MessageTemplate.objects.filter( tenant_id=tenant_id, code=template_code, language=language, is_active=True, ).first() or MessageTemplate.objects.filter( tenant_id=tenant_id, code=template_code, language="en", is_active=True, ).first() ) if tmpl is None or not tmpl.subject: return "" env = _get_jinja2_env() try: return env.from_string(tmpl.subject).render(**context) except Exception as exc: raise PlatformCoreError( f"Subject rendering failed for {template_code!r}: {exc}" ) from exc def create_message_template( *, tenant_id: int, organization_node_id: int, code: str, name: str, channel: str = MessageChannel.EMAIL, subject: str = "", body: str, variables: list | None = None, language: str = "en", actor, ) -> MessageTemplate: """Create and persist a new :class:`MessageTemplate`.""" actor = _resolve_actor(actor) tmpl = MessageTemplate.objects.create( tenant_id=tenant_id, organization_node_id=organization_node_id, code=code, name=name, channel=channel, subject=subject, body=body, variables=variables or [], language=language, is_active=True, created_by=actor, ) record_service_event("platform_core.message_template_created", resource=tmpl) return tmpl def update_message_template( tmpl: MessageTemplate, *, actor, **fields, ) -> MessageTemplate: """Update allowed fields on *tmpl*. Allowed: ``name``, ``channel``, ``subject``, ``body``, ``variables``, ``language``, ``is_active``. """ _resolve_actor(actor) allowed = {"name", "channel", "subject", "body", "variables", "language", "is_active"} invalid = set(fields) - allowed if invalid: raise PlatformCoreError(f"Cannot update field(s): {', '.join(sorted(invalid))}") for key, val in fields.items(): setattr(tmpl, key, val) tmpl.save(update_fields=list(fields.keys()) + ["updated_at"]) record_service_event("platform_core.message_template_updated", resource=tmpl) return tmpl def delete_message_template(tmpl: MessageTemplate, *, actor) -> None: """Hard-delete *tmpl* after recording an audit event.""" _resolve_actor(actor) record_service_event("platform_core.message_template_deleted", resource=tmpl) tmpl.delete() # --------------------------------------------------------------------------- # Document Template Engine # --------------------------------------------------------------------------- def create_document_template( *, tenant_id: int, organization_node_id: int, code: str, name: str, entity_type: str, description: str = "", sections: list[dict[str, Any]] | None = None, variables: list[dict[str, Any]] | None = None, css: str = "", paper_size: str = PaperSize.A4, orientation: str = PageOrientation.PORTRAIT, actor: AbstractUser | None = None, ) -> DocumentTemplate: """Create a new :class:`DocumentTemplate` for *entity_type*.""" actor = _resolve_actor(actor) doc = DocumentTemplate.objects.create( tenant_id=tenant_id, organization_node_id=organization_node_id, code=code, name=name, entity_type=entity_type, description=description, sections=sections or [], variables=variables or [], css=css, paper_size=paper_size, orientation=orientation, created_by=actor, is_active=True, ) record_service_event( "platform_core.document_template_created", resource=doc, after={"code": code, "name": name, "entity_type": entity_type}, ) dispatch( "core.document_template_created", { "tenant_id": tenant_id, "template_id": str(doc.pk), "code": code, "entity_type": entity_type, }, ) return doc def update_document_template( doc: DocumentTemplate, *, actor: AbstractUser | None = None, **fields: Any, ) -> DocumentTemplate: """Update allowed fields of *doc* and return the refreshed instance.""" actor = _resolve_actor(actor) allowed = { "name", "description", "sections", "variables", "css", "paper_size", "orientation", "is_active", } before = {f: getattr(doc, f) for f in allowed if f in fields} for field, value in fields.items(): if field in allowed: setattr(doc, field, value) doc.save(update_fields=list(fields.keys() & allowed) + ["updated_at"]) record_service_event( "platform_core.document_template_updated", resource=doc, before=before, after={f: getattr(doc, f) for f in before}, ) dispatch( "core.document_template_updated", { "tenant_id": doc.tenant_id, "template_id": str(doc.pk), "code": doc.code, "entity_type": doc.entity_type, }, ) return doc def delete_document_template( doc: DocumentTemplate, *, actor: AbstractUser | None = None, ) -> None: """Hard-delete *doc* after recording an audit event.""" actor = _resolve_actor(actor) record_service_event( "platform_core.document_template_deleted", resource=doc, before={"code": doc.code, "name": doc.name, "entity_type": doc.entity_type}, ) dispatch( "core.document_template_deleted", { "tenant_id": doc.tenant_id, "template_id": str(doc.pk), "code": doc.code, "entity_type": doc.entity_type, }, ) doc.delete() def render_document( doc: DocumentTemplate, context: dict[str, Any], *, language: str = "en", ) -> str: """Render a :class:`DocumentTemplate` with *context* and return the full HTML string. Each section's content is rendered as a Jinja2 template with the provided context, then assembled into a complete HTML document with the template's CSS and paper-size settings. """ from jinja2 import Environment, StrictUndefined env = Environment(undefined=StrictUndefined, autoescape=True) context.setdefault("language", language) sections_html_parts: list[str] = [] for section in doc.sections: section_content = section.get("content", "") if not section_content: continue try: tmpl = env.from_string(section_content) rendered = tmpl.render(**context) section_type = section.get("type", "body") section_id = section.get("id", "") sections_html_parts.append( f'
\n{rendered}\n
' ) except Exception as exc: raise PlatformCoreError( f"Section rendering failed for '{section.get('id', '?')}' " f"in document template '{doc.code}': {exc}" ) from exc body_html = "\n".join(sections_html_parts) page_css = ( f"@page {{ size: {doc.paper_size} {doc.orientation}; margin: 15mm; }}" ) full_css = f"{page_css}\n{doc.css}" return ( "\n" "\n" "" f"" "\n" f"{body_html}\n" "" ) def render_document_pdf( doc: DocumentTemplate, context: dict[str, Any], *, language: str = "en", ) -> bytes: """Render a :class:`DocumentTemplate` and return a PDF byte string. Uses **xhtml2pdf** as the PDF backend. Raises :class:`PlatformCoreError` if PDF generation fails. """ html = render_document(doc, context, language=language) try: from io import BytesIO from xhtml2pdf import pisa buf = BytesIO() result = pisa.CreatePDF(html, dest=buf) if result.err: raise PlatformCoreError( f"PDF generation failed for document template {doc.pk}: " f"{result.err} error(s) encountered." ) return buf.getvalue() except PlatformCoreError: raise except Exception as exc: raise PlatformCoreError( f"PDF generation failed for document template {doc.pk}: {exc}" ) from exc # --------------------------------------------------------------------------- # Export Layout Engine # --------------------------------------------------------------------------- def create_export_layout( *, tenant_id: int, organization_node_id: int, code: str, name: str, entity_type: str, export_format: str = "csv", columns: list[dict[str, Any]] | None = None, grouping: list[str] | None = None, aggregation: dict[str, str] | None = None, styling: dict[str, Any] | None = None, header_footer: dict[str, Any] | None = None, page_setup: dict[str, Any] | None = None, is_default: bool = False, actor: AbstractUser | None = None, ) -> ExportLayout: """Create a new :class:`ExportLayout` for *entity_type*.""" actor = _resolve_actor(actor) layout = ExportLayout.objects.create( tenant_id=tenant_id, organization_node_id=organization_node_id, code=code, name=name, entity_type=entity_type, export_format=export_format, columns=columns or [], grouping=grouping or [], aggregation=aggregation or {}, styling=styling or {}, header_footer=header_footer or {}, page_setup=page_setup or {}, is_default=is_default, created_by=actor, is_active=True, ) record_service_event( "platform_core.export_layout_created", resource=layout, after={"code": code, "name": name, "entity_type": entity_type, "export_format": export_format}, ) dispatch( "core.export_layout_created", { "tenant_id": tenant_id, "layout_id": str(layout.pk), "code": code, "entity_type": entity_type, }, ) return layout def update_export_layout( layout: ExportLayout, *, actor: AbstractUser | None = None, **fields: Any, ) -> ExportLayout: """Update allowed fields of *layout*.""" actor = _resolve_actor(actor) allowed = { "name", "columns", "grouping", "aggregation", "styling", "header_footer", "page_setup", "is_default", "is_active", } before = {f: getattr(layout, f) for f in allowed if f in fields} for field, value in fields.items(): if field in allowed: setattr(layout, field, value) layout.save(update_fields=list(fields.keys() & allowed) + ["updated_at"]) record_service_event( "platform_core.export_layout_updated", resource=layout, before=before, after={f: getattr(layout, f) for f in before}, ) dispatch( "core.export_layout_updated", { "tenant_id": layout.tenant_id, "layout_id": str(layout.pk), "code": layout.code, "entity_type": layout.entity_type, }, ) return layout def delete_export_layout( layout: ExportLayout, *, actor: AbstractUser | None = None, ) -> None: """Hard-delete *layout* after recording an audit event.""" actor = _resolve_actor(actor) record_service_event( "platform_core.export_layout_deleted", resource=layout, before={"code": layout.code, "name": layout.name, "entity_type": layout.entity_type}, ) dispatch( "core.export_layout_deleted", { "tenant_id": layout.tenant_id, "layout_id": str(layout.pk), "code": layout.code, "entity_type": layout.entity_type, }, ) layout.delete()