Cluster · разработчикам

API модуля

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

app содержит методы ядра. Они асинхронные, кроме app.stop(). Сам объект передаётся при загрузке модуля, создавать его не нужно. Макет класса и обработчиков находится в MODULES.md.

Состав app:

Атрибут	Назначение
`db`	Данные экземпляра
`config`	Значения конфигурации
`core`	Состояние юзербота и конфигуратор модулей
`client`, `bot`, `inline`	Telegram, личный бот и формы с кнопками
`files`	Файлы
`tasks`	Расписание
`modules`	Функции и события других модулей
`web`	HTTPS и WSS
`requests`	Входящие HTTP-запросы
`ui`	Значения страниц модуля
`exchange`	Обмен юзерботов
`logs`	Журнал

Права задаются в описании пакета и принимаются пользователем. Если право отсутствует, вызов недоступен. При действии человека также учитываются овнерка, доверка и доступ обработчика. Формат прав находится в PERMISSIONS.md.

Типы данных

В аргументах и результатах используются JSON: строки, конечные числа, bool, None, списки и словари со строковыми ключами. Байты передаются отдельно через файлы или base64 там, где это предусмотрено методом.

Для подсказок можно описать JSON так:

from typing import TypeAlias, TypedDict

JSON: TypeAlias = None | bool | int | float | str | list["JSON"] | dict[str, "JSON"]

class File(TypedDict):
    id: str
    unit: str
    instance: str
    filename: str
    size: int
    sha256: str | None
    written: int
    state: str
    created: float
    expires: float | None
    ttl: int | None

class FilesPage(TypedDict):
    files: list[File]
    after: str | None

Это описание формы данных для документации, не новые экспорты SDK. File показывает основные поля записи; ядро может добавить служебные поля. У готового файла state="ready", sha256 содержит контрольную сумму.

Данные

await app.db.get(key, default=None, module=None, slot=None)
await app.db.set(key, value, ttl=None, module=None, slot=None)
await app.db.delete(key, module=None, slot=None)
await app.db.keys(module=None, slot=None, after=None, limit=100)

key: str, value: JSON, default: JSON.
ttl: int | None: срок в секундах, от 1 до 315360000. None означает без срока.
module: str | None, slot: str | None: раздел, исходно текущий экземпляр.
after: str | None: курсор из предыдущего ответа, limit: int от 1 до 1000.

get возвращает значение либо default, если записи нет или срок истёк. set возвращает True. delete возвращает bool, была ли запись. keys возвращает {"keys": ["..."], "after": "..."}. Курсор передаётся без изменений, None завершает просмотр. Пустая порция с курсором возможна, если выбранные записи уже истекли.

Нужны db.read для получения и списка, db.write для записи и удаления. Доступ к другому разделу своего юзербота дополнительно требует db.full. Данные другого юзербота через эти методы недоступны.

Одна запись вместе с ключом и служебными полями занимает не больше половины предела сообщения ядра, исходно 512 KiB. Запись учитывается в квоте.

class Module:
    async def init(self) -> None:
        count = await app.db.get("count", default=0)
        await app.db.set("count", count + 1)

Файлы

await app.files.upload(filename, data, ttl=None)
app.files.chunks(id)
await app.files.list(after=None, limit=100)
await app.files.delete(id)

filename: str: подпись файла, до 255 символов, без управляющих символов.
data: bytes: содержимое для загрузки.
ttl: int | None: срок готового файла в секундах, от 1 до 315360000.
id: str: ID файла Cluster.
after: str | None: курсор списка, limit: int от 1 до 100.

upload возвращает описание готового файла. chunks является асинхронным генератором bytes, части до 64 KiB. list возвращает FilesPage, delete возвращает bool.

Нужны files.read для получения и списка, files.write для загрузки и удаления. Для чужого раздела своего юзербота требуется files.full. Отдельно выбранный в конфигураторе файл можно получить без files.full; это не даёт права изменять его или читать остальные файлы.

