#1157: Написание плагинов для автоматизации действий на Python

Содержание

Часть I. Основы

1. Что можно программировать и какой механизм выбрать.

2. Как устроена среда выполнения.

3. Минимальный скрипт и класс Runner.

4. Параметры, типы данных и дополнительный контекст.

5. Первый рабочий скрипт: загрузка, веб-форма и проверка результата.

Часть II. Триггеры запуска

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

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

Часть III. Работа с данными

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

9. Заявки, комментарии и чаты.

10. Пользовательские поля и счётчик назначений.

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

12. Массовые операции, транзакции и предварительный просмотр.

Часть IV. Действия и интеграции

13. Уведомления, шаблоны и выбор получателей.

14. Отчёты CSV и Excel.

15. Внешние API и постраничная загрузка.

16. Импорт пользователей, компаний и статей.

Часть V. Эксплуатация

17. Диагностика, производительность и эксплуатация.

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

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

Загружать скрипты и настраивать их запуск через Веб-форму, Действия по событию, Действия по расписанию могут администраторы системы в разделе Настройки > Автоматизация и маршрутизация > Скрипты

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

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

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

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

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

Обозначения в руководстве

• Django — стандартный фреймворк. Такие методы и модули описаны в официальной документации Django.
• Swarmica — внутренние модели, хелперы и сервисные объекты продукта. Их сигнатуры могут меняться между версиями; сверяйтесь с тестовой установкой и рабочими примерами из статей.
• Если не указано иное, перед использованием любого фрагмента получайте все переменные (ticket, group, schedule, channel) до строки с примером.

Часть I. Основы

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

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

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

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

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

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

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

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

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

Скрипты выполняются внутри контейнера, поэтому у них нет доступа к аппаратному узлу / виртуальной машине, на которой установлена Swarmica. Однако сценарии имеют доступ к общим разделам диска и основной базе данных, используемой Swarmica, поэтому будьте осторожны при выполнении потенциально опасных операций, таких как удаление и изменение любых данных или файлов.

Кроме того, ресурсы памяти/процессора используются совместно со всем инсталляцией Swarmica, поэтому настоятельно рекомендуется минимизировать импорт библиотек и модулей до только необходимого минимального набора функций. Например, вместо импорта всего модуля import datetime импортируйте только необходимую вам функцию: from datetime import timedelta.

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

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

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

Библиотеки

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

Ряд предустановленных библиотек доступен для импорта и использования в ваших скриптах. Смотрите список доступных методов в документации соответствующей библиотеки:

import re               # Для работы с регулярными выражениями
import os               # Для работы с вещами ОС, такими как пути к файлам и т.д.
import bs4              # Для парсинга HTML, XML и т.д. с помощью BeautifulSoup
import csv              # Для чтения и записи CSV
import json             # Для парсинга JSON
import time             # Для работы с текущим временем, например, генерации временных меток
import pandas           # Для работы с наборами данных
import urllib           # Для парсинга и записи параметров URL, urlencode и т.д.
import dateutil         # Для работы с датами, например, парсинга даты из строки
import datetime         # Для работы с объектами datetime, например, вычисления длительностей и т.д.
import openpyxl         # Для чтения и записи Excel
import requests         # Для работы с удаленными API и вебхуками

Обзор движка

Есть несколько способов запуска сценария:

[Триггер события] ---> +context --\
                                  \
[Расписание] --------> +context----\
                                    >----> Runner.run(kwargs={data: **params, **context})
[CLI] -------------> +params-------/
                                  /
[API/WEB] ---------> +params-----/

Итак, в основном, если сценарий времени выполнения запускается Триггером События или по Расписанию, то данные события и пользовательские контекстные данные, настроенные в соответствующем триггере / расписании, передаются в параметры сценария.

Допустим, сценарий запускается событием UserEvent, тогда контекст можно разобрать следующим образом:

class Runner(BaseRunner):
    def run(self, *args, **kwargs):
        data = kwargs.get('data', {})
        if not data or 'event_id' not in data:
            logger.error(f"No event_id passed, exiting")
            return
        event = UserEvent.objects.filter(id=data['event_id']).first()
        if not event:
            logger.error(f"No event with {data['event_id']} found, exiting")
            return

Если мы настроим Триггер События на передачу дополнительного контекста в скрипт, например, списка email-адресов для получения оповещения при изменении учетной записи пользователя, то он также будет передан в скрипт.

При условии, что пользовательский контекст Триггера События выглядит так:

{
  "recipients": [
    "user1@domain.tld",
    "user2@domain.tld"
  ]
}

Теперь вы можете получить доступ к значению переменной внутри скрипта следующим образом:

data = kwargs.get('data', None)

if data:
    recipients = data.get('recipients', None)

Доступ к объектам Swarmica

Обычно вам нужно манипулировать моделями и наборами запросов (querysets) для объектов Swarmica в ваших скриптах. Поскольку Swarmica использует Django в качестве фреймворка, большинство методов моделей и наборов запросов также работают здесь.

Смотрите полную справку по методам моделей Django и наборов запросов на официальном сайте документации:

Querysets

Models

Большинство объектов, которые вам могут понадобиться, находятся в пакете core, за исключением объектов User - они находятся в пакете swarmica_auth.

Также вам, как правило, потребуется импортировать общие константы, используемые Swarmica, следующим образом:

from swarmica import defs

print(defs.USER_ROLES_INTERNAL)

3. Минимальный скрипт и класс Runner

Любой скрипт — это Python-файл, в котором объявлен класс Runner, унаследованный от runtime.runners.BaseRunner, и метод run. Всё, что делает скрипт, размещается в run или в вызываемых из него функциях.

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

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(f"Скрипт запущен; ключи параметров: {list(data)}")

Этот скрипт ничего не читает из БД и не меняет данные. Он просто пишет в лог ключи, которые пришли в data. Его удобно использовать как «Hello, world» и как первую проверку подключения: запустить, посмотреть в лог, убедиться, что параметры доходят.

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

Что доступно из Runner

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

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

Обработка ошибок

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

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

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

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

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

{
  "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", [])

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

Веб-форма с датами

Вы также можете настроить скрипт так, чтобы у него была собственная веб-форма, которая может принимать, проверять и передавать параметры в скрипт.

Допустим, мы пишем скрипт, который генерирует пользовательский отчет в Excel и отправляет его списку email-получателей. И мы хотим, чтобы пользователи передавали даты начала и окончания отчета и список получателей, разделенный запятыми.

В веб-интерфейсе Swarmica в настройках скрипта мы настраиваем следующие параметры веб-формы:

[
  {
    "name": "start_date",
    "type": "datetime",
    "readonly": false,
    "displayName": "Report start date"
  },
  {
    "name": "end_date",
    "type": "datetime",
    "readonly": false,
    "displayName": "Report end date"
  },
  {
    "name": "recipients",
    "type": "text",
    "displayName": "Comma-separated recipients"
  }
]

Теперь мы можем получить доступ к этим параметрам в скрипта следующим образом:

params = kwargs.get('data', {})
if params:
    if 'recipients' in params:
        recipients = params.get('recipients')
        if recipients:
            recipients = [x.strip() for x in recipients.split(',') if '@' in x]
        if not recipients:
            logger.info("Input error: empty/invalid list of 'recipients' specified")
            return
        
        start_date = params.get('start_date', None)
        if isinstance(start_date, str) and start_date.strip():
            report_start_date = parse_datetime(start_date)
            
        end_date = params.get('end_date', None)
        if isinstance(end_date, str) and end_date.strip():
            report_end_date = parse_datetime(end_date)

        if not report_end_date:
            report_end_date = now().replace(hour=23, minute=30, second=0, microsecond=0).replace(tzinfo=default_time_zone)
        if not report_start_date:
            report_start_date = report_end_date - timedelta(days=1)

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

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

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

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

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

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

Для логического параметра нельзя использовать bool("false"): непустая строка даст True. В примерах есть хелпер Swarmica 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.

5. Первый рабочий скрипт: загрузка, веб-форма и проверка результата

Это следующий шаг после «Hello, world»: читаем данные из БД, ничего не меняем, пишем результат в лог.

Задача. Найти группу по UID и показать ограниченный список новых и открытых заявок.
Триггер. Веб-форма.
Параметры. group_uid, limit (1–100).
Ожидаемый результат. Записи вида ticket_id=… status=… assignee_uid=… в логе и итоговая строка показано=N.

