#2083: Программирование собственных расширений Swarmica на Python

Скрипты-плагины позволяют вашей команде создавать серверную логику под процессы поддержки: проверять условия, работать с заявками и связанными объектами, формировать отчёты, выполнять массовые операции и обмениваться данными с внешними системами.

Вы пишете Python-файл, подключаете его в Swarmica и выбираете способ запуска: событие, расписание или веб-форму. Через параметры один и тот же скрипт можно использовать для разных сценариев.

Это руководство рассчитано на разработчика или администратора с базовыми знаниями Python. Оно объясняет механизм и приёмы программирования на основе приложенных к базе знаний примеров.

Описанные ниже сценарии и примеры могут быть неполными и не охватывают всех возможностей скриптов-плагинов Swarmica. Если вы не нашли нужный метод, не понимаете, как реализовать свой сценарий, или не смогли написать работающий скрипт, обратитесь в техподдержку Swarmica со своим вопросом. Опишите, что хотите сделать, укажите версию системы и приложите свой код и текст ошибки, если они есть. Мы поможем разобраться с доступными механизмами и способами реализации.

Версии и проверка. Материал подготовлен по документации и файлам, доступным 17 сентября 2026 года. Внутренние модели и методы могут различаться между версиями. Учебные примеры нужно проверить в тестовой установке вашей версии Swarmica перед рабочим запуском. Метка VERSION внутри файла обозначает версию самого скрипта, а не минимальную версию Swarmica.

Содержание

1. Что можно программировать и какой механизм выбрать.
2. Как устроена среда выполнения.
3. Первый скрипт: загрузка, веб-форма и проверка результата.
4. Структура Python-файла и возможности Runner.
5. Параметры, типы данных и дополнительный контекст.
6. Запуск по событию и работа с event.
7. Поиск объектов и связанные данные.
8. Работа с заявками, комментариями и чатами.
9. Пользовательские поля и счётчик назначений.
10. Выполнение по расписанию и расчёт рабочего времени.
11. Уведомления, шаблоны и выбор получателей.
12. Отчёты CSV и Excel.
13. Внешние API и постраничная загрузка.
14. Импорт пользователей, компаний и статей.
15. Массовые операции, транзакции и предварительный просмотр.
16. Повторные запуски и параллельная обработка.
17. Диагностика, производительность и эксплуатация.
18. Каталог примеров и маршрут дальнейшего изучения.

1. Что можно программировать

Собственное расширение полезно, когда нужное поведение требует кода: дополнительных проверок, вычислений, работы с несколькими объектами или обработки ответа внешней системы.

Например, Python-скрипт может подсчитать назначения заявки и записать результат в поле, найти обращения без ответа, сформировать файл отчёта или импортировать данные из CSV.

Веб-виджеты — отдельный механизм расширения интерфейса. Их настройка описана в статье «Собственные виджеты в Swarmica». В этом руководстве рассматривается серверный Python-код.

2. Среда выполнения

Скрипт выполняется внутри установки Swarmica и может импортировать модели и внутренние функции продукта. Поэтому для доступа к локальным данным не обязательно выполнять HTTP-запросы к своему же API: примеры используют Django ORM.

Это карта импортов из исследованных файлов, а не обещание неизменного программного интерфейса. Описание многих объектов доступно в статье «Объекты и параметры для триггеров».

Права и ресурсы

В документации указано, что скрипты выполняются от root внутри контейнера, имеют доступ к основной БД и общим файловым разделам и используют ресурсы совместно со Swarmica. Изоляция контейнера не защищает данные продукта от ошибочного кода.

Назначение роли для доступа к веб-форме не превращает прямой запрос ORM в пользовательский запрос с автоматической проверкой видимости всех объектов. Если форму могут запускать разные сотрудники, ограничения на доступные данные и действия нужно учитывать в реализации.

Для изменений используйте тестовую установку, ограниченную выборку и резервную копию. Не выводите пароли, токены и весь контекст запуска в лог. Эти требования особенно важны для импортов и массовых операций.

Условия использования механизма приведены в документации по Python-плагинам. Для скриптов-плагинов в опубликованных примерах указан тариф Премиум или выше; актуальные условия смотрите в тарифах.

Библиотеки

Документация перечисляет, в частности, requests, openpyxl, pandas, bs4, а также средства работы с JSON, CSV, датами и регулярными выражениями. Импортируйте только то, что нужно. Наличие произвольной сторонней библиотеки и её версию проверяйте в своей установке.

3. Первый скрипт: посмотреть выбранные заявки

Начните с чтения данных и записи результата в лог. Это позволяет проверить подключение, параметры и доступ к моделям без изменения заявок.

Шаг 1. Создайте Python-файл

Сохраните следующий самостоятельный пример как read_group_tickets.py.

import logging


from core.models import Group, Ticket

from runtime.runners import BaseRunner

from swarmica import defs




logger = logging.getLogger(name)

VERSION = "1.0.0"




class Runner(BaseRunner):

def run(self, *args, **kwargs):

prefix = f"{self.script.name} ({self.script.uid})"

data = kwargs.get("data") or {}

if not isinstance(data, dict):

logger.error("%s: data должен быть словарём", prefix)

return



    group_uid = data.get("group_uid")
    if not isinstance(group_uid, str) or not group_uid.strip():
        logger.error("%s: укажите group_uid", prefix)
        return

    raw_limit = data.get("limit", 10)
    if isinstance(raw_limit, bool) or not isinstance(raw_limit, (int, str)):
        logger.error("%s: limit должен быть целым числом", prefix)
        return
    try:
        limit = int(raw_limit)
    except ValueError:
        logger.error("%s: limit должен быть целым числом", prefix)
        return
    if not 1 <= limit <= 100:
        logger.error("%s: limit должен быть от 1 до 100", prefix)
        return

    group = Group.objects.filter(uid=group_uid.strip()).first()
    if group is None:
        logger.warning("%s: группа не найдена", prefix)
        return

    tickets = Ticket.objects.filter(
        group=group,
        status__in=[defs.TICKET_STATUSES_NEW, defs.TICKET_STATUSES_OPEN],
    ).select_related("assignee").order_by("id")[:limit]

    shown = 0
    for ticket in tickets:
        assignee_uid = ticket.assignee.uid if ticket.assignee else None
        logger.info(
            "%s: ticket_id=%s status=%s assignee_uid=%s",
            prefix, ticket.id, ticket.status, assignee_uid,
        )
        shown += 1
    logger.info("%s: версия=%s показано=%s", prefix, VERSION, shown)

Скрипт ищет группу по UID и выводит ограниченный список новых и открытых заявок. Если группа не существует или параметры неверны, он завершает работу с пояснением. Если заявок нет, в логе будет показано=0.

Шаг 2. Подключите файл

