Что такое FunPayCardinal и где в нём плагин

FunPayCardinal (FPC) — бот-автоматизатор для FunPay. Внутри он состоит из трёх слоёв:

• FunPayAPI — обёртка над сайтом FunPay: Account (запросы), Runner (long-polling обновлений), types (модели данных), events (события).
• Cardinal — ядро: держит конфиги, аккаунт, списки хэндлеров, автоподнятие лотов, автовыдачу, ЧС.
• tg_bot — Telegram-панель управления на pyTelegramBotAPI.

Плагин — это обычный модуль Python, который Cardinal импортирует на старте и у которого забирает списки функций-хэндлеров. Никакого базового класса наследовать не нужно: достаточно объявить нужные переменные уровня модуля.

Раздел 02

Быстрый старт

От пустого файла до плагина, который отвечает в чат FunPay и умеет команду в Telegram.

Минимальный плагин

Создайте файл plugins/hello.py:

from __future__ import annotations
from typing import TYPE_CHECKING
import logging

if TYPE_CHECKING:
    from cardinal import Cardinal
    from FunPayAPI.updater.events import NewMessageEvent

logger = logging.getLogger("FPC.hello")

NAME = "Hello"
VERSION = "1.0.0"
DESCRIPTION = "Отвечает на слово 'привет' в чате FunPay."
CREDITS = "@yourname"
UUID = "0f2d5a17-9c4b-4a4e-b8de-1b7a5c3f0e21"   # свой уникальный UUID4!
SETTINGS_PAGE = False
BIND_TO_DELETE = None


def on_message(c: Cardinal, e: NewMessageEvent):
    if e.message.author_id in (0, c.account.id):
        return                                  # системное / своё сообщение
    if "привет" in (e.message.text or "").lower():
        c.send_message(e.message.chat_id, "И тебе привет!", e.message.chat_name)


BIND_TO_NEW_MESSAGE = [on_message]

Это всё. Положите файл в plugins/, перезапустите бота — плагин появится в Telegram-ПУ в разделе «Плагины».

Установка, логи, отладка

• Установка: файл кладётся в plugins/ вручную или отправляется боту в Telegram (команда /upload_plugin / кнопка в списке плагинов).
• Перезапуск обязателен. Загрузка плагинов происходит один раз в Cardinal.init(). Горячей перезагрузки нет — после правки файла перезапускайте FPC (/restart).
• Логи: берите логгер с префиксом FPC. — logging.getLogger("FPC.МойПлагин"). Тогда сообщения попадут в общий лог и в /logs. В логах поддерживаются цветовые метки вида $YELLOW…$RESET.
• Ошибки хэндлеров не роняют бота. Cardinal.run_handlers() ловит любое исключение, пишет ошибку в лог и идёт дальше. Полный traceback виден только на уровне DEBUG.
• Отключить файл от загрузки — первой строкой файла поставить комментарий со словом noplug: # noplug. Такой файл будет проигнорирован (удобно для вспомогательных модулей рядом с плагинами).

Раздел 03

Анатомия плагина

Что именно ядро читает из вашего модуля и в каком порядке.

Обязательные поля модуля

Если хотя бы одного из этих семи имён нет — плагин не загрузится (FieldNotExistsError).

Как FPC загружает плагин

1. Берёт все *.py из plugins/ (не рекурсивно).
2. Читает первую строку: если это комментарий со словом noplug — пропускает файл.
3. importlib исполняет модуль целиком. Весь код верхнего уровня выполняется здесь, ещё до инициализации аккаунта.
4. Читает семь обязательных полей, проверяет UUID.
5. Создаёт PluginData и кладёт в cardinal.plugins[UUID]; состояние «включён» подтягивается из storage/cache/disabled_plugins.json.
6. После загрузки всех плагинов вызывает add_handlers(): у каждой функции из BIND_TO_* проставляется атрибут plugin_uuid, функции складываются в общие списки ядра.

Выключение плагина в ПУ не выгружает модуль: ядро просто пропускает его хэндлеры при вызове (plugin_uuid сверяется с cardinal.plugins[uuid].enabled). Поэтому:

def is_enabled(c: Cardinal) -> bool:
    pl = c.plugins.get(UUID)
    return bool(pl and pl.enabled)

Плагин из нескольких файлов

Ядро считает плагином каждый .py в plugins/, но перед загрузкой добавляет эту папку в sys.path. Значит, вспомогательные модули можно класть рядом и импортировать по имени — главное пометить их noplug, чтобы они не грузились как отдельные плагины:

plugins/
├── myshop.py            # сам плагин с NAME/UUID/BIND_TO_*
├── myshop_api.py        # первая строка: # noplug
└── myshop_ui.py         # первая строка: # noplug

# myshop.py
from myshop_api import fetch_prices
from myshop_ui import build_settings_kb

Внешние данные — шаблоны, БД, изображения — держите в storage/plugins/, а не рядом с кодом: содержимое plugins/ пользователь чистит и переустанавливает.

Раздел 04

Жизненный цикл и хуки

Шестнадцать точек привязки. Каждая — просто список функций на уровне модуля.

Таблица BIND_TO_*

Порядок вызова на старте

Cardinal.init()
 ├─ регистрируются встроенные handlers.py и announcements.py
 ├─ load_plugins()          # исполняется код ваших модулей
 ├─ add_handlers()          # BIND_TO_* попадают в ядро
 ├─ создаётся TGBot, регистрируются встроенные разделы ПУ
 ├─ ▶ PRE_INIT              # ваш код: tg.msg_handler / cbq_handler / команды
 ├─ setup_commands(), запуск потока Telegram-бота
 ├─ авторизация аккаунта, Runner, загрузка профиля и лотов
 └─ ▶ POST_INIT             # ваш код: запросы к FunPay, стартовые проверки

Cardinal.run()
 ├─ поток Runner.loop()
 ├─ ▶ PRE_START → ▶ POST_START
 ├─ потоки автоподнятия лотов и обновления сессии
 └─ process_events()        # бесконечный цикл: события → ваши хэндлеры

Несколько хэндлеров на одном событии

# Порядок вызова = порядок в списке; сначала встроенные, потом плагины
BIND_TO_NEW_MESSAGE = [log_handler, main_handler]
BIND_TO_NEW_ORDER = [on_order]
BIND_TO_POST_START = [start_worker]

Хэндлеры выполняются синхронно и последовательно в потоке обработки событий. Долгая работа в хэндлере тормозит весь бот — выносите её в поток (см. Практики).

Раздел 05

Объект Cardinal

Первый аргумент любого хэндлера. Через него доступно всё: аккаунт, конфиги, Telegram, ЧС, профиль, другие плагины.

Основные атрибуты

Свойства-переключатели

Читаются из _main.cfg и удобны для проверок: autoraise_enabled, autoresponse_enabled, autodelivery_enabled, multidelivery_enabled, autorestore_enabled, autodisable_enabled, old_mode_enabled, keep_sent_messages_unread, bl_delivery_enabled, bl_response_enabled, block_tg_login и другие.

Секции _main.cfg

delay = int(c.MAIN_CFG["Other"]["requestsDelay"])
on = c.MAIN_CFG["FunPay"].getboolean("autoDelivery")

c.MAIN_CFG["Other"]["watermark"] = "🐦 my shop"      # запись
c.save_config(c.MAIN_CFG, "configs/_main.cfg")

Методы, которые нужны чаще всего

Правильный способ написать в чат FunPay. В отличие от account.send_message(), он:

• добавляет водяной знак из конфига (watermark=False отключает);
• режет текст на куски по 20 строк — FunPay не принимает длинные сообщения;
• понимает разметку: $photo=ID — отправить изображение, $sleep=1.5 — пауза в секундах, $new — принудительно начать новое сообщение;
• делает до attempts повторов при ошибке.

