Перейти к содержанию

Plugin и Router#

Plugin — корень плагина, Router — набор объявлений модуля. Всё, что умеет Router, доступно и у Plugin. Руководство: Плагин и модули.

Plugin #

Plugin(
    name: str,
    *,
    version: str = "0.0.0",
    author: str = "",
    description: str = "",
    url: str = "",
)

Корень плагина: название, версия, автор и все объявления.

В пакете плагина создаётся ровно один Plugin(...); другие модули объявляют обработчики в своих Router и подключаются через include.

Параметры#
  • name — название в amxpy list и логах.
  • version — версия плагина.
  • author — автор.
  • description — краткое описание.
  • url — страница плагина.
Пример#
plugin = Plugin("VIP", version="1.2.0", author="you")
cfg = plugin.include(Settings(), perks.router)

log #

log: Logger

Логгер плагина: plugin.log.info("...") пишет в лог сервера с именем плагина.

config_file #

config_file(name: str, default: str = '') -> Path

Файл настроек плагина на сервере: addons/amxpy/configs/<плагин>/<name>.

Если файла нет, он создаётся с текстом default — общий отступ и первая пустая строка убираются. Существующий файл не перезаписывается никогда: ни при перезагрузке, ни при amxpy deploy — это файл владельца сервера. Формат любой, читайте его сами (configparser, tomllib, json), лучше в events.Init или events.MapStart.

Параметры#
  • name — имя файла в папке плагина, можно с подпапкой: texts/ru.txt.
  • default — содержимое нового файла.
Возвращает#

Путь к файлу (pathlib.Path).

Исключения#
  • ValueError — путь выходит за папку плагина.
Пример#
path = plugin.config_file("adverts.ini", default='''
    [messages]
    1 = !gНаш сайт: !texample.com
''')

@plugin.on(events.Init)
def load(e: events.Init) -> None:
    ini = configparser.ConfigParser()
    ini.read(path, encoding="utf-8")

keep #

keep(key: str, value: Any) -> None

Сохранить значение на время горячей перезагрузки (должно поддерживать pickle).

Новая версия плагина получит его через kept — в том числе при импорте модуля. Вызывать можно в любой момент, в том числе в on_unload (в async-колбэке — до первого await). Данные игроков проще хранить в PlayerState(keep=True).

Пример#
@plugin.on_unload
def save() -> None:
    plugin.keep("round", current_round)

kept #

kept(key: str, default: Any = None) -> Any

Значение, сохранённое через keep предыдущей версией плагина.

Пример#
current_round = plugin.kept("round", 0)

on_unload #

on_unload(callback: Callable[[], Any]) -> Callable[[], Any]

Вызвать при выгрузке и перезагрузке плагина; может быть async.

Пример#
@plugin.on_unload
async def flush() -> None:
    await table.put(stats)

Router #

Router()

Набор объявлений модуля: обработчики, команды, нативы, форварды, задачи.

Большой плагин делится на модули: в каждом свой Router(), а корневой Plugin подключает их через include. Декораторы только записывают объявления — регистрирует их рантайм при загрузке плагина.

Пример#
router = Router()

@router.command("/hp")
def hp(player: Player) -> None:
    player.print(f"HP: {player.health}")

log #

log: Logger

Логгер плагина: plugin.log.info("...") пишет в лог сервера с именем плагина.

on #

on(
    event: type[E], *, post: bool | None = None, priority: int = 0, **filters: Any
) -> HandlerDecorator[E]

Обработчик события.

В pre-фазе (по умолчанию) обработчик может менять изменяемые поля, отменить действие игры через e.cancel() или вернуть значение — ответ за игру. В post-фазе событие уже случилось: только чтение. Async-обработчик видит событие до первого await — нужные поля скопируйте заранее.

Параметры#
  • event — класс события из amxpy.events.
  • post — True — после действия игры, False — до; по умолчанию — первая фаза события (pre, а у событий только с post-фазой, как FreezeEnd, — post).
  • priority — больше — раньше среди обработчиков этого события.
  • **filters — поле=значение — вызывать только при совпадении поля.
Исключения#
  • TypeError — у события нет такой фазы или полей для фильтра.
Пример#
@plugin.on(events.TakeDamage)
def double(e: events.TakeDamage) -> None:
    if e.attacker in vips:
        e.damage *= 2

@plugin.on(events.Say, team=True)  # только say_team
def team_chat(e: events.Say) -> None: ...

command #

command(
    *names: str,
    access: Access | int = Access.NONE,
    help: str = "",
    cooldown: float = 0.0,
    hide: bool = True,
) -> CommandDecorator

Команда игрока: /name или !name — в чате, иначе — в консоли.