1. Войдите в Swarmica как администратор.
2. Откройте Настройки → Автоматизация и маршрутизация → Скрипты. В некоторых версиях раздел называется Настройки → Скрипты.
3. Создайте скрипт и задайте название.
4. Для первого теста разрешите запуск только администратору.
5. Загрузите read_group_tickets.py в поле скрипта.
6. Сохраните настройки.

Шаг 3. Добавьте веб-форму

В параметры веб-формы внесите:

[
  {
    "name": "group_uid",
    "type": "string",
    "required": true,
    "readonly": false,
    "displayName": "UID группы"
  },
  {
    "name": "limit",
    "type": "string",
    "required": true,
    "readonly": false,
    "displayName": "Сколько заявок показать: от 1 до 100"
  }
]

name — имя, которое код читает через data.get(...). displayName — подпись для пользователя. В этом примере лимит поступает из текстового поля и преобразуется в число самим скриптом.

В исследованных формах также встречаются типы boolean, date, datetime и dropdown. Для выпадающего списка нужны варианты choices; итоговое значение должно соответствовать тому, что ожидает код. Точный набор типов и формат вариантов сверяйте с интерфейсом своей версии и примером смены группы.

Шаг 4. Запустите и посмотрите лог

Во вкладке веб-формы укажите реальный UID группы и лимит 10, затем отправьте форму.

Для типовой Docker-установки примеры предлагают смотреть лог celeryworker:

docker logs swarmica-celeryworker-1 -f --tail 100

Название контейнера может отличаться. В интерфейсе скрипта доступны средства тестирования, состав которых зависит от версии. Запись return в коде сама по себе не означает, что веб-форма покажет пользователю результат: в этом учебном примере результат находится в логе.

4. Структура файла и класс Runner

Минимальный самостоятельный файл:

import logging


from runtime.runners import BaseRunner




logger = logging.getLogger(name)


class Runner(BaseRunner):

def run(self, *args, **kwargs):

data = kwargs.get("data") or {}

logger.info("Скрипт запущен; ключи параметров: %s", list(data))

Движок вызывает метод run. Основную работу выполняйте внутри него или в функциях, которые он вызывает. Не размещайте запросы к БД, сетевые вызовы или изменения данных на уровне импорта модуля: они могут произойти ещё до обработки конкретного запуска.

В исследованных примерах используются:

Полный набор методов BaseRunner здесь не описывается. Наличие перечисленных атрибутов подтверждается приложенными файлами, но это не полный контракт среды выполнения.

Обрабатывайте ожидаемые ошибки параметров явными проверками. Для неожиданных исключений используйте logger.exception(...), чтобы сохранить трассировку. Не скрывайте все ошибки общим except: pass.

5. Параметры и контекст

Три источника данных

Например, дополнительный контекст:

{
  "group_uid": "GROUP_UID",
  "hours": 48,
  "recipients": ["support-lead@example.com"],
  "apply": false
}

Python-фрагмент внутри run:

data = kwargs.get("data") or {}
group_uid = data.get("group_uid")
hours = data.get("hours", 48)
recipients = data.get("recipients", [])

Это позволяет менять группу, порог и получателей без редактирования кода. Переменная не становится параметром автоматически: скрипт должен явно прочитать её и использовать.

Различайте JSON и Python

В настройках контекст задаётся корректным JSON: true, false, null, двойные кавычки, без комментариев. В Python используются True, False, None.

Не вставляйте в JSON записи вроде ['email'], <HOURS> или # комментарий. Заполнитель GROUP_UID в примерах замените на настоящий UID.

Дополнительный контекст должен содержать JSON-совместимые значения. При этом среда может добавить к данным запуска объект модели event: весь итоговый словарь data уже не обязан быть сериализуемым в JSON.

Проверяйте типы

Текстовое поле передаёт строку. Дата может потребовать преобразования. В одном из примеров ручного запуска предусмотрена обработка одиночного значения, пришедшего списком или кортежем. Нормализуйте такие формы только для конкретных скалярных параметров; список получателей или каналов должен остаться списком.

Для логического параметра нельзя использовать bool("false"): непустая строка даст True. В примерах есть common.utils.str2bool. Можно также использовать собственную строгую функцию:

def parse_apply(value):
    if value is None:
        return False
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        normalized = value.strip().lower()
        if normalized in {"true", "1", "yes"}:
            return True
        if normalized in {"false", "0", "no", ""}:
            return False
    raise ValueError("apply должен быть логическим значением")

Неизвестное значение не должно случайно разрешать изменение данных.

Имена параметров

Не используйте для собственных настроек event и event_id: эти имена заняты данными события. Для своих параметров выбирайте понятные названия: group_uid, cf_uid, reminder_hours, recipient.

Общее объяснение контекста: статья 1351. Специфика Python-событий: статья 1614.

6. Запуск по событию

Объект события и код события

Слово event используется на двух уровнях:

События разных сущностей используют разные модели: TicketEvent, UserEvent, ArticleEvent, CustomFieldEvent, KCSEvent. Нельзя искать событие пользователя в таблице событий заявок только потому, что совпал ID.

Получение события

Следующий самостоятельный файл записывает факт смены ответственного в лог. Он поддерживает переданный объект и загрузку по ID.

import logging


from core.models import TicketEvent

from runtime.runners import BaseRunner

from swarmica import defs




logger = logging.getLogger(name)




class Runner(BaseRunner):

def run(self, *args, **kwargs):

data = kwargs.get("data") or {}

event = data.get("event")

if event is not None and not isinstance(event, TicketEvent):

logger.error("Ожидался объект TicketEvent")

return

if event is None:

event_id = data.get("event_id")

if event_id is None:

logger.warning("Не переданы event и event_id")

return

if isinstance(event_id, bool) or not isinstance(event_id, (int, str)):

logger.warning("Некорректный event_id")

return

try:

event_id = int(event_id)

except ValueError:

logger.warning("Некорректный event_id")

return

event = TicketEvent.objects.filter(id=event_id).first()

if event is None:

logger.warning("Событие не найдено")

return

if event.event != defs.TICKET_EVENTS_ASSIGNEE_CHANGE:

logger.info("Событие %s пропущено: другой тип", event.id)

return



    ticket = event.ticket
    logger.info("Смена ответственного: event_id=%s ticket_id=%s", event.id, ticket.id)

Настройка триггера

1. Создайте Действие по событию.
2. Выберите модель TicketEvent и действие типа «Скрипт».
3. Укажите подключённый скрипт.
4. Настройте событие «Смена ответственного».
5. При необходимости задайте дополнительные условия и контекст.
6. Включите действие и проверьте его на тестовой заявке.

В JSON-настройках для этого типа события в опубликованных примерах используется:

{
  "event": 10
}

В Python используйте именованную константу defs.TICKET_EVENTS_ASSIGNEE_CHANGE. Коды для настройки и другие события приведены в статье 913.

Старое и новое значения

В примерах встречаются old, new, old_value, new_value. Имена, доступные как атрибуты объекта или поля условий интерфейса, не обязательно являются полями для ORM-фильтра.