c.send_message(e.message.chat_id, "Спасибо за покупку!", e.message.chat_name)
c.send_message(chat_id, "$photo=1234567890")          # только картинка

Достаёт полный Order из события с кэшированием и защитой от гонок: повторный вызов на том же объекте не сделает второй запрос. Для Message ID заказа вытаскивается из текста системного сообщения. Всегда используйте это вместо ручного account.get_order() в хэндлерах.

order = c.get_order_from_object(e.order)     # NewOrderEvent
if order and order.review and order.review.stars <= 2:
    ...

Регистрирует команды плагина: они появятся в карточке плагина, а при True в третьем элементе — и в меню команд бота. Кнопка «Команды» в карточке отрисовывается только если список непустой, поэтому вызывать метод стоит даже ради одной команды. Сам обработчик команды это не создаёт — его всё равно нужно зарегистрировать через tg.msg_handler(handler, commands=[...]).

cardinal.add_telegram_commands(UUID, [
    ("mystats", "статистика плагина", True),
    ("myreset", "сброс счётчиков", False),
])

Вызывает список хэндлеров с перехватом ошибок. Пригодится, если ваш плагин сам порождает событие (так делает тест автовыдачи: собирает фейковый NewOrderEvent и вызывает c.run_handlers(c.new_order_handlers, (c, fake_event))).

from cardinal import Cardinal, get_cardinal

c = get_cardinal()      # None, если экземпляр ещё не создан

Раздел 06

События

Второй аргумент хэндлера. Все классы — в FunPayAPI.updater.events, типы — в FunPayAPI.common.enums.EventTypes.

У каждого события есть type (EventTypes), time и runner_tag — тег запроса Runner'а, по которому можно понять, что несколько событий пришли одной пачкой.

Два режима получения сообщений

В _main.cfg есть параметр oldMsgGetMode. Он определяет, какой хук у вас вообще будет срабатывать:

def on_msg(c: Cardinal, e):
    obj = e.message if hasattr(e, "message") else e.chat
    text = obj.text if hasattr(obj, "text") else obj.last_message_text
    chat_id, chat_name = obj.chat_id if hasattr(obj, "chat_id") else obj.id, obj.chat_name if hasattr(obj, "chat_name") else obj.name
    ...

BIND_TO_NEW_MESSAGE = [on_msg]
BIND_TO_LAST_CHAT_MESSAGE_CHANGED = [on_msg]

Фильтрация сообщений

В хук прилетает всё: ваши собственные сообщения, системные уведомления FunPay, сообщения бота. Минимальный набор проверок:

from FunPayAPI.common.enums import MessageTypes

def on_message(c: Cardinal, e: NewMessageEvent):
    m = e.message
    if m.author_id == 0:            # системное сообщение FunPay
        return
    if m.author_id == c.account.id: # наше собственное
        return
    if m.by_bot:                    # отправлено этим же ботом
        return
    if m.type is not MessageTypes.NON_SYSTEM:
        return                      # отзывы, оплаты, возвраты и т.д.

Типы системных сообщений (MessageTypes): ORDER_PURCHASED, ORDER_CONFIRMED, NEW_FEEDBACK, FEEDBACK_CHANGED, FEEDBACK_DELETED, NEW_FEEDBACK_ANSWER, FEEDBACK_ANSWER_CHANGED, FEEDBACK_ANSWER_DELETED, ORDER_REOPENED, REFUND, PARTIAL_REFUND, ORDER_CONFIRMED_BY_ADMIN, REFUND_BY_ADMIN, DISCORD, DEAR_VENDORS, NON_SYSTEM.

MessageEventsStack

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

def on_message(c: Cardinal, e: NewMessageEvent):
    if e.stack and e.message.id != e.stack.get_stack()[-1].message.id:
        return                       # отвечаем только на последнее в пачке
    full_text = "\n".join(ev.message.text or "" for ev in e.stack.get_stack())

ID заказа из текста сообщения

from FunPayAPI.common.utils import RegularExpressions

res = RegularExpressions()
found = res.ORDER_ID.findall(str(e.message))   # ['#ABCD1234']
if found:
    order = c.account.get_order(found[0][1:])

В RegularExpressions лежат готовые шаблоны для всех системных сообщений (ORDER_PURCHASED, ORDER_CONFIRMED, NEW_FEEDBACK, REFUND, PRODUCTS_AMOUNT, EXCHANGE_RATE…). Класс — синглтон, создавать можно свободно.

Раздел 07

Типы данных

Модели из FunPayAPI.types — то, что вы получаете из событий и методов аккаунта.

Message

str(message) возвращает текст сообщения.

ChatShortcut

Виджет чата со страницы /chat/: id, name, last_message_text (≤250 символов), last_message_type, unread, node_msg_id, user_msg_id, last_by_bot, html.

OrderShortcut

Приходит в NewOrderEvent / OrderStatusChangedEvent. Это «карточка» заказа из списка продаж, без деталей лота.

Order

Полный заказ (account.get_order() / c.get_order_from_object()): status, subcategory, fields, sum, currency, amount, buyer_id/username, seller_id/username, chat_id, review, player, server, side, order_secrets.

Полезные свойства: title, short_description, full_description, payment_msg, lot_params, lot_params_text, lot_params_dict, character_name, а также get_field(key) / get_field_value(key).

Review: stars, text, reply, anonymous, hidden, html.

Лоты: LotShortcut, MyLotShortcut, LotFields

• LotShortcut — лот в профиле/выдаче: id, server, side, description (он же title), amount, price, currency, subcategory, auto (автовыдача FunPay), public_link; для лотов из таблицы поиска ещё seller, promo, attributes.
• MyLotShortcut — свой лот из get_my_subcategory_lots(), с состоянием активности.
• LotFields — форма редактирования лота. Ключевое для изменения лотов из плагина.

LotFields: title_ru/title_en, description_ru/en, payment_msg_ru/en, price, amount, active, deactivate_after_sale, images, auto_delivery, secrets, subcategory, currency, public_link, private_link, calc_result.

fields = c.account.get_lot_fields(lot_id)
fields.price = round(fields.price * 0.95, 2)
fields.active = True
c.account.save_lot(fields.renew_fields())

UserProfile, Category, SubCategory

profile = c.account.get_user(c.account.id)
profile.get_lots()                    # list[LotShortcut]
profile.get_lot(lot_id)               # LotShortcut | None
profile.get_sorted_lots(1)            # {lot_id: LotShortcut}
profile.get_sorted_lots(2)            # {SubCategory: {lot_id: LotShortcut}}
profile.get_sorted_lots(3)            # {SubCategoryTypes: {lot_id: LotShortcut}}
profile.get_common_lots(); profile.get_currency_lots()

SubCategory: id, name, type (COMMON / CURRENCY), category, fullname, свойства is_lots, is_chips, is_currency, метод telegram_text(). Category: id, name, position, get_subcategories().

Balance, Currency, Wallet

Balance: total_rub, available_rub, total_usd, available_usd, total_eur, available_eur. Currency — RUB/USD/EUR/UNKNOWN, str(Currency.RUB) → ₽, .code → rub. Wallet (enum) — направления вывода: QIWI, BINANCE, TRC, CARD_RUB, CARD_USD, CARD_EUR, WEBMONEY, YOUMONEY.

Раздел 08

Account API

cardinal.account — прямые запросы к FunPay. Каждый вызов = HTTP-запрос, поэтому кэшируйте и не спамьте.

Чаты и сообщения

Для отправки в чат из плагина предпочитайте cardinal.send_message() — он разбивает текст и ретраит. Прямой account.send_message() нужен, когда водяной знак и разбиение мешают.

Заказы

get_sales() отдаёт страницу продаж; для обхода всех страниц передавайте полученный next_id в start_from, пока он не станет None. Между страницами обязательно делайте паузу.

