Celery tasks

MEG Docs

(celery task)megdocs.tasks.parse_document_content(version_id: int)

Proxy that evaluates object once.

Proxy will evaluate the object each time, while the promise will only evaluate it once.

(celery task)megdocs.tasks.download_cloud_version(log_uid: str) → None

Download file from OneDrive and attach it to a Version.

Parameters:

log_uid – UID of the CloudVersionDownloadLog.

megdocs.tasks.get_review_notification_recipients(document: Document) → set[Auditor]

Return the owner, reviewer, and their direct delegates for a review notification. Direct delegates only — delegators and co-delegates are not notified for review reminders. Relies on prefetched owner__delegates and current_version__reviewer__delegates.

Parameters:

document – The document whose owner and reviewer should be notified.

Returns:

Set of auditors to notify.

(celery task)megdocs.tasks.send_review_notifications()

Proxy that evaluates object once.

Proxy will evaluate the object each time, while the promise will only evaluate it once.

(celery task)megdocs.tasks.send_attestation_notifications_batch(checkbox_id: int, auditor_ids: list[int]) → None

Sends attestation notifications for a batch of auditors for a given DocumentCheckbox Batches auditor notifications to reduce number of celery tasks queued

Parameters:

  • checkbox_id – The ID of the checkbox object to filter users for

  • auditor_ids – List of auditor IDs to send notifications for

(celery task)megdocs.tasks.share_document(recipient_id: int, sharer_name: str, document_id: int, note: str = '') → None

Emails a user to notify them that another user has shared a document with them.

Parameters:

  • recipient_id – The recipient auditor’s id.

  • sharer_name – Name of the user who shared the document.

  • document_id – The document’s id.

  • note – The note to be sent to the user

(celery task)megdocs.tasks.submit_draft_as_new_version(draft_id: int, auditor_id: int, version_data: dict, *, publish=False) → int

Renders draft into a PDF file and submits a new version

Parameters:

  • draft_id – pk of the draft object

  • auditor_id – the id of the user who is creating the version

  • version_data – dict of any additional kwargs for the new Version

  • publish – whether to approve and publish the version automatically

Returns:

pk of the new version object

(celery task)megdocs.tasks.update_version_draft(version_id: int)

Updates version in review with the latest contents from draft:

  • pdf version

  • documents plaintext contents

This job runs if DRAFT_UPDATE_REVIEW is set

(celery task)megdocs.tasks.generate_pdf_preview(html: str) → str

Generate a pdf document from given input HTML string. Uses html_to_pdf() to write preview to a temporary pdf file and returns its url (for example /media/CACHE/pdf_preview/0F0F0F0F0F0F0F.pdf). The preview file is deleted after time defined in PDF_PREVIEW_EXPIRY_MINS.

Uses file contents as the hash to determine filename, and if already exists, existing file is returned instead of rendering the pdf again.

Parameters:

html – The source HTML string

Returns:

the url to the generated PDF file

(celery task)megdocs.tasks.import_word_document_ckeditor(document_id: int, user_id: int) → int

Imports document’s word version and creates or updates document’s draft with its contents. Creates draft object if one does not exist. If exists, overrides draft’s contents

The document’s current version must have a current version with “source” being a word docx file. The user must have access to the document

Parameters:

  • document_id – pk of the Document

  • user_id – pk of the User

Returns:

pk of the draft object

(celery task)megdocs.tasks.import_word_document_from_file_ckeditor(document_id: int, user_id: int, file_path: str) → int

Parses The MS Word docx file and creates or updates Document’s draft with its contents. Creates draft object if one does not exist. If exists, overrides draft’s contents After successful import, the docx file is deleted.

Parameters:

  • document_id – pk of the Document

  • user_id – pk of the User - The user must have access to the document

  • file_path – absolute path to the Word DOCX file, must be in an accessible directory such as the /media/ folder

Returns:

pk of the draft object

(celery task)megdocs.tasks.create_user_notifications_version_pending_review(reviewer_id: int) → None

Task to bulk create pending-review notifications for versions awaiting the reviewer’s action. Excludes versions belonging to archived documents. Given the unique constraint on UserNotification, conflicts will be ignored in bulk create and duplicates will not be created.

Parameters:

reviewer_id – the id of the reviewer the notifications will be created for

(celery task)megdocs.tasks.create_user_notifications_document_due_review(reviewer_id: int) → None