Например, счётчик назначений использует ORM-условие new__isnull=False, а пример pending-обработки переводит строковый статус в числовое значение события через defs.TICKET_EVENTS_STATUSES. Не подставляйте строковый статус заявки в любое поле события по аналогии.

Сначала проверьте описание модели и рабочий пример вашей версии. При отложенной обработке также учитывайте: событие описывает прошлое изменение, а event.ticket может уже содержать текущее состояние заявки. Перед действием заново проверьте необходимые условия.

7. Поиск объектов и связанные данные

Внутри скрипта доступны модели Django. Ниже — фрагменты для вставки в вашу логику, а не самостоятельные файлы.

Поиск одного объекта

from core.models import Group, Ticket
group = Group.objects.filter(uid="GROUP_UID").first()

ticket = Ticket.objects.filter(id=123).first()

Если объект не найден, first() возвращает None. Метод get(...) требует ровно одного результата и может вызвать исключение.

Выборка заявок

from core.models import Ticket
from swarmica import defs
tickets = Ticket.objects.filter(

status__in=[defs.TICKET_STATUSES_NEW, defs.TICKET_STATUSES_OPEN],

group__uid="GROUP_UID",

).order_by("id")

Двойное подчёркивание позволяет обращаться к связанному объекту или задавать условие: group__uid, created_at__gte, body__icontains. Получайте только нужную выборку и ограничивайте её для первых проверок.

Проверка наличия комментария

Фрагмент после получения переменной ticket:

has_comment = ticket.comments.filter(
    public=False,
    body__icontains="требуется проверка",
).exists()

public=False ограничивает поиск внутренними комментариями. icontains выполняет поиск без учёта регистра.

Последняя запись

Фрагмент для истории заявки:

last_event = ticket.events.order_by("-date", "-id").first()

При сортировке по убыванию first() возвращает самую позднюю запись, а last() — самую раннюю. Дополнительная сортировка по ID делает выбор определённее при равных датах.

Связанные объекты могут отсутствовать

requester = ticket.requester
organization = requester.organization if requester else None
schedule = organization.schedule if organization else None

Не предполагайте наличие компании, расписания или ответственного у каждой заявки. Для часто используемых связей примеры применяют select_related(...); для больших выборок можно использовать iterator(chunk_size=...). Общие методы описаны в справочнике Django QuerySet.

8. Заявки, комментарии и сообщения чата

Изменение статуса

Фрагмент после получения заявки:

from swarmica import defs
ticket.status = defs.TICKET_STATUSES_SOLVED

ticket.save()

SOLVED — статус «Решение предоставлено» / «Решено» в терминологии соответствующей версии, а не CLOSED. Надпись в тексте уведомления должна соответствовать фактическому статусу, который устанавливает код.

После save() могут выполняться модельные обработчики, события и автоматизации. Проверяйте их влияние на сценарий. Один вызов save() не доказывает эквивалентность любого изменения всем действиям штатного интерфейса.

Создание внутреннего автоматического комментария

Фрагмент с переменной ticket:

from core.models import TicketComment
from swarmica_auth.models import User
from swarmica import defs
author = User.objects.get_swarmica_service()

TicketComment.objects.create(

ticket=ticket,

author=author,

responsible=author,

body="<p>Дополнительная проверка выполнена.</p>",

public=False,

is_staff=True,

is_autocomment=True,

source=defs.TICKET_COMMENT_SOURCES_API,

)

Эти атрибуты встречаются в исследованных примерах. Для публичного ответа используется public=True. Доставка клиенту зависит от настроек каналов и обработки комментария — создание записи не следует описывать как универсальную гарантию отправки во все каналы.

Если текст содержит значения от пользователя или внешней системы, экранируйте их перед вставкой в HTML. Шаблон с разрешённой HTML-разметкой и обычный текст — разные типы входных данных.

Полный пример: добавить заметку по результату проверки

Задача: при переводе заявки в SOLVED проверить внутренние комментарии и добавить соответствующую автоматическую заметку. Этот учебный вариант по мотивам статьи 1830 дополнительно проверяет состояние заявки и повторный запуск того же события.

Подключите файл add_check_note.py, настройте событие «Статус изменён → Решение предоставлено» и контекст:

{
  "text": "требуется проверка",
  "comment_if_text": "В заявке есть запрос на дополнительную проверку.",
  "comment_ifnot_text": "Запрос на дополнительную проверку не найден."
}

import logging
from html import escape


from django.db import transaction

from core.models import Ticket, TicketComment, TicketEvent

from runtime.runners import BaseRunner

from swarmica import defs

from swarmica_auth.models import User




logger = logging.getLogger(name)




class Runner(BaseRunner):

def run(self, *args, **kwargs):

data = kwargs.get("data") or {}

event = data.get("event")

if not isinstance(event, TicketEvent):

logger.warning("Передайте объект TicketEvent через действие по событию")

return

if event.event != defs.TICKET_EVENTS_STATUS_CHANGE:

return



    names = ("text", "comment_if_text", "comment_ifnot_text")
    if not all(isinstance(data.get(k), str) and data[k].strip() for k in names):
        logger.error("Укажите непустые строки: %s", ", ".join(names))
        return
    marker = f"<!-- custom-check-note:event:{event.id} -->"
    author = User.objects.get_swarmica_service()

    with transaction.atomic():
        ticket = Ticket.objects.select_for_update().filter(id=event.ticket_id).first()
        if ticket is None or ticket.status != defs.TICKET_STATUSES_SOLVED:
            logger.info("Состояние заявки изменилось; действие пропущено")
            return
        if ticket.comments.filter(author=author, public=False, body__contains=marker).exists():
            logger.info("Событие %s уже обработано", event.id)
            return

        found = ticket.comments.filter(
            public=False,
            is_autocomment=False,
            body__icontains=data["text"].strip(),
        ).exists()
        message = data["comment_if_text"] if found else data["comment_ifnot_text"]
        body = "<p>" + escape(message.strip()).replace("\n", "<br/>") + "</p>" + marker
        TicketComment.objects.create(
            ticket=ticket, author=author, responsible=author,
            body=body, public=False, is_staff=True, is_autocomment=True,
            source=defs.TICKET_COMMENT_SOURCES_API,
        )
    logger.info("Заметка добавлена: ticket_id=%s event_id=%s", event.ticket_id, event.id)

В этом варианте поиск исключает автоматические комментарии, чтобы собственные заметки не влияли на следующие проверки. Это уточнение учебного сценария относительно исходного файла. Маркер защищает от повторения одного события при сохранении комментария; если его удалить, защиту нужно восстановить другим способом.

Сообщение веб-чата

В примере приветствия используются другая модель и привязка к сессии чата. Фрагмент внутри Runner, после получения заявки:

from core.models import ChatMessage
session = ticket.chat_sessions.last()

if session is not None:

ChatMessage.objects.create(

session=session,

public=True,

author=self.service_user,

body="Здравствуйте! Опишите, пожалуйста, ваш вопрос.",

)