Лоты и категории

Аккаунт, баланс, вывод

Поля аккаунта: id, username, active_sales, active_purchases, currency, total_balance, csrf_token, phpsessid, golden_key, locale, proxy.

Раздел 09

Автовыдача

Хуки PRE_DELIVERY / POST_DELIVERY позволяют подменить товар, дополнить сообщение или отреагировать на результат.

Цепочка при новом заказе выглядит так:

NEW_ORDER
 ├─ log_new_order_handler
 ├─ setup_event_attributes_handler   # ищет лот в конфиге автовыдачи, вешает атрибуты на event
 ├─ send_new_order_notification_handler
 └─ deliver_product_handler
      ├─ проверки: autoDelivery, ЧС, disable у лота
      ├─ ▶ PRE_DELIVERY
      ├─ deliver_goods()             # достаёт товар из файла и шлёт покупателю
      └─ ▶ POST_DELIVERY

Атрибуты, которые ядро вешает на NewOrderEvent

Читайте их через getattr(e, "delivered", False) — на «чужих» событиях (например, при тесте автовыдачи) атрибутов может не быть.

Пример: подменить текст выдачи

_ORIGINAL = {}

def pre_delivery(c: Cardinal, e: NewOrderEvent):
    cfg = getattr(e, "config_section_obj", None)
    name = getattr(e, "config_section_name", None)
    if cfg is None or name is None:
        return
    _ORIGINAL[name] = cfg["response"]                  # запомнили оригинал
    cfg["response"] = cfg["response"] + "\n\nПромокод на след. покупку: " + issue_promo()


def restore(c: Cardinal, e: NewOrderEvent):
    name = getattr(e, "config_section_name", None)
    if name in _ORIGINAL:
        c.AD_CFG[name]["response"] = _ORIGINAL.pop(name)   # вернули как было


def post_delivery(c: Cardinal, e: NewOrderEvent):
    if getattr(e, "error", 0):
        c.telegram and c.telegram.send_notification(
            f"❌ Не выдал заказ {e.order.id}: {getattr(e, 'error_text', '')}")
        return
    if (left := getattr(e, "goods_left", -1)) != -1 and left < 5:
        c.telegram and c.telegram.send_notification(f"⚠️ Товара осталось: {left}")


BIND_TO_PRE_DELIVERY = [pre_delivery]
BIND_TO_POST_DELIVERY = [restore, post_delivery]      # восстановление — первым

Собственная выдача без конфига FPC

Если лота нет в auto_delivery.cfg, штатная выдача не запустится и хуки PRE/POST не сработают. Тогда работайте прямо в BIND_TO_NEW_ORDER:

def on_new_order(c: Cardinal, e: NewOrderEvent):
    if "Мой особый лот" not in e.order.description:
        return
    order = c.get_order_from_object(e.order)          # полные данные, если нужны
    chat_id = (chat := c.account.get_chat_by_name(e.order.buyer_username)) and chat.id or e.order.chat_id
    for _ in range(e.order.amount or 1):
        c.send_message(chat_id, issue_key(), e.order.buyer_username)

Подстановки в текстах

Чтобы плагин вёл себя так же, как встроенная автовыдача, прогоняйте свои шаблоны через те же функции.

$product в этот список не входит: его подставляет сама автовыдача в deliver_goods(), уже после форматирования, вместо взятых из товарного файла строк. А $photo=ID, $sleep=СЕК и $new — не переменные, а разметка сообщения, которую разбирает cardinal.send_message() (см. Объект Cardinal).

Обратите внимание: $username и $chat_name подставляются через safe_text() — между символами вставляется невидимый разделитель, чтобы ник не превратился в ссылку-упоминание.

Раздел 10

Telegram-панель

cardinal.telegram — обёртка TGBot над pyTelegramBotAPI. Внутри — обычный telebot.TeleBot в cardinal.telegram.bot с parse_mode="HTML".

Регистрация хэндлеров

Все три обёртки ловят исключения внутри вашего хэндлера и пишут их в лог, так что бот не падает.

def init(cardinal: Cardinal):
    if not cardinal.telegram:
        return
    tg, bot = cardinal.telegram, cardinal.telegram.bot

    def cmd_stats(m: Message):
        bot.reply_to(m, f"Заказов обработано: <b>{COUNTER}</b>")

    def cb_reset(c: CallbackQuery):
        global COUNTER
        COUNTER = 0
        bot.answer_callback_query(c.id, "Сброшено", show_alert=False)

    tg.msg_handler(cmd_stats, commands=["mystats"])
    tg.cbq_handler(cb_reset, lambda c: c.data == f"{CBT_PREFIX}:reset")
    cardinal.add_telegram_commands(UUID, [("mystats", "статистика плагина", True)])


BIND_TO_PRE_INIT = [init]

Состояния и ввод текста

Правильный способ «спросить у пользователя значение» — механизм состояний FPC, а не register_next_step_handler: состояния видит вся ПУ, есть кнопка «Отмена», и чужой ввод не перехватывается.

STATE_PRICE = f"{CBT_PREFIX}_wait_price"

def act_set_price(c: CallbackQuery):
    msg = bot.send_message(c.message.chat.id, "Введите новую цену:",
                           reply_markup=CLEAR_STATE_BTN())
    tg.set_state(c.message.chat.id, msg.id, c.from_user.id, STATE_PRICE, {"lot": 123})
    bot.answer_callback_query(c.id)

def set_price(m: Message):
    data = tg.get_state(m.chat.id, m.from_user.id)["data"]
    tg.clear_state(m.chat.id, m.from_user.id, True)
    try:
        price = float(m.text.replace(",", "."))
    except ValueError:
        bot.reply_to(m, "Нужно число, например 149.90")
        return
    SETTINGS["price"] = price; save_config()
    bot.reply_to(m, "✅ Цена обновлена")

tg.cbq_handler(act_set_price, lambda c: c.data == f"{CBT_PREFIX}:setprice")
tg.msg_handler(set_price, func=lambda m: tg.check_state(m.chat.id, m.from_user.id, STATE_PRICE))

Кнопка «Отмена» — готовая: from tg_bot.static_keyboards import CLEAR_STATE_BTN (callback CBT.CLEAR_STATE сбрасывает состояние сам).

Чтобы принять файл, зарегистрируйте tg.file_handler(STATE_X, handler) — он вызовется, когда пользователь в состоянии STATE_X пришлёт document или photo.

Уведомления

Рассылает во все чаты, где включён этот тип уведомлений. Типы (tg_bot.utils.NotificationTypes): bot_start, new_message, command, new_order, order_confirmed, review, lots_restore, lots_deactivate, delivery, lots_raise, other, announcement, ad, critical, important_announcement.

from threading import Thread
Thread(target=c.telegram.send_notification, args=(text,), daemon=True).start()

HTML и экранирование

Бот работает в parse_mode="HTML". Любой текст от пользователя или из FunPay (ники, описания лотов) экранируйте, иначе сообщение не отправится:

from tg_bot import utils
bot.send_message(chat_id, f"Покупатель: <b>{utils.escape(order.buyer_username)}</b>")

Полезное из tg_bot.utils и keyboards

• utils.escape(text), utils.bool_to_text(v, on="🟢", off="🔴"), utils.split_by_limit(list, 4096);
• utils.add_navigation_buttons(keyboard, curr_offset, max_on_page, elements_on_page, elements_amount, callback_text, extra=None) — готовая пагинация;
• utils.get_offset(index, max_on_page);
• keyboards.* — клавиатуры ядра, static_keyboards — постоянные кнопки.

Авторизация и доступ

Доступ к ПУ получают только пользователи, приславшие боту секретный ключ; их список лежит в tg.authorized_users ({user_id: {...}}) и сохраняется между запусками.