Task to bulk create periodic due-review notifications for documents whose review interval has elapsed. Given the unique constraint on UserNotification, conflicts will be ignored in bulk create and duplicates will not be created.

Parameters:

reviewer_id – the id of the reviewer the notifications will be created for

(celery task)megdocs.tasks.create_user_notifications_version_awaiting_publish(reviewer_id: int) → None

Task to bulk create document publish notifications for the versions reviewer Given the unique constraint on UserNotification, conflicts will be ignored in bulk create and duplicates will not be created

Param:

reviewer_id: the id of the documents versions reviewer that the notification will be created for

(celery task)megdocs.tasks.create_user_notifications_version_publication_process(version_id: int) → None

Task to bulk create document publication process notifications for those users in the version approval config’s current step that have not yet approved Given the unique constraint on UserNotification, conflicts will be ignored in bulk create and duplicates will not be created

Param:

version_id: the ids of the document version that will be queried to get the approval config and approvers

(celery task)megdocs.tasks.create_user_notifications_approval_config_publication_process(approval_config_id: int) → None

Task to bulk create document publication process notifications for those users in the version approval config’s current step that have not yet approved Given the unique constraint on UserNotification, conflicts will be ignored in bulk create and duplicates will not be created

Param:

approval_config_id: the id of the approval config object by which the task was triggered

(celery task)megdocs.tasks.create_user_notifications_version_decline(version_id: int) → None

Create bell notifications for the document owner and controller when a version is declined.

The notification points to the declined document and is created once per decline for each recipient. The owner-side recipients utilize the institution’s use_owner_field config. The unique constraint on UserNotification ensures repeat declines do not duplicate a recipient’s notification.

Parameters:

version_id – the id of the declined version

(celery task)megdocs.tasks.create_user_notifications_attestation_required(document_id: int) → None

Create user notifications for documents that need to be attestation Queries the checkbox model associated with the document and annotates the auditor_ids on the model Loops through all relevant auditor ids for both teams and standalone auditors and creates a notification for each

Parameters:

document_id – The ID of the document that was recently updated with a new current version (published)

(celery task)megdocs.tasks.publish_document_version_from_schedule(version_id: int, auditor_id: int) → None

Task to publish a document version checking if the auditor who invoked the task is indeed the reviewer Will fail if the version is no approved (for example if the review has been revoked)

Parameters:

  • version_id – the id of the version to be published

  • auditor_id – the id of the auditor who triggered the task

(celery task)megdocs.tasks.delete_current_version_due_review_notifications(version_id: int)

Delete relevant notifcation when a version is marked as reviewed according to its interval

(celery task)megdocs.tasks.delete_relevant_version_notifications(version_id: int, current_step: int)

Update user notification by deleting the relevant notification based on the notification type

Parameters:

  • version_id – the version id for which the notification will be filtered by

  • current_step – the current step of the publication process

(celery task)megdocs.tasks.delete_previous_version_notifications(version_id: int) → None

Task to cleanup any redundant notifications relevant to the version’s document Queries all previous versions of the document except the current version on publish and deletes stale notifications

Parameters:

version_id – The id of the version to exclude (document’s current version)

(celery task)megdocs.tasks.delete_existing_notification_approval_config(approval_config_id: int, document_id: int | None = None) → None

Task to delete any redundant notifications relevant to the approval config’s documents. When document_id is provided, only notifications for that document’s versions are deleted.

Parameters:

  • approval_config_id – the id of the approval config whose notifications should be deleted

  • document_id – optional document id to scope deletion to a single document

megdocs.tasks.get_comment_mention_notification_filter(document_id: int) → Q

Builds a Q filter matching comment mention notifications for a document. Handles both CKEditor inline comments (DocumentComment on draft) and side-panel comments (Comment on version).

Parameters:

document_id – ID of the document

Returns:

Q filter for matching notifications, or a Q that matches nothing if none found

(celery task)megdocs.tasks.delete_comment_mention_notifications_for_document(document_id: int) → int

Deletes all comment mention notifications linked to comments on a document. Called when a document’s workflow step (review/approval) completes.

Parameters:

document_id – ID of the document whose comment mention notifications should be cleared

Returns:

Number of deleted notifications

(celery task)megdocs.tasks.generate_llm_response_for_document_chat_message(auditor_id: int, ai_message_id: int, ai_chat_id: int, version_id: int) → dict[str, bool]