Имя является подписью, путь сервера из него не строится. Размер и 2048 байт метаданных резервируются до загрузки. На юзербота разрешено до 10000 файлов. SHA256 проверяется при завершении. Для незаконченной загрузки срок считается от создания: переданный ttl либо 3600 секунд, если он не указан. После успешного завершения срок готового файла отсчитывается заново; без ttl он хранится до удаления. Истёкшие записи удаляются при очередной очистке ядра.

Следующую порцию списка нужно получать, пока after не станет None. Проверка доступа может оставить порцию пустой с непустым курсором.

class Module:
    async def init(self) -> None:
        file = await app.files.upload("note.txt", b"Hello", ttl=3600)
        size = 0

        async for part in app.files.chunks(file["id"]): size += len(part)
        await app.logs.write(str(size))

Для загрузки без хранения всего файла в памяти есть files/create, files/write, files/finish через app.request. create принимает имя, размер, необязательные SHA256 и срок. write принимает ID, последовательный offset и base64-часть до 64 KiB после декодирования. finish проверяет размер и сумму. При ручной загрузке автор также отвечает за удаление незавершённого файла при ошибке.

app.client.download_media(chat, message, maximum=10485760) сохраняет вложение Telegram в этот же раздел. Нужны telegram.read и files.write; для последующего получения частей также files.read. Размер задаётся в байтах, проверяются реальный чат, сообщение, автор, текущие права и подключение. При отмене или ошибке частичный файл и резерв удаляются. Подробности отправки и получения находятся в TELEGRAM.md.

При ошибке app.files.upload SDK пытается удалить незавершённый файл и освободить резерв. Если соединение потеряно или право отозвано, очистка может быть недоступна: незавершённую запись удалит ядро после истечения срока незавершённой загрузки. Ошибка очистки не заменяет исходную ошибку загрузки. Готовые временные файлы автор удаляет сам, обычно в finally.

Расписание

await app.tasks.create(handler, at, values=None, interval=None, retry=0, request=None)
await app.tasks.get(id)
await app.tasks.list()
await app.tasks.cancel(id)
await app.tasks.delete(id)

handler: str: имя метода с @loader.task(fields=...).
at: int | float: Unix-время запуска, не в прошлом и не дальше 315360000 секунд от текущего момента.
values: dict | None: поля обработчика, исходно {}, до 16 KiB.
interval: int | None: повтор через указанное число секунд, от 1 до 315360000.
retry: int: от 0 до 10 повторов при ошибке.
request: str | None: ID создания для защиты от повтора.
id: str: ID созданной задачи.

Все методы требуют tasks.create. create, get и cancel возвращают запись задачи. list() возвращает список задач экземпляра, без аргументов. delete возвращает True.

Основные поля записи: id, instance, handler, values, at, interval, retry, attempt, state, result, error. Задача может ожидать выполнения (pending), выполняться (running), завершиться (done), получить ошибку (failed) или быть отменена (cancelled).

from __future__ import annotations
import time
from typing import TYPE_CHECKING

if TYPE_CHECKING: from router import Event

class Module:
    @loader.task(fields={"value": {"type": "integer", "required": True}})
    async def save(self, ctx: Event) -> None:
        await app.db.set("value", ctx.values["value"])

    async def init(self) -> None:
        await app.tasks.create("save", time.time() + 60, {"value": 5})

Описание задачи должно совпадать с interface.tasks пакета. request хранится 30 суток и занимает квоту. Повтор с тем же ID возвращает ту же задачу только при совпадении всех параметров. Если задача уже удалена, старый ID до очистки записи повторно не используется. После истечения записи ID можно использовать заново. Для повторного запроса нужно сохранить и исходное at, новое time.time() изменит параметры.

До 1000 сохранённых задач на юзербота. Завершённые тоже занимают место до удаления. Для каждой резервируется 64 KiB, чтобы завершение и отмена оставались доступны при заполненном хранилище.

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

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

Функции и события модулей