OWNER_ID = 123456789
tg.msg_handler(cmd_danger, commands=["wipe"], func=lambda m: m.from_user.id == OWNER_ID)

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

Ограничения Telegram API

Два исключения, которые встречаются постоянно:

from telebot.apihelper import ApiTelegramException

try:
    bot.edit_message_text(text, chat_id, message_id, reply_markup=kb)
except ApiTelegramException as ex:
    if "message is not modified" not in str(ex):   # текст и клавиатура не изменились
        raise

Второе — 403 Forbidden: bot was blocked by the user: чат надо убирать из своих рассылок. Встроенный send_notification() делает это сам, ваши собственные рассылки должны так же.

Приём файлов

STATE_FILE = f"{CBT_PREFIX}_wait_file"

def act_upload(c: CallbackQuery):
    msg = bot.send_message(c.message.chat.id, "Пришлите .txt файл", reply_markup=CLEAR_STATE_BTN())
    tg.set_state(msg.chat.id, msg.id, c.from_user.id, STATE_FILE)
    bot.answer_callback_query(c.id)

def upload(m: Message):
    tg.clear_state(m.chat.id, m.from_user.id, True)
    if not m.document or not m.document.file_name.endswith(".txt"):
        bot.reply_to(m, "Нужен файл .txt")
        return
    data = bot.download_file(bot.get_file(m.document.file_id).file_path)
    with open("storage/plugins/myplugin_data.txt", "wb") as f:
        f.write(data)
    bot.reply_to(m, f"✅ Принято, {len(data)} байт")

tg.cbq_handler(act_upload, lambda c: c.data == f"{CBT_PREFIX}:upload")
tg.file_handler(STATE_FILE, upload)     # именно file_handler, не msg_handler

Локализация текстов плагина

ПУ переведена на русский, английский и украинский. В Localizer есть два метода специально для плагинов — тексты хранятся с префиксом UUID, поэтому не конфликтуют с ключами ядра и других плагинов.

from locales.localizer import Localizer

localizer = Localizer()

for lang, texts in {
    "ru": {"greet": "Привет, {}!", "saved": "Сохранено"},
    "en": {"greet": "Hi, {}!",     "saved": "Saved"},
    "uk": {"greet": "Привіт, {}!", "saved": "Збережено"},
}.items():
    for key, value in texts.items():
        localizer.add_translation(UUID, key, value, lang)


def _p(key: str, *args) -> str:
    return localizer.plugin_translate(UUID, key, *args)

bot.send_message(chat_id, _p("greet", username))

Регистрируйте переводы в BIND_TO_PRE_INIT. translate() при отсутствии ключа возвращает само имя ключа — если в интерфейсе появился myplugin_greet, значит перевод не зарегистрирован. Язык берётся из настроек FPC; отдельным аргументом language= можно принудительно выбрать другой.

Описания команд в add_telegram_commands() тоже прогоняются через локализатор, поэтому туда можно передавать как готовый текст, так и ключ перевода.

Раздел 11

Страница настроек плагина

Поставьте SETTINGS_PAGE = True — и в карточке плагина появится кнопка «Настройки». Нажатие шлёт callback CBT.PLUGIN_SETTINGS:UUID:offset, обработать его — ваша задача.

from tg_bot import CBT
from telebot.types import InlineKeyboardMarkup as K, InlineKeyboardButton as B, CallbackQuery

CBT_PREFIX = "MyPlugin"          # уникальный префикс своих callback'ов

def open_settings(c: CallbackQuery):
    kb = K()
    kb.add(B(f"Автоответ: {'🟢' if SETTINGS['enabled'] else '🔴'}",
             callback_data=f"{CBT_PREFIX}:toggle:enabled"))
    kb.add(B(f"Задержка: {SETTINGS['delay']} c", callback_data=f"{CBT_PREFIX}:set:delay"))
    kb.add(B("◀️ Назад", callback_data=f"{CBT.EDIT_PLUGIN}:{UUID}:0"))
    bot.edit_message_text("⚙️ Настройки MyPlugin", c.message.chat.id, c.message.id, reply_markup=kb)
    bot.answer_callback_query(c.id)

def toggle(c: CallbackQuery):
    key = c.data.split(":")[2]
    SETTINGS[key] = not SETTINGS[key]
    save_config()
    open_settings(c)

tg.cbq_handler(open_settings, lambda c: c.data.startswith(f"{CBT.PLUGIN_SETTINGS}:{UUID}"))
tg.cbq_handler(toggle, lambda c: c.data.startswith(f"{CBT_PREFIX}:toggle:"))

Полезные константы CBT

Удаление плагина

def on_delete(cardinal: Cardinal, call: CallbackQuery):
    """Вызывается перед удалением файла плагина через ПУ."""
    STOP_EVENT.set()
    if os.path.exists("storage/plugins/myplugin.json"):
        os.remove("storage/plugins/myplugin.json")

BIND_TO_DELETE = on_delete

Раздел 12

Хранение данных

У плагинов есть своя папка: storage/plugins/. Всё остальное в storage/ принадлежит ядру.

Шаблон конфига плагина

import json, os

CFG_PATH = "storage/plugins/myplugin.json"
DEFAULTS = {"enabled": True, "delay": 30, "text": "Спасибо за покупку!"}
SETTINGS = dict(DEFAULTS)


def load_config():
    global SETTINGS
    data = {}
    if os.path.exists(CFG_PATH):
        try:
            with open(CFG_PATH, "r", encoding="utf-8") as f:
                data = json.load(f)
        except (json.JSONDecodeError, OSError):
            logger.warning("Битый конфиг, беру значения по умолчанию")
    SETTINGS = {**DEFAULTS, **data}     # новые ключи добавятся при обновлении плагина
    save_config()


def save_config():
    os.makedirs(os.path.dirname(CFG_PATH), exist_ok=True)
    with open(CFG_PATH, "w", encoding="utf-8") as f:
        json.dump(SETTINGS, f, indent=4, ensure_ascii=False)

Утилиты ядра (Utils.cardinal_tools)

• load_blacklist() / cache_blacklist(list) — чёрный список;
• get_products(path, amount) / add_products(path, products, at_zero_position=False) / count_products(path) — работа с файлами товаров;
• format_order_text(text, order) / format_msg_text(text, obj) — подстановки переменных;
• time_to_str(seconds), get_month_name(n), safe_text(text);
• hash_password(p) / check_password(p, hashed);
• restart_program(), shut_down() — использовать очень аккуратно.

Раздел 13

Исключения

FunPayAPI.common.exceptions. Все сетевые ошибки наследуют RequestFailedError, у которого есть status_code, url, response и короткое описание short_str().

import FunPayAPI

try:
    c.account.raise_lots(category_id)
except FunPayAPI.exceptions.RaiseError as ex:
    wait = ex.wait_time or 60
    logger.warning(f"Поднятие не удалось: {ex.short_str()}, жду {wait} с")
except FunPayAPI.exceptions.RequestFailedError as ex:
    if ex.status_code in (429, 503, 403):
        time.sleep(60)
    logger.debug("TRACEBACK", exc_info=True)

Раздел 14

Сторонние библиотеки

Плагин — один файл, положить рядом requirements.txt некуда. Поэтому зависимости плагин ставит сам, в момент загрузки. Здесь важен и способ установки, и то, каким интерпретатором она выполняется — ошибка в любом из двух пунктов приводит к «библиотека вроде поставилась, а импорт всё равно падает».

Что уже установлено

Прежде чем что-то ставить, проверьте — возможно, нужное уже есть. В окружении FunPay Cardinal всегда доступны:

Плюс вся стандартная библиотека Python. Задачи вроде HTTP-запросов, парсинга HTML, работы с картинками, JSON, SQLite, потоками и планировщиком решаются без единой новой зависимости — это всегда предпочтительный вариант.