Чтобы выбрать последнюю сессию, порядок связанного набора должен соответствовать ожидаемому в вашей версии. Сам исходный файл использует last() без явного порядка.

Для выбора приветствия по графику файл trigger_new_chat_greeting.py читает schedule_name, message и out_of_schedule_message; тексты — словари по языкам. Нужны существующее расписание и подходящий язык заявителя. Подробная настройка: статья 1274.

Для пустых чатов пример close_empty_chat.py сначала проверяет наличие любого комментария, затем создаёт публичный комментарий и переводит заявку в SOLVED. Наличие даже служебного комментария может остановить такой сценарий. Условия канала, статуса и времени задаются триггером; код не заменяет их настройку. См. статью 664.

9. Пользовательские поля

Чтение

После получения объекта заявки:

value = ticket.custom_fields.get_value("CUSTOM_FIELD_UID")

Этот способ используется в файле обработки CSAT. Используйте UID поля своей установки, а не значение из чужого примера. Проверяйте None отдельно от нуля и False, если они имеют разный смысл.

Запись

В примере счётчика назначений значение хранится через CustomFieldValue. Фрагмент с переменной ticket:

from custom_field.models import CustomField, CustomFieldValue
from django.contrib.contenttypes.models import ContentType
from core.models import Ticket
field = CustomField.objects.get(uid="CUSTOM_FIELD_UID")

content_type = ContentType.objects.get_for_model(Ticket)

CustomFieldValue.objects.update_or_create(

field=field,

content_type=content_type,

object_id=ticket.id,

defaults={"data": {"value": 7}},

)

ContentType указывает, к какой модели относится значение, а object_id — конкретный объект. Этот пример относится к числовому полю заявки. Для меню, списков и других типов не предполагается тот же формат значения. Уточните формат и ограничения поля перед записью.

Полный пример: число назначений заявки

Исходный файл из статьи 1333 пересчитывает события назначения с непустым новым значением. Это устойчивее простого прибавления единицы при каждом запуске: повторная обработка того же события не увеличит число снова.

Определение метрики: считаются все назначения на сотрудника, включая первое. Снятие ответственного на «Никто» не учитывается. Это не готовый коэффициент передач и не число уникальных инженеров. Если бизнесу нужны именно передачи между сотрудниками, потребуется другая выборка событий.

Создайте целочисленное поле заявки и укажите его UID в дополнительном контексте:

{
  "cf_uid": "CUSTOM_FIELD_UID"
}

Настройте событие «Смена ответственного». Учебный самостоятельный файл update_assignment_count.py:

import logging


from core.models import Ticket, TicketEvent

from custom_field.models import CustomField, CustomFieldValue

from django.contrib.contenttypes.models import ContentType

from django.db import transaction

from runtime.runners import BaseRunner

from swarmica import defs




logger = logging.getLogger(name)




class Runner(BaseRunner):

def run(self, *args, **kwargs):

data = kwargs.get("data") or {}

event = data.get("event")

if not isinstance(event, TicketEvent):

logger.warning("Передайте TicketEvent")

return

if event.event != defs.TICKET_EVENTS_ASSIGNEE_CHANGE:

return

cf_uid = data.get("cf_uid")

if not isinstance(cf_uid, str) or not cf_uid.strip():

logger.error("Не указан cf_uid")

return

field = CustomField.objects.filter(uid=cf_uid.strip()).first()

if field is None:

logger.error("Поле не найдено")

return



    content_type = ContentType.objects.get_for_model(Ticket)
    with transaction.atomic():
        ticket = Ticket.objects.select_for_update().filter(id=event.ticket_id).first()
        if ticket is None:
            return
        count = TicketEvent.objects.filter(
            ticket=ticket,
            event=defs.TICKET_EVENTS_ASSIGNEE_CHANGE,
            new__isnull=False,
        ).count()
        CustomFieldValue.objects.update_or_create(
            field=field, content_type=content_type, object_id=ticket.id,
            defaults={"data": {"value": count}},
        )
    logger.info("ticket_id=%s assignments_count=%s", event.ticket_id, count)

Перед запуском проверьте, что поле предназначено для заявки и имеет целочисленный тип. Блокировка координирует экземпляры этого учебного скрипта; другая логика, которая меняет поле без той же блокировки, требует отдельного согласования.

Инструкция и исходный файл: «Как посчитать Handover Rate».

10. Запуск по расписанию и рабочее время

Настройка

1. Подключите Python-файл.
2. Создайте Действие по расписанию типа «Скрипт».
3. Выберите файл и задайте расписание.
4. Внесите дополнительные параметры, если код их читает.
5. Включите действие после проверки на небольшой выборке.

В опубликованном примере напоминаний используется * * * * * — запуск каждую минуту. Другие значения и часовой пояс выполнения сверяйте со своей конфигурацией и справкой crontab.

Периодический запуск сам по себе не передаёт конкретную заявку. Выборку нужно сформировать в run. Не ожидайте event только потому, что класс скрипта совпадает с примером событийной обработки.

Календарное и рабочее время

Для календарного порога фрагмент выглядит так:

from datetime import timedelta
from django.utils import timezone
cutoff = timezone.now() - timedelta(hours=48)

Это 48 обычных часов. Для рабочего времени нужно использовать расписание и исключать нерабочие интервалы. Исследованные файлы обращаются к Schedule.business_time_duration(...), schedule.tz, schedule.is_business_hours(), а также методам построения рабочих интервалов. Это внутренние методы, которые нужно проверять на целевой версии.

Пример приветствия

В файле trigger_new_chat_greeting.py проверяется одна секунда с момента создания заявки:

from datetime import timedelta
start = ticket.created_at

end = start + timedelta(seconds=1)

is_working_time = schedule.business_time_duration(start=start, end=end).total_seconds() > 0

Переменные ticket и schedule должны быть получены заранее. Это проверка момента создания, а не текущего времени запуска.

Пример pending-обработки

Файл recurring_autonotify_autosolve.py показывает составной процесс:

1. Найти заявки в PENDING с расписанием компании заявителя.
2. Найти момент последнего перехода в PENDING.
3. Проверить, ответил ли клиент после этого момента.
4. Если автоматического комментария ещё нет — рассчитать срок напоминания.
5. Если комментарий один — рассчитать срок перехода в SOLVED.
6. Выполнить действие, когда наступил срок и текущее время рабочее.

Два интервала задаются параметрами reminder_hours и close_hours; по умолчанию оба равны одному часу.

Особенность приложенного файла: source_channels используется как список UID: source__uid__in=source_channels. Передавайте реальные UID каналов. Если нужна фильтрация по типам, потребуется изменить запрос; строка email не должна автоматически трактоваться как UID.

Основная выборка исключает заявки без расписания компании, хотя вспомогательные функции содержат варианты работы без него. Это различие важно при адаптации: наличие ветки в функции ещё не означает, что основной цикл передаст ей такие заявки.