await app.modules.call(module, function, values=None, slot=None, timeout=30)
await app.modules.emit(module, signal, values=None, slot=None)

module: str является ID модуля, function: str и signal: str именами обработчиков. values: dict | None содержит их поля. slot: str | None выбирает экземпляр. timeout: int | float ограничивает ожидание результата в SDK.

@loader.function(fields=...) принимает вызов и возвращает JSON либо None. call ждёт этот результат, при ошибке задания поднимает RuntimeError. Истечение timeout поднимает TimeoutError и прекращает ожидание, уже начавшееся действие от этого не отменяется. Отдельного метода app.modules.result в SDK нет: получение состояния выполняет сам call. @loader.signal(fields=...) принимает событие, emit возвращает запись постановки в очередь с request.

from __future__ import annotations
from typing import TYPE_CHECKING

if TYPE_CHECKING: from router import Event

class Module:
    @loader.function(fields={"value": {"type": "integer", "required": True}})
    async def double(self, ctx: Event) -> int:
        return ctx.values["value"] * 2

Вызывающему нужен modules.call с разрешённым ID модуля. Если экземпляр один, он выбирается автоматически. Если их несколько, выбор задаётся в настройках зависимости или slot. Выключенная зависимость возвращает ошибку доступности.

Вызов сохраняет происхождение и права человека, даже после завершения исходного обработчика. Ограничения каждого участника остаются в силе. Цикл и цепочка длиннее восьми переходов отклоняются.

Вебклиент

await app.web.request(url, method="GET", headers=None, body=None, binary=False)
await app.web.profile(proxy=None, cookies=True, domain=None)
await app.web.download(url, filename, maximum=10485760)
await app.web.upload(url, file, method="POST", headers=None)

url: str: HTTPS-адрес на порту 443.
method: str: GET, POST, PUT, PATCH, DELETE, HEAD; для upload только POST, PUT, PATCH.
headers: dict[str, str] | None: до 32 заголовков. Служебные заголовки соединения менять нельзя.
body: str | None: до 64 KiB UTF-8.
binary: bool: вернуть тело ответа в base64.
proxy: str | None: URL HTTP, SOCKS4 или SOCKS5-прокси.
cookies: bool: сохранять cookies профиля.
domain: str | None: сайт для проверки права при изменении профиля.
maximum: int: резерв загрузки в байтах, исходно 10 MiB.
filename: str, file: str: имя создаваемого файла и ID отправляемого файла.

request возвращает {"status": 200, "headers": {...}, "body": "...", "binary": false}. Статус HTTP, в том числе 4xx или 5xx, сам по себе не становится исключением SDK. Исходный предел тела ответа 256 KiB. Заголовок set-cookie не возвращается модулю, cookies обрабатываются профилем. profile возвращает True.

Нужен web.request с разрешённым именем сайта или web.full. Имена в правах задаются без схемы и пути. Проверяются все адреса DNS, локальные, частные и служебные адреса закрыты. Соединение использует проверенный IP, TLS проверяется по исходному имени. Автоматического перехода по перенаправлению нет: новый URL проходит отдельную проверку.

У каждого экземпляра свой профиль cookies и прокси. Адрес прокси тоже должен входить в принятые права. Профиль с секретами хранится зашифрованно, сохраняется до 100 cookies.

download возвращает готовый файл Cluster, требует files.write и резервирует maximum до загрузки. При ошибке резерв и частичный файл удаляются. upload требует files.read, отправляет байты готового файла как тело и не строит multipart. Результат отправки содержит status и строковое body до 64 KiB; заголовки в этом ответе не возвращаются.

import json

class Module:
    async def init(self) -> None:
        result = await app.web.request("https://example.org/data.json")
        if result["status"] == 200:
            await app.db.set("response", json.loads(result["body"]))

WSS

await app.web.open(url)
await app.web.send(id, data)
await app.web.receive(id, wait=10)
await app.web.close(id)