Как ставить правильно

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

import os
import sys
import tempfile
import importlib
import subprocess

# FunPayCardinal использует requests_toolbelt 0.10.1, который импортирует
# urllib3.contrib.appengine. В urllib3 2.x этого модуля нет. Если pip при
# установке зависимостей плагина подтянет urllib3 2.x, бот перестанет
# запускаться вообще — ImportError на старте, до всех плагинов.
# Constraints-файл запрещает pip трогать urllib3 при любой установке.
_CONSTRAINTS = os.path.join(tempfile.gettempdir(), "myplugin_pip_constraints.txt")
try:
    with open(_CONSTRAINTS, "w", encoding="utf-8") as f:
        f.write("urllib3<2\n")
except OSError:
    _CONSTRAINTS = None


def _ensure(module: str, package: str | None = None) -> bool:
    """Импортирует module, при необходимости ставит package."""
    try:
        importlib.import_module(module)
        return True
    except ImportError:
        pass

    cmd = [sys.executable, "-m", "pip", "install", package or module, "-q"]
    if _CONSTRAINTS:
        cmd += ["-c", _CONSTRAINTS]
    try:
        subprocess.check_call(cmd, stdout=subprocess.DEVNULL,
                              stderr=subprocess.DEVNULL, timeout=300)
    except Exception:
        return False

    importlib.invalidate_caches()          # иначе свежий пакет не найдётся
    try:
        importlib.import_module(module)
        return True
    except ImportError:
        return False


for _mod, _pkg in (("bs4", "beautifulsoup4"), ("pytz", "pytz"), ("yarl", "yarl")):
    _ensure(_mod, _pkg)

Разберём, почему каждая строка именно такая.

sys.executable, а не pip

Команда pip install ставит пакет в то окружение, на которое настроен исполняемый файл pip, а бот может работать под другим Python: venv из install-fpc.sh, portable-сборка на Windows, python3.11 при системном python3.9. Установка формально завершится успехом, а импорт всё равно упадёт. [sys.executable, "-m", "pip", …] гарантирует то же окружение, в котором крутится бот.

Имя модуля ≠ имя пакета

Проверять надо импортом модуля, а ставить — по имени дистрибутива. Они часто различаются: bs4 ← beautifulsoup4, PIL ← pillow, telebot ← pyTelegramBotAPI, dateutil ← python-dateutil, cv2 ← opencv-python, yaml ← PyYAML. Если проверять по имени пакета, установка будет запускаться при каждом старте бота.

Constraints-файл вместо аргумента "urllib3<2"

importlib.invalidate_caches() после установки

Python кэширует содержимое директорий sys.path. Если процесс уже пытался импортировать модуль и не нашёл его, то сразу после pip install импорт в том же процессе снова провалится — интерпретатор возьмёт ответ из кэша. invalidate_caches() сбрасывает его. Без этой строки установка «не срабатывает» до перезапуска бота.

Таймаут и подавление вывода

pip может уйти в долгую сборку из исходников или зависнуть на сети. Загрузка плагинов синхронна: пока pip работает, бот не стартует. timeout=300 ограничивает ожидание, stdout/stderr=DEVNULL не даёт pip засорить консоль FPC (его вывод не проходит через логгер и ломает читаемость лога).

Порядок установки: зависимости раньше основного пакета

Для «тяжёлых» пакетов с большим деревом зависимостей одна команда pip install X часто ведёт себя хуже, чем несколько последовательных: сначала ставятся зависимости, по одной, и только в конце — сам пакет.

# Порядок — снизу вверх по дереву зависимостей: сначала листья,
# потом то, что их использует, и в самом конце — основной пакет.
_DEPS = (                          # (пакет для pip, модуль для проверки)
    ("attrs", "attr"),             # нужен aiohttp
    ("multidict", "multidict"),    # нужен yarl / aiohttp
    ("propcache", "propcache"),    # нужен yarl / aiohttp
    ("frozenlist", "frozenlist"),  # нужен aiosignal / aiohttp
    ("aiohappyeyeballs", "aiohappyeyeballs"),
    ("aiosignal", "aiosignal"),    # требует frozenlist
    ("yarl", "yarl"),              # требует multidict, propcache
    ("aiohttp", "aiohttp"),        # транспорт
    ("ton-core", "ton_core"),      # слой ячеек/BOC
)

for _pkg, _mod in _DEPS:
    _ensure(_mod, _pkg)            # каждая зависимость — отдельным вызовом pip

_ensure("tonutils", "tonutils")    # основной пакет — последним

Что этот приём реально даёт

• Установка перестаёт быть «всё или ничего». pip install tonutils — одна транзакция: если хотя бы одна транзитивная зависимость не собралась, pip откатывает всё и не ставит ничего, даже те десять пакетов, что уже скачались. По одной — прогресс сохраняется, и повторный запуск доделывает остаток.
• Видно, что именно упало. В общей установке ошибка тонет в длинном выводе резолвера. Поштучно вы получаете точное имя пакета и можете написать его пользователю в отчёте.
• Резолвер не уходит в перебор. Когда требования не удовлетворены, pip начинает backtracking: качает десятки версий подряд, подбирая совместимый набор. На слабом VPS это минуты ожидания и иногда обрыв по таймауту. Если зависимости уже стоят, при установке основного пакета резолверу нечего перебирать.
• Меньше шансов упереться в память и таймаут. Пакеты с C-расширениями, для которых нет готового wheel под этот Python, собираются из исходников. Несколько сборок в одной транзакции — это один длинный процесс, который легче упирается в лимит; отдельные вызовы имеют каждый свой timeout.
• Пропуск уже установленного. Проверив _already(dep), вы не трогаете то, что и так есть в окружении FPC, — а значит, не рискуете его обновить.

Жёсткий вариант: --no-deps

Если список зависимостей выверен и все они уже установлены, ставьте основной пакет с запретом трогать что-либо ещё:

cmd = [sys.executable, "-m", "pip", "install", "--no-deps", "tonutils", "-q"]

Тогда pip физически не сможет обновить urllib3 или подменить requests. Обратная сторона — ответственность за полноту списка на вас: пропустили зависимость, и пакет упадёт уже при импорте. Поэтому --no-deps применяют только к пакетам с проверенным деревом, а сам импорт после установки обязательно проверяют.

Как узнать список зависимостей

python -m pip install --dry-run tonutils    # покажет, что было бы установлено (pip 22.2+)
python -m pip show tonutils                 # поле Requires у уже установленного пакета

Порядок внутри списка составляйте от листьев к корню: сначала пакеты без зависимостей (attrs, multidict), затем те, что их используют (yarl, aiosignal), затем крупные узлы (aiohttp). Для pip такой порядок не обязателен — он в любом случае разберётся, — но при нём каждый шаг ставится «начисто», без резолва.

Проверка результата и отчёт пользователю

Установка «прошла успешно» и «библиотека работает» — не одно и то же: пакет мог поставиться, а нужный класс появиться только в более новой версии. Поэтому финальный шаг — импорт того, ради чего всё затевалось:

from importlib.metadata import version, PackageNotFoundError

def verify() -> tuple[list[str], bool]:
    lines, ok_all = [], True
    try:
        lines.append(f"📦 tonutils {version('tonutils')}")
    except PackageNotFoundError:
        lines.append("📦 tonutils не установлен")
        return lines, False

    importlib.invalidate_caches()
    for module, cls in (("tonutils.contracts.wallet", "WalletV5R1"),
                        ("tonutils.contracts.wallet", "WalletV4R2")):
        try:
            m = importlib.import_module(module)
            ok = hasattr(m, cls)
            lines.append(f"{'🟢' if ok else '🔴'} {cls}")
            ok_all &= ok
        except Exception as e:
            ok_all = False
            lines.append(f"🔴 {cls} — {type(e).__name__}: {str(e)[:60]}")
    return lines, ok_all