Также файл считает все автоматические API-комментарии сервисного пользователя после перехода в PENDING. Другая автоматизация может повлиять на этот счётчик. В собственной реализации используйте специфический маркер процесса.

Инструкция: статья 1819. Перед рабочим запуском проверьте выходные, праздники, переход через конец рабочего дня, отсутствие интервалов и параллельный ответ клиента.

11. Уведомления и шаблоны

Выбор email-канала

Примеры отчётов используют стандартный исходящий канал:

from core.models import EmailChannel
channel = EmailChannel.objects.filter(default=True, is_deleted=False).first()

Если канала нет или send_from не задан, скрипт должен завершиться с понятным сообщением. Для собственного SMTP используется channel.get_smtp_connection(); для системного реле примеры передают connection=None.

Отправка письма

Фрагмент после проверки канала и получателей:

from core.tasks.notifications import send_email
send_email(

from_email=channel.send_from,

to=["support-lead@example.com"],

subject="Результат дополнительной проверки",

text="Проверка завершена. Подробности доступны в заявке.",

connection=None if channel.use_system_smtp_relay else channel.get_smtp_connection(),

)

Это сигнатура из приложенных примеров, а не полный справочник send_email. Запись «отправлено» в логе скрипта не доказывает доставку в почтовый ящик: проверяйте результат почтовой отправки и настройки канала.

HTML-шаблон Jinja

Движок шаблонов в файлах получается через engines["jinja2"]. Фрагмент с переменной ticket:

from django.template import engines
from core.models import Instance
template = engines["jinja2"].from_string(

"<p>Заявка #{{ ticket.id }}. Статус: {{ ticket.status }}</p>"

)

html_body = template.render(context={

"instance": Instance.get_solo(),

"ticket": ticket,

})

В письме HTML можно передать через html=html_body вместе с текстовой версией. Проверьте экранирование динамических значений и поведение выбранного шаблона. Существующий базовый шаблон notifications/base_email.html также используется в исходных файлах.

Получатели зависят от причины события

Для CSAT и других классифицированных событий удобно задавать словарь «причина → получатели» в контексте. Обезличенный пример настройки:

{
  "reason_map": {
    "productbug": ["product-owner@example.com"],
    "documentation": ["docs-owner@example.com", "support-lead@example.com"]
  },
  "survey_score_threshold": 79
}

Учебная функция выбора получателей:

def recipients_for_reasons(selected_reasons, reason_map):
    if not isinstance(reason_map, dict):
        raise ValueError("reason_map должен быть словарём")
    if not isinstance(selected_reasons, (list, tuple)):
        raise ValueError("Причины должны быть списком")
    recipients = set()
    for reason in selected_reasons:
        addresses = reason_map.get(reason, [])
        if not isinstance(addresses, list):
            raise ValueError("Получатели для каждой причины должны быть списком")
        for address in addresses:
            if not isinstance(address, str) or not address.strip():
                raise ValueError("Получатель должен быть непустой строкой")
            recipients.add(address.strip())
    return sorted(recipients)

Функция только выбирает адреса: она не валидирует email, не находит опрос и не отправляет письмо. Эти шаги нужно добавить отдельно.

CSAT-пример показывает, как связать TicketEvent, CSATSurvey, дополнительные поля и шаблон письма. Оценка в исследованном файле сравнивается с порогом на шкале 0–100; не переносите число из пятибалльного отображения напрямую. Опрос должен соответствовать именно обрабатываемому событию — выбор последней записи по заявке и автору может быть недостаточен при задержанной обработке.

Уведомления об обращениях без ответа

Файл recurring_ticket_group_notify_noanswer.py показывает выбор группы, получение email её участников, анализ последних событий заявки и отправку письма. LIMIT_HOURS, CHECK_GROUP, MAIL_FROM в нём заданы в коде, а не читаются из контекста.

При адаптации вынесите нужные значения в параметры и определите, что означает «без ответа»: время с создания, последнего ответа инженера или последнего сообщения клиента. Сам по себе возраст заявки не определяет эту метрику. Инструкция: статья 1449.

12. Отчёты CSV и Excel

Выберите смысл отчёта

Сначала определите период, часовой пояс, фильтр объектов и единицу строки. Например, отчёт по событиям должен различать текущую группу заявки и группу в момент события: фильтр ticket__group=group относится к текущей группе.

Файл из статьи 1906 выбирает события за период и формирует CSV; файл из статьи 1434 группирует созданные заявки по часам и дням недели и формирует XLSX.

Полный учебный пример: список активных заявок в CSV

Этот самостоятельный файл send_active_tickets_csv.py использует знакомые модели и отправку файла из исследованных примеров. Он экспортирует ID и статусы максимум 1 000 новых и открытых заявок одной группы.

Параметры веб-формы:

[
  {
    "name": "group_uid",
    "type": "string",
    "required": true,
    "displayName": "UID группы"
  },
  {
    "name": "recipient",
    "type": "string",
    "required": true,
    "displayName": "Email получателя"
  }
]

Для первого запуска разрешите форму только администратору. Если расширяете доступ, ограничьте допустимые группы и получателей по правилам вашей компании. Для отправки отчётов адресат не должен автоматически получать любые данные только потому, что знает UID группы.

import csv
import logging
import os
import tempfile


from django.core.validators import validate_email

from django.core.exceptions import ValidationError

from core.models import EmailChannel, Group, Ticket

from core.tasks.notifications import send_email

from runtime.runners import BaseRunner

from swarmica import defs




logger = logging.getLogger(name)

MAX_ROWS = 1000




class Runner(BaseRunner):

def run(self, *args, **kwargs):

data = kwargs.get("data") or {}

group_uid = data.get("group_uid")

recipient = data.get("recipient")

if not isinstance(group_uid, str) or not isinstance(recipient, str):

logger.error("Укажите group_uid и recipient как строки")

return

recipient = recipient.strip()

try:

validate_email(recipient)

except ValidationError:

logger.error("Некорректный email")

return

group = Group.objects.filter(uid=group_uid.strip()).first()

channel = EmailChannel.objects.filter(default=True, is_deleted=False).first()

if group is None or channel is None or not channel.send_from:

logger.error("Группа или исходящий email-канал не настроены")

return



    rows = list(Ticket.objects.filter(
        group=group,
        status__in=[defs.TICKET_STATUSES_NEW, defs.TICKET_STATUSES_OPEN],
    ).order_by("id").values_list("id", "status")[:MAX_ROWS + 1])
    if len(rows) > MAX_ROWS:
        logger.error("Выборка больше %s строк; сузьте отчёт", MAX_ROWS)
        return

    path = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w", encoding="utf-8-sig", newline="",
            suffix=".csv", delete=False,
        ) as output:
            path = output.name
            writer = csv.writer(output, delimiter=";")
            writer.writerow(["Ticket ID", "Status"])
            writer.writerows(rows)

        send_email(
            from_email=channel.send_from,
            to=[recipient],
            subject="Активные заявки группы",
            text=f"В приложении {len(rows)} записей.",
            attachment_files=[path],
            connection=None if channel.use_system_smtp_relay else channel.get_smtp_connection(),
        )
        logger.info("Отчёт передан на отправку; строк=%s", len(rows))
    finally:
        if path is not None and os.path.exists(path):
            os.remove(path)