url: str имеет схему wss и порт 443. open возвращает объект с id: str. send принимает data: str, до 64 KiB UTF-8, и возвращает True. receive принимает wait: int от 0 до 10 секунд, возвращает {"data": "...", "binary": false} либо None после ожидания. Двоичные ответы имеют binary=True и base64 в data. Отправка двоичных сообщений этим методом не предусмотрена. close возвращает True.

До 16 соединений на юзербот, сообщение до 64 KiB. Соединение принадлежит экземпляру и закрывается при остановке или отзыве прав. Действуют те же ограничения сайтов и адресов, что у HTTPS. Прямого сетевого сокета у контейнера нет.

Входящие запросы

await app.requests.create(handler)
await app.requests.revoke(handler)

handler: str является именем метода с @loader.request(fields=...). Нужен requests.manage. create возвращает {"path": "/api/hooks/...", "token": "..."}. Ключ выдаётся один раз, в базе остаётся хеш. Повторное создание заменяет ключ. revoke возвращает bool.

from __future__ import annotations
from typing import TYPE_CHECKING

if TYPE_CHECKING: from router import Event

class Module:
    @loader.request(fields={"text": {"type": "string", "required": True}})
    async def save(self, ctx: Event) -> bool:
        await app.db.set("incoming", ctx.values["text"])
        return True

    async def init(self) -> None:
        endpoint = await app.requests.create("save")
        await app.db.set("endpoint", endpoint)

Внешний клиент отправляет POST на полученный путь с Authorization: Bearer TOKEN и полями JSON. Ответ содержит request, ID результата. POST по тому же пути с добавленным /REQUEST_ID получает state, result, error; нужен тот же ключ.

До 64 KiB тела, до 60 обращений в минуту по адресу источника для юзербота. Такой вызов не получает личность овнера, выполняется только с правами модуля. Ключ не стоит публиковать или писать в журнал. В примере он сохраняется в разделе самого модуля; доступ к этому разделу также нужно учитывать при выборе прав других модулей.

Журнал

await app.logs.write(message, level="info")

message: str, до 4096 байт UTF-8. level: str: info, warn или error. Результат содержит id, time, level, group, instance, message.

Нужен logs.write. До 100 записей в минуту от экземпляра. Записи занимают квоту, срок хранения задаётся в ядре или для модуля. Ядро скрывает очевидные токены и записи вида password=..., но собственные секреты автор всё равно не включает в сообщения.

Прямой запрос

await app.request(path, **data)
app.stop()

path: str передаётся позиционно, data содержит именованные аргументы маршрута. request возвращает его результат, не внешнюю оболочку {"ok": ..., "result": ...}. При отказе ядра поднимается RuntimeError с кодом ошибки. Для Telegram, Bot API и форм есть дополнительный результат операции; их обёртки SDK сами проверяют его и возвращают результат операции.

Запросы одного app выполняются последовательно. Строка запроса и строка ответа вместе с переводом строки ограничены 1 MiB со стороны SDK. Для отдельных методов действуют более низкие пределы. Значения NaN и бесконечность не передаются. Текущее событие добавляется SDK автоматически.

stop() возвращает None и ставит признак остановки цикла обработчиков. Текущий обработчик не прерывается, новое событие после завершения цикла не запрашивается.

Описание страниц и конфигурации: MODULES.md. Telegram и личный бот: TELEGRAM.md. Полные наборы методов, конструкторы и формы: BUILDERS.md. Обмен юзерботов: EXCHANGE.md. Маршруты и точные параметры ядра: ROUTES.md.

Состояние юзербота и настройки модулей

app.core.info() требует core.info. Возвращает name, version, prefix, uptime в секундах, modules и running. Время считается от запуска службы ядра. Почта, подписка, сессия Telegram и адреса файлов в ответ не входят.

class Module:
    async def status(self) -> str:
        info = await app.core.info()
        return f"{info['name']}: {info['running']} / {info['modules']}"

Для конфигуратора других модулей предусмотрено право modules.configure. Ограничение modules содержит ID публикаций, а не имена или экземпляры. Без ограничения доступны настройки всех установленных модулей этого юзербота.