Если установка запускается по команде из Telegram, выполняйте её в отдельном потоке, показывайте «⏳ ставлю…» и затем редактируйте сообщение отчётом по каждому пакету. Пользователь видит, какой именно шаг не прошёл, и может доустановить руками.

Почему установка не срабатывает

Запасной вариант с --user

    try:
        subprocess.check_call(cmd, stdout=subprocess.DEVNULL,
                              stderr=subprocess.DEVNULL, timeout=300)
    except subprocess.CalledProcessError:
        try:                                  # нет прав на системный site-packages
            subprocess.check_call(cmd + ["--user"], stdout=subprocess.DEVNULL,
                                  stderr=subprocess.DEVNULL, timeout=300)
        except Exception:
            return False

Ленивая установка: не тормозить старт бота

Если библиотека нужна не всегда (например, matplotlib только для команды с графиком), не ставьте её при загрузке модуля. Ставьте при первом обращении и корректно сообщайте пользователю, если не вышло:

def ensure_matplotlib() -> bool:
    try:
        import matplotlib  # noqa: F401
        return True
    except ImportError:
        pass
    ok = _ensure("matplotlib")
    if not ok:
        logger.warning(f"{LOGGER_PREFIX} matplotlib не установился")
    return ok


def cmd_chart(m: Message):
    if not ensure_matplotlib():
        bot.reply_to(m, "График недоступен: не удалось установить matplotlib.\n"
                        "Установите вручную: <code>pip install matplotlib \"urllib3<2\"</code>")
        return
    import matplotlib
    matplotlib.use("Agg")            # обязательно: на сервере нет дисплея
    ...

Долгая доустановка в фоне

Компоненты, которые качают сотни мегабайт (браузеры Playwright, модели), запускайте отдельным потоком, чтобы не блокировать загрузку плагинов:

def _ensure_chromium():
    # команда идемпотентна: если браузер уже стоит — завершится мгновенно
    try:
        subprocess.check_call([sys.executable, "-m", "playwright", "install", "chromium"],
                              stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=600)
    except Exception:
        pass

threading.Thread(target=_ensure_chromium, daemon=True).start()

Раздел 15

Рекомендации по разработке

Ограничения платформы и типичные причины, по которым плагин мешает работе бота.

Не блокируйте поток событий

Все хэндлеры событий выполняются последовательно в одном цикле. time.sleep(60) внутри хэндлера останавливает обработку всех сообщений и заказов. Всё долгое — в поток:

from threading import Thread

def on_new_order(c: Cardinal, e: NewOrderEvent):
    Thread(target=heavy_work, args=(c, e), daemon=True, name="MyPlugin-order").start()

Фоновые циклы запускайте один раз в BIND_TO_POST_START и обязательно с daemon=True, обёрнутыми в try/except внутри цикла — иначе одна ошибка убьёт поток навсегда.

Прокси

Если у пользователя настроен прокси, ваши собственные HTTP-запросы должны идти через него — иначе FunPay увидит два разных IP:

requests.get(url, proxies=cardinal.proxy or None, timeout=10)

Защитное программирование

• Проверяйте cardinal.telegram на None перед любым обращением к ПУ.
• Проверяйте наличие атрибутов события через getattr(e, "…", default): у фейковых событий (тест автовыдачи) их может не быть.
• Ловите исключения в потоках. Ядро перехватывает ошибки только в хэндлерах, потоки — ваша ответственность.
• Не бросайте исключения из BIND_TO_PRE_INIT — ошибка будет проглочена, но плагин останется наполовину проинициализированным.
• Не читайте MAIN_CFG напрямую там, где есть свойство (c.autodelivery_enabled вместо c.MAIN_CFG["FunPay"].getboolean("autoDelivery")).
• Всегда encoding="utf-8" при работе с файлами: на Windows дефолтная кодировка не UTF-8, и кириллица превратится в мусор.

Ограничения FunPay

• Сообщение длиннее ~20 строк не отправится целиком — используйте cardinal.send_message(), он режет сам.
• Слишком частая отправка сообщений разным пользователям вызывает флуд-ошибку; аккаунт помечает время в account.last_flood_err_time / last_multiuser_flood_err_time.
• Лоты игровой валюты (SubCategoryTypes.CURRENCY) поднимать нельзя — ядро их пропускает, делайте так же.
• get_user(), get_sales(), get_lot_fields() — тяжёлые запросы. Кэшируйте результат хотя бы на минуту.

Поведение в ПУ

• Уникальный префикс во всех callback_data, фильтры через startswith, а не in.
• Всегда вызывайте bot.answer_callback_query(c.id) — иначе у пользователя кнопка «крутится».
• Экранируйте пользовательский текст через utils.escape().
• Не используйте bot.register_next_step_handler в плагинах для ПУ: он перехватывает следующее сообщение пользователя глобально и ломает чужие диалоги. Используйте состояния FPC.
• Логируйте с префиксом плагина: logging.getLogger("FPC.MyPlugin").

Раздел 16

Отладка и диагностика

Ядро проглатывает ошибки плагинов, чтобы бот не падал. Обратная сторона — сломанный плагин ведёт себя как «просто ничего не делает». Вот как выяснять, что произошло.

Плагин не появился в списке

Проверяйте по порядку — это закрывает почти все случаи:

1. Файл лежит именно в plugins/, расширение .py, не во вложенной папке.
2. Первая строка не содержит noplug.
3. Объявлены все семь полей: NAME, VERSION, DESCRIPTION, CREDITS, UUID, SETTINGS_PAGE, BIND_TO_DELETE. Отсутствие любого — FieldNotExistsError.
4. UUID — валидный UUID4 и не совпадает с другим плагином.
5. Модуль импортируется без ошибок. Синтаксическая ошибка или падение кода верхнего уровня → в логе Не удалось загрузить плагин.
6. Бот перезапущен после добавления файла.

Быстрая проверка синтаксиса и импорта, не поднимая бота (из корня FPC, тем же интерпретатором):

python -c "import importlib.util as u, sys; sys.path.insert(0,'plugins'); \
s=u.spec_from_file_location('p','plugins/myplugin.py'); m=u.module_from_spec(s); s.loader.exec_module(m); \
print(m.NAME, m.VERSION, m.UUID)"

Логи

Причина ошибки в хэндлере пишется на уровне ERROR, а полный traceback — только на DEBUG. Поэтому в своих except повторяйте приём ядра:

try:
    risky()
except Exception:
    logger.error(f"{LOGGER_PREFIX} не удалось выполнить операцию")
    logger.debug("TRACEBACK", exc_info=True)

• Логи пишутся в logs/log.log с ротацией по 20 МБ (до 25 файлов); из Telegram — команды /logs и /del_logs.
• В консоль идёт только INFO и выше, в файл — DEBUG. Значит logger.debug("TRACEBACK", exc_info=True) на экране не появится, но в файле будет. Traceback всегда ищите в logs/log.log.
• В LOGGER_CONFIG (Utils/logger.py) настроены логгеры main, FunPayAPI, FPC, TGBot, TeleBot. Имя FPC.МойПлагин наследует настройки FPC — поэтому именно такой префикс и нужен: произвольное имя вроде myplugin в файл не попадёт.
• Цветовые метки в тексте сообщения: $YELLOW, $CYAN, $MAGENTA, $BLUE, $GREEN, $BLACK, $WHITE, фоновые $B_* и $RESET — возврат к цвету уровня. В файле они автоматически вырезаются.
• Не пишите в лог golden_key, токены, пароли и содержимое товаров: пользователи отправляют логи в поддержку.

Как воспроизвести событие, не дожидаясь его