Given an existing placeholder message and an AI chat, creates a LLM reply and updates the placeholder with the reply.

Steps:

  1. Feeds chat history to LLM and requests a new response

  2. If response contains a title adds the title to the chat.

  3. Updates placeholder-message content field object with LLM reply.

(celery task)megdocs.tasks.delete_attestation_required_notification(document_id: int, auditor_ids: list[int] | None = None) → None

Delete attestation required notifications for specific auditors and document.

This task removes user notifications of type NOTIFICATION_TYPE_DOCUMENT_ATTESTATION_REQUIRED for the given auditors and document. It is typically called when: - A user completes an attestation (checks the checkbox) - A checkbox is deleted

Parameters:

  • document_id – The ID of the document for which to delete notifications

  • auditor_ids – List of auditor IDs whose notifications should be deleted

(celery task)megdocs.tasks.expire_checkbox_attestations(checkbox_id: int) → None

Unpublish all DocumentCheckboxState records for a given checkbox when the review_interval has expired. This forces users to re-attest.

After expiring attestations, schedules the next expiration and reminder to maintain the review cycle.

Parameters:

checkbox_id – The ID of the DocumentCheckbox whose attestations should expire

(celery task)megdocs.tasks.send_attestation_reminder_emails(checkbox_id: int) → None

Send reminder emails to users who have not yet attested to a document checkbox and reschedule the next reminder at the configured interval until every relevant user has attested.

Parameters:

checkbox_id – The ID of the DocumentCheckbox to send reminders for

Files

(celery task)files.tasks.delete_media_file(path: str)

Deletes a file in the media folder. The task will fail if the file does not exist, of path leads outside the media folder. Use this task to lazily delete a file using celery or put it on a timer.

Parameters:

path – relative path to the preview file within media directory (for example CACHE/test.jpeg)

Raises:

FileNotFoundError – if the file does not exist

Audit builder

(celery task)audit_builder.tasks.calculate_llm_topics_for_fields(session_id: 'int')

Query llm to determine topics in text of text custom fields and output the result to a multiple choice field.

Parameters:

session_id – pk of the AuditSession obj

(celery task)audit_builder.tasks.calculate_observations(schedule_id: 'int') → 'int'

Runs field re-calculation in form and fields defined by given schedule

(celery task)audit_builder.tasks.calculate_sentiment_for_fields(session_id: 'int')

Runs sentiment analysis on text custom fields and outputs the result to a choice field.

Parameters:

session_id – pk of the AuditSession obj

(celery task)audit_builder.tasks.calculate_session_fields(session_id: 'int', user_id: 'int | None' = None)

Trigger calculation for dynamic fields in given session

Parameters:

  • session_id – pk of the AuditSession obj

  • user_id – optional user id if the action was triggered by user. When provided, the action will be logged against that user.

audit_builder.tasks.convert_openai_to_google_format(messages: list[ChatCompletionMessageParam]) → tuple[str | None, list[genai_types.Content]]

Converts text-only OpenAI messages to Google GenAI format, separating the system instruction .

(celery task)audit_builder.tasks.daily_calculate_observations()

Triggers fields calculation for all forms where calculation is scheduled

(celery task)audit_builder.tasks.decompress_observation_sessions(session_ids: 'list[int]', user_id: 'int | None' = None)

Decompresses the given audit sessions using decompress_observations().

Every session shares a single transaction, so the whole batch is either migrated or rolled back, leaving no partially converted selection. Each session that changed is recorded in the admin log.

Parameters:

  • session_ids – pks of the AuditSession objects to decompress

  • user_id – optional user id if the action was triggered by user. When provided, the action will be logged against that user.

audit_builder.tasks.generate_llm_response(messages_for_llm: list[ChatCompletionMessageParam], safety_identifier: str | None = None) → str

Takes Openai style prompt and feeds it to ChatGPT or Gemini or does nothing, depending on env vars. Handles errors and produces reusable simple errors.

Parameters:

  • messages_for_llm – Openai-style chat messages to send to the LLM.

  • safety_identifier – Tenant-scoped identifier forwarded to the OpenAI backend for abuse monitoring.

(celery task)audit_builder.tasks.generate_titles_for_fields(session_id: 'int')

Generates titles from text custom fields and outputs to another text field.

Parameters:

session_id – pk of the AuditSession obj