Первый параметр обработчика — игрок, вызвавший команду. Остальные аргументы разбираются по аннотациям: str, int, float, bool, Player (по нику, #userid или SteamID), Enum, Literal[...], X | None; *rest: str забирает остаток строки. Usage и сообщения об ошибках строятся сами: amx_heal <target> [amount].

Параметры#
  • *names — имена команды; несколько — синонимы ("/hp", "!hp").
  • access — нужные флаги доступа; без них игрок получит отказ.
  • help — описание для списка команд.
  • cooldown — секунд между вызовами одним игроком.
  • hide — не показывать чат-команду в чате.
Пример#
@plugin.command("amx_heal", access=Access.SLAY, help="Вылечить игрока")
def heal(admin: Player, target: Player, amount: int = 100) -> None:
    target.health = min(target.health + amount, 255)

server_command #

server_command(name: str, *, help: str = '') -> Decorator

Команда серверной консоли и rcon (без игрока).

Аргументы разбираются по аннотациям, как у command.

Пример#
@plugin.server_command("amx_vip_reload", help="Перечитать VIP из базы")
async def reload_vips() -> None: ...

native #

native(fn: F) -> F

native(fn: str | None = None, /, *, name: str | None = None) -> Decorator

native(
    fn: Callable[..., Any] | str | None = None, /, *, name: str | None = None
) -> Any

Натив для Pawn-плагинов; сигнатура берётся из аннотаций.

Типы параметров: int, float, bool, str, Player; результат — они же или None. Натив попадает в сгенерированный <плагин>.inc вместе с docstring; новый натив или смена сигнатуры доступны Pawn после рестарта сервера. Натив не может быть async: Pawn ждёт ответ сразу.

Пример#
@plugin.native
def vip_is_user_vip(player: Player) -> bool:
    return player in vips

@plugin.native("vip_days")  # имя для Pawn отличается от имени функции
def days_left(player: Player) -> int: ...

forward #

forward(name: str, *types: type, **named: type) -> Forward

Объявить форвард для Pawn- и Python-подписчиков.

Параметры#
  • name — имя форварда (латиница, как в Pawn).
  • *types — типы безымянных параметров (arg1, arg2...).
  • **named — именованные параметры: имя=тип.
Возвращает#

Forward; разослать — .emit(...).

Пример#
granted = plugin.forward("vip_granted", player=Player, days=int)
...
granted.emit(player, 30)

on_forward #

on_forward(name: str) -> Decorator

Подписка на форвард по имени: Python-плагина, Pawn-плагина или самого AMXX.

Форварды Pawn (CreateMultiForward + ExecuteForward) и форварды ядра (client_putinserver...) доходят через плагин-приёмник amxpy_forwards.amxx, который AMXX загружает с каждой картой: новая подписка начинает работать со следующей карты. Аргументы разбираются по аннотациям: int, bool, float, str, Player; amxx.Ref/amxx.FloatRef — параметр по ссылке, его .value вернётся в Pawn. Без аннотации — число. Возвращённое число — ответ форварда (PLUGIN_HANDLED = 1); из нескольких подписчиков берётся наибольший.

Пример#
@plugin.on_forward("jb_day_start")          # форвард Pawn-мода
def day_start(day: int, title: str) -> None:
    server.print(f"!gДень {day}: !t{title}")

@plugin.on_forward("vip_granted")           # форвард Python-плагина
def on_vip(player: Player, days: int) -> None: ...

every #

every(seconds: float, *, delay: float | None = None) -> TaskDecorator

Периодическая задача, пока плагин работает; может быть async.

Параметры#
  • seconds — период в секундах.
  • delay — задержка перед первым запуском (по умолчанию — один период).
Пример#
@plugin.every(60)
def advert() -> None:
    server.print("!gСайт сервера: !texample.com")

include #

include() -> None

include(first: C, /, *rest: Router | Config) -> C

include(*components: Router | Config) -> Any

Подключить роутеры модулей и конфиги; возвращает первый компонент.

Пример#
cfg = plugin.include(Settings(), perks.router, api.router)
cfg.hp_bonus  # тип cvar'а известен редактору

spawn #

spawn(coro: Coroutine[Any, Any, R]) -> asyncio.Task[R]

Запустить фоновую задачу плагина.

Задача отменяется при выгрузке или перезагрузке плагина; запущенная из обработчика игрока — ещё и при его выходе.

Пример#
plugin.spawn(countdown(player))

offload #

offload(fn: Callable[[*Ts], R], *args: *Ts) -> R

Выполнить блокирующую функцию в потоке, не задерживая кадр сервера.

Пример#
data = await plugin.offload(requests.get, "https://example.com/api")

wait_for #

wait_for(
    event: type[E],
    *,
    check: Callable[[E], bool] | None = None,
    timeout: float | None = None,
) -> E

Дождаться события и вернуть копию его полей.

Параметры#
  • event — класс события.
  • check — условие; событие без совпадения пропускается.
  • timeout — секунд ожидания; по истечении — TimeoutError.
Возвращает#

Копия полей события (методы вроде cancel() у копии не работают).

Пример#
e = await plugin.wait_for(events.Say, check=lambda e: e.player is player, timeout=30)
player.print(f"Вы написали: {e.text}")

Forward #

Forward(name: str, params: list[tuple[str, str]])

Форвард — событие плагина для Pawn- и Python-подписчиков.

Объявляется через Router.forward; в сгенерированном .inc появляется forward <имя>(...), Pawn-плагины ловят его обычной public-функцией, Python-плагины — Router.on_forward.

Пример#
granted = plugin.forward("vip_granted", player=Player, days=int)
granted.emit(player, 30)

emit #

emit(*args: Any) -> int

Разослать форвард всем подписчикам.

Параметры#
  • *args — значения в порядке объявления параметров.
Возвращает#

Результат ExecuteForward: что вернули Pawn-обработчики (0, если их нет).

Исключения#
  • TypeError — число аргументов не совпадает с объявлением.