Ждать реального заказа ради проверки хэндлера не нужно. Есть два способа.

Штатный тест автовыдачи — команда /test_lot в ПУ. Ядро собирает OrderShortcut с ID ADTEST и прогоняет через него всю цепочку NEW_ORDER, включая ваши хэндлеры. Учтите: у такого заказа нет реальных данных, get_order_from_object() вернёт None.

Собственный тестовый прогон — временная команда в ПУ, которая берёт настоящий заказ и вызывает ваш обработчик:

from FunPayAPI.common.utils import random_tag
from FunPayAPI.updater.events import NewOrderEvent

def cmd_test(m: Message):
    order_id = m.text.split(maxsplit=1)[1]           # /mytest ABCD1234
    shortcut = cardinal.account.get_order_shortcut(order_id)
    event = NewOrderEvent(random_tag(), shortcut)
    cardinal.run_handlers(cardinal.new_order_handlers, (cardinal, event))
    bot.reply_to(m, "Прогнал цепочку NEW_ORDER")

Так же тестируются сообщения: NewMessageEvent(tag, message_obj), где message_obj берётся из account.get_chat_history(chat_id)[-1]. Тег генерируется через FunPayAPI.common.utils.random_tag().

Симптомы и причины

Перед отдачей плагина пользователю

• Запустить бота с плагином «с нуля»: без файла конфига, без установленных зависимостей.
• Проверить работу при выключенном Telegram ([Telegram] enabled = 0).
• Выключить плагин в ПУ и убедиться, что он действительно перестал действовать.
• Удалить плагин через ПУ и проверить, что BIND_TO_DELETE прибрал за собой.
• Прогнать основной сценарий дважды подряд — ловятся ошибки состояния и повторной инициализации.

Раздел 17

Типовые задачи

Готовые фрагменты кода под самые частые сценарии.

Команда в чате FunPay

def on_message(c: Cardinal, e: NewMessageEvent):
    m = e.message
    if m.author_id in (0, c.account.id) or m.by_bot:
        return
    if (m.text or "").strip().lower() == "!сколько":
        left = cardinal_tools.count_products("storage/products/keys.txt")
        c.send_message(m.chat_id, f"В наличии: {left} шт.", m.chat_name)

BIND_TO_NEW_MESSAGE = [on_message]

Уведомление о заказе с кнопкой ответа

from tg_bot import CBT, utils
from telebot.types import InlineKeyboardMarkup as K, InlineKeyboardButton as B

def on_new_order(c: Cardinal, e: NewOrderEvent):
    if not c.telegram:
        return
    chat = c.account.get_chat_by_name(e.order.buyer_username)
    chat_id = chat.id if chat else e.order.chat_id
    kb = K().add(B("✉️ Ответить", callback_data=f"{CBT.SEND_FP_MESSAGE}:{chat_id}:{e.order.buyer_username}"))
    text = (f"🛒 <b>{utils.escape(e.order.description)}</b>\n"
            f"👤 {utils.escape(e.order.buyer_username)} · {e.order.price} {e.order.currency}")
    Thread(target=c.telegram.send_notification, args=(text, kb), daemon=True).start()

BIND_TO_NEW_ORDER = [on_new_order]

Реакция на отзыв

from FunPayAPI.common.enums import MessageTypes

def on_review(c: Cardinal, e: NewMessageEvent):
    if e.message.type not in (MessageTypes.NEW_FEEDBACK, MessageTypes.FEEDBACK_CHANGED):
        return
    if e.message.author_id != 0:
        return
    order = c.get_order_from_object(e.message)
    if not order or not order.review:
        return
    if order.review.stars <= 2:
        logger.warning(f"Плохой отзыв на {order.id}: {order.review.text}")
        c.telegram and c.telegram.send_notification(f"⭐ {order.review.stars} по заказу {order.id}")

BIND_TO_NEW_MESSAGE = [on_review]

Изменить цену всех своих лотов подкатегории

def discount(c: Cardinal, subcategory_id: int, k: float = 0.95):
    for lot in c.account.get_my_subcategory_lots(subcategory_id):
        try:
            fields = c.account.get_lot_fields(lot.id)
            if fields.price is None:
                continue
            fields.price = round(fields.price * k, 2)
            c.account.save_lot(fields.renew_fields())
            logger.info(f"Лот {lot.id}: новая цена {fields.price}")
        except Exception:
            logger.debug("TRACEBACK", exc_info=True)
        time.sleep(2)          # пауза обязательна

Фоновый цикл

STOP = threading.Event()

def worker(c: Cardinal):
    while not STOP.wait(SETTINGS["interval"]):
        pl = c.plugins.get(UUID)
        if not pl or not pl.enabled:
            continue                      # плагин выключен в ПУ
        try:
            do_work(c)
        except Exception:
            logger.error("Ошибка в фоновом цикле")
            logger.debug("TRACEBACK", exc_info=True)

def start_worker(c: Cardinal):
    Thread(target=worker, args=(c,), daemon=True, name="MyPlugin-worker").start()

BIND_TO_POST_START = [start_worker]

Пагинация в Telegram-ПУ

from tg_bot import utils

MAX_ON_PAGE = 5

def open_list(c: CallbackQuery):
    offset = int(c.data.split(":")[-1])
    items = list(DATA.items())
    kb = K()
    for key, val in items[offset:offset + MAX_ON_PAGE]:
        kb.add(B(f"{key} — {val}", callback_data=f"{CBT_PREFIX}:item:{key}:{offset}"))
    utils.add_navigation_buttons(kb, offset, MAX_ON_PAGE, len(items[offset:offset + MAX_ON_PAGE]),
                                 len(items), f"{CBT_PREFIX}:list")
    kb.add(B("◀️ Назад", callback_data=f"{CBT.EDIT_PLUGIN}:{UUID}:0"))
    bot.edit_message_text("Список", c.message.chat.id, c.message.id, reply_markup=kb)
    bot.answer_callback_query(c.id)

tg.cbq_handler(open_list, lambda c: c.data.startswith(f"{CBT_PREFIX}:list:"))

Отправить изображение в чат FunPay

image_id = c.account.upload_image("storage/plugins/manual.png", type_="chat")
c.send_message(chat_id, f"$photo={image_id}", chat_name)     # ID можно переиспользовать

Изображение выгружается один раз и живёт на сервере FunPay — кэшируйте полученный image_id в своём конфиге, а не грузите файл при каждой отправке. Для картинки лота нужен type_="offer".

Добавить покупателя в чёрный список

from Utils import cardinal_tools

def ban(c: Cardinal, username: str):
    if username in c.blacklist:
        return False
    c.blacklist.append(username)
    cardinal_tools.cache_blacklist(c.blacklist)   # без этого потеряется при перезапуске
    return True

Обойти все продажи постранично

def iter_sales(c: Cardinal, max_pages: int = 20):
    start_from, seen = None, []
    for _ in range(max_pages):
        next_id, orders, _locale, _subcats = c.account.get_sales(
            start_from=start_from, exclude_ids=[o.id for o in seen])
        seen.extend(orders)
        if not next_id:
            break
        start_from = next_id
        time.sleep(2)                 # без паузы прилетит 429
    return seen

Метод возвращает кортеж (next_id, orders, locale, subcategories). Есть фильтры: buyer, state, game, server, side, include_paid/closed/refunded — используйте их, чтобы не тянуть лишние страницы.

Сообщение после подтверждения заказа

def on_status(c: Cardinal, e: OrderStatusChangedEvent):
    if e.order.status is not OrderStatuses.CLOSED:
        return
    chat = c.account.get_chat_by_name(e.order.buyer_username)
    if chat:
        c.send_message(chat.id, "Спасибо! Будем рады видеть вас снова 🤝", e.order.buyer_username)