Шаг 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):
        LOG_PREFIX = f"{self.script.name} ({self.script.uid})"
        data = kwargs.get("data") or {}
        if not isinstance(data, dict):
            logger.error(f"{LOG_PREFIX}: data должен быть словарём")
            return

        group_uid = data.get("group_uid")
        if not isinstance(group_uid, str) or not group_uid.strip():
            logger.error(f"{LOG_PREFIX}: укажите group_uid")
            return

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

        group = Group.objects.filter(uid=group_uid.strip()).first()
        if group is None:
            logger.warning(f"{LOG_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(
                f"{LOG_PREFIX}: ticket_id={ticket.id} status={ticket.status} "
                f"assignee_uid={assignee_uid}"
            )
            shown += 1
        logger.info(f"{LOG_PREFIX}: версия={VERSION} показано={shown}")

Что здесь важного для дальнейшего:

• LOG_PREFIX = f"{self.script.name} ({self.script.uid})" — этот приём стоит копировать во все скрипты. Название скрипта в каждой строке лога облегчает поиск при нескольких активированных расширениях.
• Валидация параметров — это не паранойя. Текстовое поле формы всегда приходит строкой, а int("") или bool могут сломать логику.
• Если заявок нет, скрипт штатно завершится с показано=0. Это нормально, не ошибка.
• select_related("assignee") избавляет от N+1 запросов: без него каждый ticket.assignee — отдельный SQL.

Шаг 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, затем отправьте форму.

Результат выполнения скрипта можно посмотреть в логе celeryworker:

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

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

Часть II. Триггеры запуска

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

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

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

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

Имена полей внутри событий могут отличаться между моделями. В одних событиях инициатор — responsible, в других — user или actor. Перед обращением к полю сверяйтесь с моделью вашей версии (см. приём с _meta.get_fields() в разделе 17).

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

Следующий самостоятельный файл записывает факт смены ответственного в лог. Он поддерживает переданный объект и загрузку по 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):
        LOG_PREFIX = f"{self.script.name} ({self.script.uid})"
        data = kwargs.get("data") or {}
        event = data.get("event")
        if event is not None and not isinstance(event, TicketEvent):
            logger.error(f"{LOG_PREFIX}: ожидался объект TicketEvent")
            return
        if event is None:
            event_id = data.get("event_id")
            if event_id is None:
                logger.warning(f"{LOG_PREFIX}: не переданы event и event_id")
                return
            if isinstance(event_id, bool) or not isinstance(event_id, (int, str)):
                logger.warning(f"{LOG_PREFIX}: некорректный event_id")
                return
            try:
                event_id = int(event_id)
            except ValueError:
                logger.warning(f"{LOG_PREFIX}: некорректный event_id")
                return
            event = TicketEvent.objects.filter(id=event_id).first()
        if event is None:
            logger.warning(f"{LOG_PREFIX}: событие не найдено")
            return
        if event.event != defs.TICKET_EVENTS_ASSIGNEE_CHANGE:
            logger.info(f"{LOG_PREFIX}: событие {event.id} пропущено: другой тип")
            return

        ticket = event.ticket
        logger.info(
            f"{LOG_PREFIX}: смена ответственного: "
            f"event_id={event.id} ticket_id={ticket.id}"
        )

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

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

В 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. Запуск по расписанию и рабочее время

Настройка

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 обычных часов. Для рабочего времени нужно использовать расписание и исключать нерабочие интервалы. Исследованные файлы обращаются к методам Swarmica 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. Перед рабочим запуском проверьте выходные, праздники, переход через конец рабочего дня, отсутствие интервалов и параллельный ответ клиента.

Часть III. Работа с данными

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

Внутри скрипта доступны модели 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. Получайте только нужную выборку и ограничивайте её для первых проверок.

Выборка заявок с фильтром по датам

Давайте выведем ID заявки, тему и имя назначенного исполнителя для всех новых и открытых заявок, созданных между указанными датами начала и окончания:

from swarmica import defs
from core.models import Ticket

class Runner(BaseRunner):
    def run(self, *args, **kwargs):
        params = kwargs.get('data', {})
        if not params:
            return
            
        start_date = params.get('start_date', None)
        
        if isinstance(start_date, str) and start_date.strip():
            report_start_date = parse_datetime(start_date)
        end_date = params.get('end_date', None)
        
        if isinstance(end_date, str) and end_date.strip():
            report_end_date = parse_datetime(end_date)

        if not report_end_date:
            report_end_date = now().replace(hour=23, minute=30, second=0, microsecond=0).replace(tzinfo=default_time_zone)
        if not report_start_date:
            report_start_date = report_end_date - timedelta(days=1)
            
        tickets = Ticket.objects.filter(created_at__range=(report_start_date,report_end_date), 
                                        status__in=[defs.TICKET_STATUSES_NEW, defs.TICKET_STATUSES_OPEN])
        
        for ticket in ticket:
            print(f"#{ticket.id}: {ticket.subject} ({ticket.assignee.name})")