Пустая выборка даст CSV с заголовком. При превышении лимита скрипт не отправляет усечённый отчёт. Временный файл удаляется в finally.

Этот порядок очистки соответствует прямому вызову send_email в исследованных файлах. Если меняете реализацию на асинхронную очередь, файл должен существовать до чтения обработчиком; немедленное удаление может сломать отправку.

XLSX

Для Excel-файла используется openpyxl. Учебная функция, принимающая список строк или другой итерируемый набор:

from openpyxl import Workbook
def save_xlsx(rows, path):

workbook = Workbook(write_only=True)

sheet = workbook.create_sheet("Заявки")

sheet.append(["Ticket ID", "Status"])

for ticket_id, status in rows:

sheet.append([ticket_id, status])

workbook.save(path)

Для свободного текста из внешних источников учитывайте возможность интерпретации значения как формулы в табличном редакторе. В приведённом примере экспортируются только ID и фиксированный статус.

В часовом отчёте применяются ExtractHour, ExtractWeekDay, Count, Case, When. Группировка выполняется в запросе; затем строки превращаются в Excel-таблицу. Часовой пояс передаётся в функции извлечения времени. Инструкция: статья 1434.

Даты и часовые пояса

Для сравнения дат используйте значения с часовым поясом. Для отчёта за целые дни удобно задать полуоткрытый интервал: от начала первого дня включительно до начала дня после последнего исключительно.

Самостоятельная вспомогательная функция для строк YYYY-MM-DD:

from datetime import date, datetime, time, timedelta
from zoneinfo import ZoneInfo
def report_period(date_from, date_to, timezone_name="Europe/Moscow"):

start_day = date.fromisoformat(date_from)

end_day = date.fromisoformat(date_to)

if end_day < start_day:

raise ValueError("Конец периода раньше начала")

tz = ZoneInfo(timezone_name)

start = datetime.combine(start_day, time.min, tzinfo=tz)

end_exclusive = datetime.combine(end_day + timedelta(days=1), time.min, tzinfo=tz)

utc = ZoneInfo("UTC")

return start.astimezone(utc), end_exclusive.astimezone(utc)

В запросе используйте date__gte=start и date__lt=end_exclusive, либо аналогичные условия для created_at. Для специальных исторических переходов часового пояса отдельно определите правила неоднозначного времени. Общая справка: Python zoneinfo.

Инструкция по исходному отчёту переходов статуса: статья 1906. При адаптации дополнительно ограничьте выборку типом события «Смена статуса» и проверьте ORM-поля старого и нового значения на своей версии.

13. Работа с внешними API

Запрос и обработка ответа

В миграционном примере используется requests. В своей реализации задайте тайм-аут, проверьте HTTP-статус и формат ответа. Следующий фрагмент иллюстрирует вызов; переменные url, headers и payload определяются вашим сценарием:

import requests
response = requests.post(url, json=payload, headers=headers, timeout=(5, 30))

response.raise_for_status()

result = response.json()

json=payload кодирует словарь как JSON. timeout ограничивает ожидание соединения и чтения. Ответ может быть не JSON даже при успешном HTTP-статусе — обработайте это отдельно. Подробнее: официальная документация Requests.

Для сетевых ошибок предусмотрите понятный результат: повторить позже, оставить запись для ручного разбора или остановить сценарий. Повторное создание внешнего объекта после тайм-аута может дать дубликат, если первая попытка фактически сработала. Используйте ключ идемпотентности, когда его поддерживает внешняя система.

Постраничная загрузка

Мигратор статей читает results и переходит по next. Ниже учебная функция для API с таким форматом. Она ограничивает число страниц и не пересылает заголовки авторизации на другой узел.

from urllib.parse import urljoin, urlsplit
import requests


def iter_api_results(start_url, headers, max_pages=100):

origin = urlsplit(start_url)

if origin.scheme != "https" or not origin.netloc or origin.username or origin.password:

raise ValueError("В этом примере требуется HTTPS URL без учётных данных")

url = start_url

visited = set()

page_count = 0



with requests.Session() as session:
    while url:
        parsed = urlsplit(url)
        if (parsed.scheme, parsed.netloc) != (origin.scheme, origin.netloc):
            raise ValueError("next ведёт на другой узел")
        if url in visited or page_count >= max_pages:
            raise ValueError("Цикл пагинации или превышен лимит страниц")
        visited.add(url)
        page_count += 1
        response = session.get(
            url, headers=headers, timeout=(5, 30), allow_redirects=False,
        )
        if 300 <= response.status_code < 400:
            raise ValueError("Перенаправление требует отдельной проверки")
        response.raise_for_status()
        page = response.json()
        if not isinstance(page, dict) or not isinstance(page.get("results"), list):
            raise ValueError("Ожидался объект с массивом results")
        for item in page["results"]:
            yield item
        next_url = page.get("next")
        if next_url is not None and not isinstance(next_url, str):
            raise ValueError("next должен быть строкой или null")
        url = urljoin(url, next_url) if next_url else None

В том же файле, внутри run, после получения проверенных host и token:

headers = {"Authorization": f"Token {token}"}
for category in iter_api_results(f"{host.rstrip('/')}/api/categories/", headers):
    logger.info("Получена категория id=%s", category.get("id"))

Схема Token соответствует использованию API-токена в исследованном миграторе Swarmica. Для другой системы способ авторизации может отличаться. Здесь HTTPS — ограничение учебной функции, а не описание всех вариантов установки Swarmica.

Не помещайте реальные секреты в публикуемый код, клиентские виджеты и лог. Способ хранения и выдачи серверного секрета согласуйте с администратором установки.

Перенос статей: почему два прохода

Файл onetime_hc_migration.py выполняет работу в два этапа:

1. Получает категории и создаёт или находит соответствующие локальные категории.
2. Создаёт статьи-заготовки с внешними идентификаторами.
3. Строит соответствие ID исходных и локальных статей.
4. Загружает вложения и переводы, заменяет ссылки на локальные адреса.

Такой подход нужен, когда статьи ссылаются друг на друга и новые ID заранее неизвестны. refresh_migrated разрешает обновление ранее перенесённой статьи; в исходном файле перед обновлением удаляются её переводы и вложения. Это операция замены, которую нужно проверять на резервной копии.

Основная загрузка исходного файла ограничена visibility=ALL. Она не является универсальным переносом всех вариантов сегментации и доступа. Инструкция: статья 306.

При адаптации храните карты соответствий внутри конкретного запуска. Состояние в глобальных переменных модуля может сохраниться, если среда переиспользует модуль между запусками.

14. Импорт данных