BIND_TO_ORDER_STATUS_CHANGED = [on_status]

Раздел 18

Полный пример плагина

Метаданные, конфиг в JSON, страница настроек с переключателем и вводом текста, хук на заказ, фоновый поток. Скелет, из которого можно делать что угодно.

"""
Thanks — благодарит покупателя после закрытия заказа
и раз в сутки шлёт в Telegram сводку по продажам.
"""
from __future__ import annotations

import json
import os
import time
import logging
import threading
from typing import TYPE_CHECKING

from telebot.types import InlineKeyboardMarkup as K, InlineKeyboardButton as B, Message, CallbackQuery

from tg_bot import CBT, utils
from tg_bot.static_keyboards import CLEAR_STATE_BTN
from FunPayAPI.common.enums import OrderStatuses

if TYPE_CHECKING:
    from cardinal import Cardinal
    from FunPayAPI.updater.events import OrderStatusChangedEvent

logger = logging.getLogger("FPC.Thanks")
LOGGER_PREFIX = "[THANKS]"

# ─── Метаданные ────────────────────────────────────────────────
NAME = "Thanks"
VERSION = "1.0.0"
DESCRIPTION = "Благодарит покупателя после подтверждения заказа + суточная сводка."
CREDITS = "@yourname"
UUID = "3f1c0b8e-58d2-4e6a-9a1f-7c2b4d9e05aa"
SETTINGS_PAGE = True

# ─── Конфиг ────────────────────────────────────────────────────
CFG_PATH = "storage/plugins/thanks.json"
DEFAULTS = {
    "enabled": True,
    "text": "Спасибо за покупку! Будем рады видеть вас снова 🤝",
    "digest": True,
}
SETTINGS = dict(DEFAULTS)
CBT_PREFIX = "Thanks"
STATE_TEXT = "Thanks_wait_text"
STOP = threading.Event()
COUNTER = {"day": time.strftime("%d"), "orders": 0, "sum": 0.0}


def save_config():
    os.makedirs(os.path.dirname(CFG_PATH), exist_ok=True)
    with open(CFG_PATH, "w", encoding="utf-8") as f:
        json.dump(SETTINGS, f, indent=4, ensure_ascii=False)


def load_config():
    global SETTINGS
    data = {}
    if os.path.exists(CFG_PATH):
        try:
            with open(CFG_PATH, "r", encoding="utf-8") as f:
                data = json.load(f)
        except (json.JSONDecodeError, OSError):
            logger.warning(f"{LOGGER_PREFIX} битый конфиг, беру значения по умолчанию")
    SETTINGS = {**DEFAULTS, **data}
    save_config()


def is_enabled(c: Cardinal) -> bool:
    pl = c.plugins.get(UUID)
    return bool(pl and pl.enabled and SETTINGS["enabled"])


# ─── Telegram-панель ───────────────────────────────────────────
def init(cardinal: Cardinal):
    load_config()
    if not cardinal.telegram:
        logger.info(f"{LOGGER_PREFIX} Telegram выключен — работает только логика чата.")
        return
    tg, bot = cardinal.telegram, cardinal.telegram.bot

    def settings_kb() -> K:
        kb = K()
        kb.add(B(f"Благодарность: {utils.bool_to_text(SETTINGS['enabled'])}",
                 callback_data=f"{CBT_PREFIX}:toggle:enabled"))
        kb.add(B(f"Суточная сводка: {utils.bool_to_text(SETTINGS['digest'])}",
                 callback_data=f"{CBT_PREFIX}:toggle:digest"))
        kb.add(B("✏️ Изменить текст", callback_data=f"{CBT_PREFIX}:edit_text"))
        kb.add(B("◀️ Назад", callback_data=f"{CBT.EDIT_PLUGIN}:{UUID}:0"))
        return kb

    def open_settings(c: CallbackQuery):
        text = (f"⚙️ <b>{NAME} v{VERSION}</b>\n\n"
                f"Текущий текст:\n<code>{utils.escape(SETTINGS['text'])}</code>")
        bot.edit_message_text(text, c.message.chat.id, c.message.id, reply_markup=settings_kb())
        bot.answer_callback_query(c.id)

    def toggle(c: CallbackQuery):
        key = c.data.split(":")[2]
        SETTINGS[key] = not SETTINGS[key]
        save_config()
        open_settings(c)

    def act_edit_text(c: CallbackQuery):
        msg = bot.send_message(c.message.chat.id, "Пришлите новый текст благодарности:",
                               reply_markup=CLEAR_STATE_BTN())
        tg.set_state(msg.chat.id, msg.id, c.from_user.id, STATE_TEXT)
        bot.answer_callback_query(c.id)

    def edit_text(m: Message):
        tg.clear_state(m.chat.id, m.from_user.id, True)
        SETTINGS["text"] = m.text
        save_config()
        bot.reply_to(m, "✅ Текст сохранён",
                     reply_markup=K().add(B("◀️ К настройкам",
                                            callback_data=f"{CBT.PLUGIN_SETTINGS}:{UUID}:0")))

    def cmd_stats(m: Message):
        bot.reply_to(m, f"📊 За сегодня: <b>{COUNTER['orders']}</b> заказ(ов) "
                        f"на <b>{COUNTER['sum']:.2f}</b>")

    tg.cbq_handler(open_settings, lambda c: c.data.startswith(f"{CBT.PLUGIN_SETTINGS}:{UUID}"))
    tg.cbq_handler(toggle, lambda c: c.data.startswith(f"{CBT_PREFIX}:toggle:"))
    tg.cbq_handler(act_edit_text, lambda c: c.data == f"{CBT_PREFIX}:edit_text")
    tg.msg_handler(edit_text, func=lambda m: tg.check_state(m.chat.id, m.from_user.id, STATE_TEXT))
    tg.msg_handler(cmd_stats, commands=["thanks_stats"])
    cardinal.add_telegram_commands(UUID, [("thanks_stats", "статистика Thanks", True)])
    logger.info(f"{LOGGER_PREFIX} v{VERSION} инициализирован.")


# ─── Логика ────────────────────────────────────────────────────
def on_status_changed(c: Cardinal, e: OrderStatusChangedEvent):
    if e.order.status is not OrderStatuses.CLOSED:
        return
    COUNTER["orders"] += 1
    COUNTER["sum"] += e.order.price
    if not is_enabled(c):
        return
    chat = c.account.get_chat_by_name(e.order.buyer_username)
    chat_id = chat.id if chat else e.order.chat_id
    c.send_message(chat_id, SETTINGS["text"], e.order.buyer_username)
    logger.info(f"{LOGGER_PREFIX} поблагодарил {e.order.buyer_username} за {e.order.id}")


def digest_worker(c: Cardinal):
    while not STOP.wait(600):
        try:
            today = time.strftime("%d")
            if today == COUNTER["day"]:
                continue
            if SETTINGS["digest"] and c.telegram:
                c.telegram.send_notification(
                    f"📅 Итоги дня: <b>{COUNTER['orders']}</b> заказ(ов), "
                    f"<b>{COUNTER['sum']:.2f}</b>")
            COUNTER.update(day=today, orders=0, sum=0.0)
        except Exception:
            logger.error(f"{LOGGER_PREFIX} ошибка в сводке")
            logger.debug("TRACEBACK", exc_info=True)


def start_worker(c: Cardinal):
    threading.Thread(target=digest_worker, args=(c,), daemon=True, name="Thanks-digest").start()


def on_delete(c: Cardinal, call: CallbackQuery):
    STOP.set()
    if os.path.exists(CFG_PATH):
        os.remove(CFG_PATH)


BIND_TO_PRE_INIT = [init]
BIND_TO_POST_START = [start_worker]
BIND_TO_ORDER_STATUS_CHANGED = [on_status_changed]
BIND_TO_DELETE = on_delete