Работа с пользователями

Давайте отключим все email-уведомления для заблокированных пользователей:

from swarmica import defs
from swarmica_auth.models import User

class Runner(BaseRunner):
    def run(self, *args, **kwargs):
        
        users = User.objects.filter(role=defs.USER_ROLES_BLOCKED)
        
        for user in users:
            user.email_notifications.clear()

Оптимизация связей

• select_related("assignee") — для ForeignKey и OneToOne. Убирает N+1, потому что делает JOIN.
• prefetch_related("comments") — для ManyToMany и обратных ForeignKey. Делает отдельный запрос и сшивает результат в Python.

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

Для очень больших наборов данных используйте iterator(chunk_size=...) — он читает строки порциями и не держит весь результат в памяти:

for ticket in Ticket.objects.filter(...).iterator(chunk_size=500):
    ...

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

Фрагмент после получения переменной 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.

9. Заявки, комментарии и чаты

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

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

from swarmica import defs

ticket.status = defs.TICKET_STATUSES_SOLVED
ticket.save()

TICKET_STATUSES_SOLVED — статус «Решение предоставлено» , а не TICKET_STATUSES_CLOSED. Надпись в тексте уведомления должна соответствовать фактическому статусу, который устанавливает код.

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

Про save(update_fields=[...]). Если вы ограничиваете запись конкретными полями (ticket.save(update_fields=["status"])), Django не будет вызывать pre_save/post_save для остальных полей, а Swarmica-автоматизации могут не сработать так же, как при полном save(). Это не ошибка, но поведение отличается от обычного сохранения. Проверьте на тестовой заявке, прежде чем использовать update_fields.

Один вызов 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,
)

User.objects.get_swarmica_service() — хелпер Swarmica, а не Django. Он может бросить исключение, если сервисный пользователь Swarmica не сконфигурирован (по умолчанию создается в процессе установки Swarmica). В production-скрипте оборачивайте его в try/except и пишите в лог понятную причину.

Атрибуты is_staff, is_autocomment, source встречаются в исследованных примерах. Для публичного ответа используется 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):
        LOG_PREFIX = f"{self.script.name} ({self.script.uid})"
        data = kwargs.get("data") or {}
        event = data.get("event")
        if not isinstance(event, TicketEvent):
            logger.warning(f"{LOG_PREFIX}: передайте объект 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(f"{LOG_PREFIX}: укажите непустые строки: {', '.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(f"{LOG_PREFIX}: состояние заявки изменилось; действие пропущено")
                return
            if ticket.comments.filter(author=author, public=False, body__contains=marker).exists():
                logger.info(f"{LOG_PREFIX}: событие {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(
            f"{LOG_PREFIX}: заметка добавлена: "
            f"ticket_id={event.ticket_id} event_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.

10. Кастомные поля и счётчик назначений

Чтение

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

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

custom_fields.get_value(...) — хелпер Swarmica, а не Django ORM. Используйте 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 и update_or_create — стандартный Django. CustomField / CustomFieldValue — модели Swarmica.

Важно про defaults={"data": {...}}. update_or_create полностью перезаписывает data. Если поле хранит не только value, но и другие ключи (например, служебные метаданные rich-text, идентификаторы вложений), они потеряются. Перед записью проверьте фактическое содержимое data у поля вашего типа и, если нужно, объединяйте словари.

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

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

Задача. Пересчитать число назначений заявки на сотрудника и записать в пользовательское поле.
Триггер. Событие «Смена ответственного».
Параметры. cf_uid — UID целочисленного поля заявки.
Ожидаемый результат. В поле записано актуальное число; повторная обработка того же события не увеличивает счётчик.

Исходный файл из статьи 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):
        LOG_PREFIX = f"{self.script.name} ({self.script.uid})"
        data = kwargs.get("data") or {}
        event = data.get("event")
        if not isinstance(event, TicketEvent):
            logger.warning(f"{LOG_PREFIX}: передайте 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(f"{LOG_PREFIX}: не указан cf_uid")
            return
        field = CustomField.objects.filter(uid=cf_uid.strip()).first()
        if field is None:
            logger.error(f"{LOG_PREFIX}: поле не найдено")
            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(f"{LOG_PREFIX}: ticket_id={event.ticket_id} assignments_count={count}")

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

Про блокировку. select_for_update() здесь блокирует строку Ticket от параллельного изменения другими такими же запусками скрипта. Важно понимать, чего блокировка не делает:

• она не мешает параллельной вставке новых TicketEvent;
• она не защищает значение поля от записи другим кодом (другим скриптом, ручной правкой в интерфейсе);

Итоговое значение будет корректным до тех пор, пока все изменения поля проходят через этот же скрипт с той же блокировкой. Как только появляется второй писатель — договоритесь об общем правиле или используйте отдельный механизм координации.

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

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

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

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

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

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

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

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

12. Массовые операции, транзакции и предварительный просмотр

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

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

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.

Отдельно: save(update_fields=[...]) — ещё один случай. Он сохранит только перечисленные поля, и pre_save/post_save для остальных полей не сработают. Swarmica-автоматизации, привязанные к изменению других полей, могут не запуститься.

Выбирайте способ записи по требуемому поведению, а не только по скорости. Аналогично проверяйте, что требуется при изменении пользовательского поля напрямую через 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):
        LOG_PREFIX = f"{self.script.name} ({self.script.uid})"
        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(f"{LOG_PREFIX}: ошибка параметров: {error}")
            return

        group = Group.objects.filter(uid=group_uid.strip()).first()
        if group is None:
            logger.error(f"{LOG_PREFIX}: группа не найдена")
            return

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

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

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

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

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

Часть IV. Действия и интеграции

13. Уведомления, шаблоны и выбор получателей

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

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

from core.models import EmailChannel

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

EmailChannel — модель Swarmica. Если канала нет или 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 — хелпер Swarmica из core.tasks.notifications. Это сигнатура из приложенных примеров, а не полный справочник. Запись «отправлено» в логе скрипта не доказывает доставку в почтовый ящик: проверяйте результат почтовой отправки и настройки канала.

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,
})