Импорт обычно объединяет чтение файла, проверку колонок, нормализацию значений, поиск существующего объекта, создание или обновление и итоговую статистику.

Файлы и контейнеры

Примеры CSV используют settings.UPLOADS_ROOT. Markdown-импорт ожидает каталог /swarmica/swarmica/old_articles, подключённый к контейнеру.

Файл должен быть доступен именно процессу, который выполняет скрипт. Копирование в контейнер django помогает только при общей файловой системе с нужным обработчиком либо при запуске в этом контейнере. Проверьте тома вашей установки, а не только имя каталога.

Не принимайте произвольный абсолютный путь из формы без ограничения. Учебная функция для файла непосредственно в разрешённом каталоге:

from pathlib import Path
def resolve_csv_file(upload_root, filename):

if not isinstance(filename, str) or not filename.strip():

raise ValueError("Укажите имя файла")

root = Path(upload_root).resolve()

candidate = (root / filename.strip()).resolve()

if candidate.parent != root:

raise ValueError("Файл должен находиться непосредственно в каталоге uploads")

if candidate.suffix.lower() != ".csv" or not candidate.is_file():

raise ValueError("CSV-файл не найден")

return candidate

Чтение CSV

Фрагмент с полученным путём path:

import csv
with open(path, encoding="utf-8-sig", newline="") as source:

reader = csv.DictReader(source, delimiter=";")

required = {"Сотрудник", "E-Mail"}

if not required.issubset(set(reader.fieldnames or [])):

raise ValueError("Нужны колонки Сотрудник и E-Mail")

for line_number, row in enumerate(reader, start=2):

name = row["Сотрудник"].strip()

email = row["E-Mail"].strip().lower()

# Проверка строки, поиск объекта и обработка выполняются здесь.

utf-8-sig учитывает возможную BOM-метку. Разделитель и названия колонок должны соответствовать вашему файлу. Фрагмент намеренно не создаёт пользователей.

Пользователи с ролью «Сотрудник»

Файл import_users.py:

• Читает имя CSV из параметра filename.
• Ожидает колонки Сотрудник и E-Mail, разделённые ;.
• Проверяет существующего пользователя по email и email-идентификатор.
• Создаёт пользователя с defs.USER_ROLES_INTERNAL_USER.
• Вызывает set_unusable_password(); вход требует отдельной настройки пароля.
• Считает созданные и пропущенные записи.

Роль «Сотрудник» в этом примере не равна автоматически роли «Агент». Не меняйте роль по названию файла без проверки вашей модели доступа. Инструкция: статья 1741.

Компании и клиенты

Файл import_organizations_and_clients.py ожидает client;email;organization, поддерживает ряд синонимов колонок и использует внутренние сервисы data_import со схемами данных.

Это полезный пример выделения импорта в отдельный слой. Но его правила объединения объектов специфичны: организация ищется по имени и внешнему ID, существующий пользователь из другой организации пропускается. Для вашей задачи заранее определите, как обрабатывать совпадающие имена, разные внешние ID и смену компании.

В прочитанной версии существующий пользователь без компании не получает компанию из CSV: ветка считает запись уже обработанной. Если вам нужна такая привязка, добавьте отдельную проверенную логику. Инструкция: статья 1753.

Статьи из Markdown

Файл article_import.py:

• Читает .md непосредственно из указанного каталога.
• Использует имя файла без расширения как тему статьи.
• Находит или создаёт категорию по имени и языку.
• Создаёт Article и ArticleTranslation.
• По умолчанию использует русский язык и статус UNAPPROVED.

В нём нет поиска уже импортированной статьи: повторный запуск создаёт новые статьи. Для повторяемого импорта добавьте устойчивый ключ источника и правило обновления. Импорт текста также не означает автоматический перенос всех локальных картинок и связанных файлов. Инструкция: статья 1661.

Отключение сигналов в миграционных примерах

Некоторые импорты используют factory.django.mute_signals(...). Это специальная техника для массового переноса, которая отключает часть модельных обработчиков в процессе выполнения. Она может повлиять на регистрацию событий и связанные действия; в общем процессе сигналы могут быть важны и для других задач.

Не переносите такой декоратор в обычную автоматизацию ради ускорения или защиты от повторов. Для собственной реализации сначала определите, какие штатные обработчики нужны, и проверьте последствия изменения способа записи.

15. Массовые операции и предварительный просмотр

Обязательные этапы

Для операции по списку объектов:

1. Разберите и проверьте параметры.
2. Найдите все объекты и покажите отсутствующие ID.
3. Проверьте допустимые статусы и другие ограничения.
4. Вычислите, что изменится.
5. Выполните изменения выбранным способом.
6. Запишите итог: найдено, изменено, пропущено, ошибок.

Исходный set_group_for_tickets.py реализует этот подход: читает tickets и group_uid, проверяет наличие всех заявок и статусы SOLVED / CLOSED, затем меняет группу только там, где она отличается.

Транзакция

Связанные изменения можно объединить в transaction.atomic(). Если из блока выходит исключение, изменения БД откатываются. Внешнее письмо или HTTP-запрос таким откатом не отменяются. См. документацию Django по транзакциям.

Фрагмент с заранее определёнными ticket_id и group:

from django.db import transaction
from core.models import Ticket
with transaction.atomic():

ticket = Ticket.objects.select_for_update().get(id=ticket_id)

if ticket.group_id != group.id:

ticket.group = group

ticket.save()

select_for_update() координирует изменения одной строки между транзакциями с совместимой схемой блокировки. Сетевые запросы внутри долгой транзакции задерживают освобождение блокировок, поэтому планируйте порядок работы отдельно.

save и массовая запись

save() и bulk_update() отличаются. В исходном примере смены группы применяется bulk_update: такой метод не вызывает save() каждой модели и стандартные сигналы pre_save / post_save. Поэтому не предполагается, что он создаст те же события, что ручная операция в интерфейсе. См. раздел bulk_update в Django.

Выбирайте способ записи по требуемому поведению, а не только по скорости. Аналогично проверяйте, что требуется при изменении пользовательского поля напрямую через CustomFieldValue.

apply и режим без изменений

Для административных действий полезен параметр apply, по умолчанию false. Это приём проектирования собственного скрипта; движок не добавляет предварительный просмотр автоматически.

Самостоятельный пример preview_group_move.py использует только изменение группы и по умолчанию ничего не записывает:

import logging


from core.models import Group, Ticket

from django.db import transaction

from runtime.runners import BaseRunner

from swarmica import defs




logger = logging.getLogger(name)




def parse_apply(value):

if value is None:

return False

if isinstance(value, bool):

return value

if isinstance(value, str):

value = value.strip().lower()

if value in {"true", "1", "yes"}:

return True

if value in {"false", "0", "no", ""}:

return False

raise ValueError("Некорректный apply")




class Runner(BaseRunner):

def run(self, *args, **kwargs):

data = kwargs.get("data") or {}

try:

apply = parse_apply(data.get("apply"))