audit_builder.tasks.get_custom_topics_from_llm(text: str, topics: list[str], examples: list[list[str]], extra_prompt: str | None = None, safety_identifier: str | None = None) → str

Prompts a LLM to determine if any of the predefined topics/classes are present in the text.

Parameters:

  • text – string to be auto classified

  • topics – list of topics/classes

  • examples – list of lists of examples (format [question, answer]

  • extra_prompt – string or None, gives LLM more context about the task

  • safety_identifier – Tenant-scoped identifier forwarded to the OpenAI backend for abuse monitoring.

Returns:

string containing topics/classes extracted

audit_builder.tasks.get_empty_answers(subform: CustomSubform) → dict

Creates empty answers dictionary for the subform’s fields. Ensures choice fields are initialized with None to preserve options.

audit_builder.tasks.get_llm_client() → AzureOpenAI | genai.Client | None

Connects to Open AI Azure API or Google Genie API

Returns:

Azure client object or Google Genie client object

audit_builder.tasks.get_request_topics_messages_for_llm(text: str, topics: list[str], examples: list[list[str]], extra_prompt: str | None = None) → list[dict[str, str]]

Creates messages to send to LLM to request it to select topics/classes from a text.

Parameters:

  • text – string to be auto classified

  • topics – list of topics/classes

  • examples – list of lists of examples (format [question, answer]

  • extra_prompt – string or None, gives LLM more context about the task

Returns:

list of dictionaries containing messages to send to LLM

audit_builder.tasks.get_sentiment_llm_prompt(text: str, extra_examples: list[list[str]], question: str | None = None) → list[dict[str, str]]

Creates messages to send to LLM to request it to select topics/classes from a text. We ask the result to be placed in brackets as sometimes the LLM produces an explanation with the answer (easier to extract answer).

Parameters:

  • text – The answer text to analyze

  • extra_examples – Extra examples for few-shot learning

  • question – Optional question context to provide semantic understanding

audit_builder.tasks.get_temp_wav_file(audio_path: str) → Generator[tuple[str, int], None, None]

Creates a temporary WAV file from an audio file with standardized format and handles cleanup.

Parameters:

audio_path – Path to the source audio file (can be any audio format)

Returns:

Path to temporary WAV file converted to mono 16kHz WAV format that will be automatically cleaned up

Raises:

  • OSError – If file operations fail

  • AudioSegmentException – If audio conversion fails

The file is converted to a standardized format required by Azure Speech Services: - Mono channel audio - 16kHz sample rate - 16-bit PCM WAV encoding

audit_builder.tasks.get_title_generation_prompt(text: str, extra_examples: list[list[str]]) → list[dict[str, str]]

Creates messages to send to LLM to generate a concise title from text.

audit_builder.tasks.handle_llm_errors(func)

A decorator to handle common API errors for both Google and OpenAI LLM clients. It captures exceptions and raises standardized, user-friendly errors.

(celery task)audit_builder.tasks.import_form_json(institution_id: 'int', *, path: 'str', user_id: 'int | None', **kwargs) → 'Sequence[int]'

Celery job that imports a form from a json file and populates it with dummy data

Parameters:

  • institution_id

  • path – path to the json file - must be available to celery worker (i.e. be part of the docker image, or located in a shared folder, like media files)

  • user_id – id of the used to associate the dummy observations with

audit_builder.tasks.increment_sequence_id(config: AuditFormConfig) → int

Increment and return next sequential id for a locked config object. This function expects the config object to already be locked with select_for_update() within a transaction.atomic() context. It does not acquire its own lock.

Parameters:

config – A locked AuditFormConfig instance (must be locked with select_for_update)

Raises:

ValueError – If the form is not sequence enabled

Returns:

The incremented sequential id

Iterates through all observations in the session and processes related observations for each. Checks if the form implements custom subforms and if so, gets the related subobservations. Otherwise gets the related custom observations.

Parameters:

  • session_id – The pk of the session containing observations to be linked.

  • created – Is true if the session was just created.

Only custom forms are supported.

(celery task)audit_builder.tasks.migrate_observations(*, source_form_id: 'int', target_form_id: 'int', move: 'bool', dry_run: 'bool' = False, report_email_address: 'str')

Proxy that evaluates object once.

Proxy will evaluate the object each time, while the promise will only evaluate it once.

(celery task)audit_builder.tasks.parse_lacas_forms(imported_forms: 'list[int]', run_calculations=True) → 'int'

Processes the LACAS form imported from JSON file by correcting the related form ids and observation answers

Parameters:

imported_forms – ids of the forms (first id should related to the summary form) returned by import_form_json job

audit_builder.tasks.perform_sentiment_analysis(text: str, extra_examples: list[list[str]], question: str | None = None, safety_identifier: str | None = None) → Literal['positive', 'neutral', 'negative', 'unknown']

Runs sentiment analysis on given text using Chat-GPT or Gemini model

Parameters:

  • text – string to be passed to the model

  • extra_examples – extra example, use for multilingual examples if needed

  • question – optional question context to provide semantic understanding of the answer

  • safety_identifier – Tenant-scoped identifier forwarded to the OpenAI backend for abuse monitoring.

Returns:

sentiment analysis result

(celery task)audit_builder.tasks.process_default_values(session_id: 'int')

Process default values for fields where ‘show_in_app’ is False

First finds subforms using default value logic and checks if they dont have a sub observation. If they are missing a sub obs its added then default values are added to these.

(celery task)audit_builder.tasks.process_sequence_ids(session_id: 'int')

If sequential ids are enabled for the form, set sequence ids for all observations in a given session

Parameters:

session_id – Session to set sequence ids for

(celery task)audit_builder.tasks.reset_audit_form_sequence_ids(form_id: 'int')

Reset audit form sequence ids

Parameters:

form_id – Audit form to reset sequence ids for

audit_builder.tasks.safe_llm_task(func: Callable[[...], Any]) → Callable[[...], Any]

Decorator for Celery tasks to handle retries and soft-failures for LLM operations.

Catches standardized LLM errors (LLMAuthError, LLMAPIError) and triggers Celery’s built-in retry mechanism. If maximum retries are exceeded and the allow_failures kwarg is True, the error is swallowed, logged to Sentry.

Note: The decorated task must use @app.task(bind=True).

(celery task)audit_builder.tasks.schedule_observation_reviews(session_id: 'int')

For forms that have enabled the feature, schedule automatic observation reviews.

(celery task)audit_builder.tasks.schedule_workflow_tasks(session_id: 'int')

Schedule workflow tasks for a given audit session. This task schedules workflow tasks based on two triggers: creation and observation answer dates. It processes all session observations to schedule tasks accordingly.

Parameters:

session_id – The ID of the AuditSession for which workflow tasks are to be scheduled.

(celery task)audit_builder.tasks.share_observation(recipient_id: 'int', sharer_name: 'str', observation_id: 'int', note: 'str' = '') → 'None'

Emails a user to notify them that another user has shared an observation with them.

Parameters:

  • recipient_id – The recipient auditor’s id.

  • sharer_name – Name of the user who shared the document.

  • observation_id – The observation id.

  • note – The note to be sent to the user

(celery task)audit_builder.tasks.transcribe_audio_fields(session_id: 'int')

Transcribes audio custom fields and outputs the result to a text field.

Parameters:

session_id – pk of the AuditSession obj

(celery task)audit_builder.tasks.update_observation_auditors_task(observation_id: 'int', auditor_fields: 'list[tuple[int | None, list[tuple[str, int]]]]') → 'int'

Async job wrapper for update_observation_auditors().

Parameters:

  • observation_id – Custom observation id

  • auditor_fields – list of subform ids, and the field names within that subform. None represents observation-level questions. Must be an exhaustive list of all auditor fields. This list will be converted to a dict before passing it to update_observation_auditors() - the dict is not json-serializable due to int and null keys.

(celery task)audit_builder.tasks.update_observation_generated_title_fields(observation_id: 'int', user_id: 'int', is_sub_observation: 'bool | None' = False) → 'None'

Updates AI-generated titles for an observation

Parameters:

  • observation_id – ID of observation to update

  • user_id – ID of user who made the change

  • is_sub_observation – whether observation is a sub observation or not

  • save – whether to save to db or not (set to false if using in bulk update)

(celery task)audit_builder.tasks.update_observation_transcribe_audio_fields(observation_id: 'int', user_id: 'int', is_sub_observation: 'bool | None' = False)

Updates AI-generated audio transcriptions for an observation when source fields change.

param: observation_id: ID of observation to update param: user_id: ID of user who made the change param: is_sub_observation: whether observation is a sub observation or not param: save: whether to save to db or not (set to false if using in bulk update)