engines — Django. Instance.get_solo() — хелпер Swarmica.

В письме 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.

14. Отчёты CSV и Excel

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

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

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

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

Задача. Выгрузить в CSV максимум 1 000 новых и открытых заявок одной группы и отправить файл письмом.
Триггер. Веб-форма.
Параметры. group_uid, recipient.
Ожидаемый результат. CSV с колонками Ticket ID;Status, приложенный к письму с указанного адреса.

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

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

[
  {
    "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):
        LOG_PREFIX = f"{self.script.name} ({self.script.uid})"
        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(f"{LOG_PREFIX}: укажите group_uid и recipient как строки")
            return
        recipient = recipient.strip()
        try:
            validate_email(recipient)
        except ValidationError:
            logger.error(f"{LOG_PREFIX}: некорректный 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(f"{LOG_PREFIX}: группа или исходящий 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(f"{LOG_PREFIX}: выборка больше {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(f"{LOG_PREFIX}: отчёт передан на отправку; строк={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-поля старого и нового значения на своей версии.

15. Внешние 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()
            try:
                page = response.json()
            except ValueError as error:
                raise ValueError(
                    f"Ответ не является JSON (Content-Type={response.headers.get('Content-Type')!r})"
                ) from error
            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(f"Получена категория id={category.get('id')}")

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

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

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

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

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

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

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

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

16. Импорт пользователей, компаний и статей

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

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

Примеры 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, поддерживает ряд синонимов колонок и использует внутренние сервисы Swarmica data_import со схемами данных.

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

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

Статьи из Markdown

Файл article_import.py:

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

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

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

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

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

Часть V. Эксплуатация

17. Диагностика, производительность и эксплуатация

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

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

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

try:
    process_ticket(ticket)
except Exception:
    logger.exception(f"ошибка обработки ticket_id={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(f"Поля TicketEvent: {field_names}")

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

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

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

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

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

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

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

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

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

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

Шаблон скрипта

Скачать шаблон sample.py

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

1. Запустите минимальный скрипт из раздела 3 и убедитесь, что параметры доходят.
2. Прогоните первый рабочий скрипт из раздела 5 на тестовой группе.
3. Настройте скрипт, который только пишет событие в лог (раздел 6).
4. На тестовой заявке попробуйте пользовательское поле или внутреннюю заметку (разделы 9–10).
5. Разберите пример по расписанию и определите правила повторного запуска (разделы 7, 11).
6. Добавьте отчёт или интеграцию под свой конкретный процесс (разделы 13–16).
7. Согласуйте сопровождение, ограничения доступа и проверку перед обновлениями (разделы 17–18).

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