raw_ids = data.get("tickets", "")

if not isinstance(raw_ids, str):

raise ValueError("tickets должен быть строкой")

ticket_ids = sorted({int(x.strip()) for x in raw_ids.split(",") if x.strip()})

if not ticket_ids or len(ticket_ids) > 100 or any(x <= 0 for x in ticket_ids):

raise ValueError("Укажите от 1 до 100 положительных ID")

group_uid = data.get("group_uid")

if not isinstance(group_uid, str) or not group_uid.strip():

raise ValueError("Укажите group_uid")

except ValueError as error:

logger.error("Ошибка параметров: %s", error)

return



    group = Group.objects.filter(uid=group_uid.strip()).first()
    if group is None:
        logger.error(&quot;Группа не найдена&quot;)
        return

    with transaction.atomic():
        tickets = list(Ticket.objects.select_for_update().filter(id__in=ticket_ids).order_by(&quot;id&quot;))
        if {t.id for t in tickets} != set(ticket_ids):
            logger.error(&quot;Часть заявок не найдена; изменений нет&quot;)
            return
        allowed = {defs.TICKET_STATUSES_SOLVED, defs.TICKET_STATUSES_CLOSED}
        if any(t.status not in allowed for t in tickets):
            logger.error(&quot;Допустимы только SOLVED и CLOSED; изменений нет&quot;)
            return
        changes = [t for t in tickets if t.group_id != group.id]
        logger.info(&quot;Найдено=%s к изменению=%s apply=%s&quot;, len(tickets), len(changes), apply)
        if not apply:
            return
        for ticket in changes:
            ticket.group = group
            ticket.save()
    logger.info(&quot;Операция завершена&quot;)

Контекст для тестового запуска:

{
  "tickets": "101, 102",
  "group_uid": "GROUP_UID",
  "apply": false
}

Для веб-формы добавьте tickets и group_uid типа string, а apply типа boolean. Изменение apply на true разрешает запись. Код намеренно использует save() вместо bulk_update() из исходного файла; последствия для событий и автоматизаций нужно проверить отдельно.

Проверка предварительного просмотра не фиксирует состояние навсегда: между просмотром и применением данные могут измениться. Поэтому проверки повторяются при каждом запуске. Исходная инструкция по операции: статья 1887.

16. Повторные запуски и параллельная обработка

Не предполагайте, что событие всегда обработается ровно один раз, а периодические задачи никогда не пересекутся. Даже ручной повторный запуск может создать дубли.

Проверка «комментария ещё нет» и последующая запись должны выполняться согласованно, иначе два процесса увидят отсутствие комментария одновременно. Блокировка заявки в учебном примере заметки помогает, если все экземпляры процесса используют ту же схему.

Для другой логики могут потребоваться отдельная запись состояния или уникальное ограничение. Один transaction.atomic() без блокировки и без уникального ключа не гарантирует отсутствие дублей.

Защита от циклов

Скрипт может породить событие, на которое настроен он сам или другая автоматизация. Например, комментарий создаёт событие комментария, изменение поля — событие поля.

Ограничивайте тип события, проверяйте автора и маркер, меняйте только отличающееся значение. Расчёт метрики назначения должен запускаться по назначению, а не по любому изменению заявки. Проверьте цепочку на тестовой установке.

17. Диагностика и эксплуатация

Что записывать в лог

• Название, UID и версию скрипта.
• ID события и объекта.
• Причину пропуска или остановки.
• Количество найденных, изменённых и ошибочных записей.
• Трассировку неожиданного исключения.

Пример фрагмента обработки одного объекта:

try:
    process_ticket(ticket)
except Exception:
    logger.exception("Ошибка обработки ticket_id=%s", ticket.id)

process_ticket здесь — ваша собственная функция. Для независимых объектов можно продолжать обработку после ошибки одного из них. Для единой операции «всё или ничего» исключение должно приводить к откату, а не скрываться внутри транзакции.

Типичные проблемы

Как проверить поля модели на своей версии

Если пример вызывает FieldError, можно временно вывести имена полей модели в тестовом диагностическом скрипте. Фрагмент для размещения внутри run после импорта модели:

from core.models import TicketEvent
field_names = sorted(field.name for field in TicketEvent._meta.get_fields())

logger.info("Поля TicketEvent: %s", field_names)

Это имена полей и связей ORM, а не все свойства и методы объекта. Наличие атрибута в описании контекста не делает его фильтруемым полем. Для моделей или методов, которых нет в документации, передайте техподдержке конкретную операцию, свою версию и ошибку; не подбирайте названия вслепую.

Производительность и частота запуска

Начинайте с ограниченной выборки. Вынесите повторяющиеся запросы из цикла, загрузите необходимые связи заранее, агрегируйте данные в БД, если это соответствует задаче. Не сохраняйте весь большой набор в память без необходимости.

Частота расписания должна соответствовать длительности обработки. Если задача может выполняться дольше интервала, предусмотрите координацию запусков. Сетевые ожидания ограничивайте тайм-аутами; временные файлы очищайте.

Добавление файлового обработчика логов при каждом импорте модуля может привести к дублированию записей. Для первого скрипта достаточно стандартного logging.getLogger(__name__).

Что проверить перед рабочим запуском

1. Корректные, отсутствующие и неверно типизированные параметры.
2. Пустую выборку и отсутствующие связанные объекты.
3. Повторное выполнение того же события.
4. Параллельное выполнение и изменение объекта пользователем.
5. Ошибку внешней системы или SMTP.
6. Рабочие часы, праздники и часовой пояс для временной логики.
7. События и другие автоматизации после записи данных.
8. Доступ к форме и допустимые данные / действия.
9. Объём обработки и длительность запуска.

Храните исходный код в системе контроля версий, фиксируйте проверенную версию Swarmica и владельца сопровождения. Перед обновлением проверяйте расширения в тестовой установке. Для диагностики подготовьте имя и версию скрипта, способ запуска, обезличенные параметры, ID тестового объекта и трассировку ошибки.

18. Каталог примеров и дальнейшее изучение

Ниже — общедоступные статьи с приложенными Python-файлами, на которых можно изучать отдельные приёмы. Файл из статьи — исходный пример для проверки и адаптации, а не автоматически готовое расширение под любой процесс.

Рекомендуемый маршрут:

1. Запустите первый скрипт из раздела 3 и проверьте параметры веб-формы.
2. Настройте скрипт, который только пишет событие в лог.
3. На тестовой заявке попробуйте пользовательское поле или внутреннюю заметку.
4. Разберите пример по расписанию и определите правила повторного запуска.
5. Добавьте отчёт или интеграцию под свой конкретный процесс.
6. Согласуйте сопровождение, ограничения доступа и проверку перед обновлениями.

Другие инструкции находятся в категории автоматизации. Для запуска через API или командную строку используйте схему вашей версии и согласованную инструкцию: это руководство не задаёт универсальную команду CLI или endpoint для всех версий.