await app.core.modules(offset=0)
await app.core.configuration(instance)
await app.core.configure(instance, values, revision)
await app.core.commands(instance)
await app.core.aliases(instance, values, revision)

offset: int: смещение списка. Ответ содержит modules и next; next=None завершает список. В списке находятся модули с настройками или командами. Одна порция содержит до 16 записей с instance, name, state.
instance: str: ID публикации и профиль экземпляра через /.
configuration возвращает fields, values, revision, instance, prefix, commands; последнее поле показывает, есть ли команды. Значения секретных полей заменены на {"set": bool}.
values: dict[str, JSON]: изменяемые поля. Ядро проверяет тип, границы и условия по схеме модуля. Остальные значения сохраняются.
revision: int: версия конфигурации, полученная при просмотре. При одновременной правке через сайт устаревший запрос получает conflict.

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

commands возвращает имена команд в виде fields, values, revision, prefix. Ключ поля соответствует ID команды, значение соответствует её текущему имени. aliases принимает изменяемые имена и ревизию модуля. Ядро проверяет совпадения с другими командами и запрещает служебные exec и load. Сброс на объявленное имя удаляет переопределение. Права на выполнение команды остаются прежними.

Общие настройки требуют отдельного права core.configure:

settings = await app.core.settings()
await app.core.preferences({"prefix": ".", "notify_update": False}, settings["revision"])

settings возвращает fields, values, revision, prefix, instance="core/main". Через preferences можно менять только:

prefix: str: от 1 до 4 символов без пробелов;
language: str: ru или en, язык уведомлений;
log_retention: int | None: срок хранения журнала в секундах, от 60 до 315360000; None снимает срок;
notify_module, notify_update, notify_info, notify_bot: переключатели обычных уведомлений и доставки в Telegram.

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

Готовый пример находится в Core Utils. В Cluster он устанавливается по умолчанию. После подключения личного бота конфигуратор доступен по команде cfg; в Telegram используется текущий префикс юзербота. Если имя уже занято другим модулем, выбирается corecfg или свободное имя с номером. Удалённый пользователем экземпляр автоматически не возвращается.

Команда cfg открывает выбор общих настроек или настроек модулей. У каждого модуля есть отдельный пункт с именами команд. Меню, кнопки и подсказки доступны на русском и английском. Язык берётся из общих настроек юзербота и меняется сразу после выбора. Названия полей стороннего модуля берутся из его переводов label; текст без переводов отображается как задан автором. Меню приходит от личного бота; при запуске из «Избранного» команда меняется на уведомление об открытии. Переключатели и варианты меняются кнопками, текст и числа отправляются следующим сообщением в течение трёх минут. /cancel отменяет ввод. Списки, JSON, секреты и файлы настраиваются на сайте.

Изменение вложений

app.client.edit_file(chat, message, file, caption="") добавляет или заменяет вложение собственного сообщения Telegram. Требует telegram.send для чата и files.read для готового файла Cluster. chat и message являются числовыми ID, file является строковым ID файла. Подпись ограничена 1024 байтами. Пути сервера и внешние ссылки вместо ID не принимаются.

app.bot.edit_caption(chat, message_id, caption) меняет подпись сообщения личного бота. Требует bot.send для чата. Для изменения текста без вложения остаётся edit_message.

Пример Ping показывает баннер, ник, версию ядра, время работы, число модулей, префикс и задержку. Для собственного сообщения заменяет содержимое, для входящего отвечает новым сообщением. При запрете картинок или нехватке места остаётся текстовый ответ. При сетевой ошибке с неизвестным результатом повторная отправка не выполняется. Задержка измеряется по одному действию отправки или изменения сообщения вместе с обращением к ядру; подготовка баннера в неё не входит. Ограничения чатов и овнерка продолжают действовать.

Для пакета Ping нужны main.py, manifest.json и banner.png. Для Core Utils нужны main.py и manifest.json. Все файлы одного пакета помещаются в корень архива.