"""
Libyra UI — API для ботов: экраны, витрины, живые графики, анимации.
Один файл, только стандартная библиотека. Кладёте рядом со своим ботом и
пишете бота как отдельную программу: экраны собираются из элементов, нажатия
приходят в ваши обработчики, живые данные обновляются сами.
from libyra_ui import Bot, ui
bot = Bot(token="...")
@bot.start
def hello(chat):
chat.show(ui.screen("menu", [
ui.hero("Мой магазин", gradient=["#22D3EE", "#7C5CFF"]),
ui.buttons([ui.button("Витрина", "SHOP", animation="bounce")]),
]))
@bot.on("SHOP")
def shop(chat, payload):
...
bot.run()
Что умеет экран (подробно — в LIBYRA_UI.md рядом)
--------------------------------------------------
текст с оформлением hero, title, headline, body, caption, price, mono, quote;
внутри текста **жирный**, _курсив_ и `код`
раскладка row, column, card (с градиентом или «стеклом»)
мелочи badge, divider, progress, stat
данные ticker (котировки), chart: line, area, bars, candles
торговля showcase (витрина), product (карточка товара)
кнопки стили primary, outline, glass, gradient;
отклик ripple, bounce, jelly, pulse, glow, shake;
исчезновение сообщения после нажатия: sand, fade, shrink
ввод и списки input (текст, телефон, почта, число), each (строки списка,
columns — сеткой), push / remove / update / clear
экран как программа timer (тик раз в N секунд), clock (часы телефона),
rand / if / index_of в выражениях — игра и логика без бота
рисование canvas с фигурами по координатам: circle, rect, line, text;
animate даёт `frame`, нажатие — `tap_x` и `tap_y`
жесты swipe (смахнуть влево/вправо, долгое нажатие),
on_drag у холста — рисование пальцем
фотографии bot.upload_photo(файл) → номер, ui.photo(номер),
photo= у товара и плитки витрины; chat.show(..., photos=[...])
звуки sound= у кнопки и жеста, ui.sound(...) по условию;
свой файл — bot.upload_sound(файл) и sound="media:номер"
мини-приложения bot.upload_app(папка) → номер, ui.app(номер) — своя
программа целиком: любой HTML/JS, окно без сети
разговор с программой @bot.method("имя") — программа зовёт и ждёт ответ;
chat.app_event(...) — бот сам шлёт ей событие
Python-программы @bot.program("имя") или bot.script("имя", "game.py") —
консольная программа на Python в окне у человека: print,
input, кнопки выбора; chat.open_program("имя")
Экран с одним `id` — одно живое сообщение: присланный снова, он переезжает
вниз, к своей последней версии, а прежняя исчезает. Так живут биржевые
графики (`chat.live(...)`) и повторные меню. Нужны отдельные сообщения —
разные `id`.
"""
from __future__ import annotations
import base64
import codecs
import fnmatch
import io
import json
import logging
import os
import queue
import re
import secrets
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.request
import zipfile
from pathlib import Path
from typing import Any, Callable, Iterable, Sequence
UI_PREFIX = "LIBYRA_UI_V1\n"
CLICK_PREFIX = "LIBYRA_BUTTON_CLICK_V1\n"
#: Данные из мини-приложения: конверт одинаков с телефона и из окна.
APP_DATA_PREFIX = "LIBYRA_WEBAPP_DATA_V1\n"
#: Разговор программы с ботом: вызов с номером, ответ тем же номером, события.
APP_PROTO_PREFIX = "LIBYRA_APP_V1\n"
DEFAULT_SERVER = "https://libyra.su/backend"
LOG = logging.getLogger("libyra-ui")
#: Стили и отклики, которые понимает клиент. Опечатка здесь — ошибка сразу,
#: а не молча обычная кнопка в чате.
BUTTON_STYLES = {"primary", "outline", "glass", "gradient"}
ANIMATIONS = {"ripple", "bounce", "jelly", "pulse", "glow", "shake"}
DISMISS = {"", "sand", "fade", "shrink"}
INPUT_KINDS = {"text", "multiline", "number", "phone", "email"}
CLOCK_FORMATS = {"time", "seconds", "date", "full"}
#: Встроенные звуки: их телефон синтезирует сам, поэтому они мгновенные.
SOUNDS = {"click", "tap", "pop", "success", "error", "warning",
"coin", "tick", "alarm", "whoosh", "chime"}
def _check_sound(sound: str) -> str:
"""Имя встроенного звука или `media:<номер>` своего файла."""
if not sound or sound.startswith("media:"):
return sound
if sound not in SOUNDS:
raise ValueError(f"звук {sound!r} не из {sorted(SOUNDS)}; свой файл — media:<номер>")
return sound
TEXT_STYLES = {"hero", "title", "headline", "body", "caption", "price", "mono", "quote"}
CHART_KINDS = {"line", "area", "bars", "candles"}
ENTER = {"rise", "fade", "scale", "cascade"}
class ui: # noqa: N801 — пространство имён построителей, пишется как ui.text(...)
"""Построители элементов экрана. Каждый возвращает словарь."""
# ── текст ────────────────────────────────────────────────
@staticmethod
def text(text: str, style: str = "body", *, color: str = "", align: str = "start",
gradient: Sequence[str] = ()) -> dict:
if style not in TEXT_STYLES:
raise ValueError(f"стиль текста {style!r} не из {sorted(TEXT_STYLES)}")
node: dict[str, Any] = {"type": "text", "text": str(text), "style": style, "align": align}
if color:
node["color"] = color
if gradient:
node["gradient"] = list(gradient)
return node
@staticmethod
def hero(text: str, **kwargs) -> dict:
return ui.text(text, "hero", **kwargs)
@staticmethod
def title(text: str, **kwargs) -> dict:
return ui.text(text, "title", **kwargs)
@staticmethod
def caption(text: str, **kwargs) -> dict:
return ui.text(text, "caption", **kwargs)
# ── раскладка ────────────────────────────────────────────
@staticmethod
def row(*children: dict, gap: float = 8) -> dict:
return {"type": "row", "children": list(children), "gap": gap}
@staticmethod
def column(*children: dict, gap: float = 8) -> dict:
return {"type": "column", "children": list(children), "gap": gap}
@staticmethod
def card(*children: dict, gradient: Sequence[str] = (), border: str = "",
glass: bool = False) -> dict:
node: dict[str, Any] = {"type": "card", "children": list(children), "glass": glass}
if gradient:
node["gradient"] = list(gradient)
if border:
node["border"] = border
return node
# ── мелочи ───────────────────────────────────────────────
@staticmethod
def app(media: str, title: str = "", *, subtitle: str = "", icon: str = "🧩",
label: str = "Открыть", fullscreen: bool = False) -> dict:
"""Карточка мини-приложения: своя программа целиком.
`media` — номер набора файлов из `bot.upload_app`. Программа
открывается в отдельном окне **без сети**: всё, что ей нужно, лежит
внутри набора, а наружу окно не ходит вовсе. С ботом она говорит
сообщениями через свой мост.
Не забудьте `chat.show(screen, apps=[media])` — иначе сервер не
отдаст набор получателю.
"""
node = {"type": "app", "media": str(media), "title": title,
"subtitle": subtitle, "icon": icon, "label": label}
if fullscreen:
node["fullscreen"] = True
return node
@staticmethod
def game(media: str, title: str = "", *, subtitle: str = "", icon: str = "🎮",
label: str = "Играть") -> dict:
"""Сенсорная HTML/JS-игра из upload_app, на весь экран Android.
Касания и кадры обрабатываются локально, не сообщениями боту.
Паспорт и ограничения набора остаются теми же. Старый клиент
откроет обычное окно. Не забудьте chat.show(..., apps=[media]).
"""
return ui.app(media, title, subtitle=subtitle, icon=icon,
label=label, fullscreen=True)
@staticmethod
def sound(play: str, *, play_if: str = "", volume: float = 0.8) -> dict:
"""Звук экрана: играет, когда условие становится истинным.
ui.sound("alarm", play_if="left == 0") — сигнал ровно один раз, когда
отсчёт дошёл до нуля, а не на каждом тике. Без условия — один раз при
появлении экрана.
Встроенные: click, tap, pop, success, error, warning, coin, tick,
alarm, whoosh, chime. Свой файл — "media:<номер>" из
`bot.upload_sound`; не забудьте `chat.show(..., sounds=[номер])`.
"""
return {"type": "sound", "play": _check_sound(play), "play_if": play_if,
"volume": max(0.0, min(1.0, volume))}
@staticmethod
def photo(media: str, *, ratio: float = 1.6, fit: str = "crop") -> dict:
"""Фотография с нашего сервера: `media` — номер из `bot.upload_photo`.
Не ссылка: адресов наружу экран бота не открывает. Картинку увидит
только тот, кому бот прислал её в сообщении — не забудьте
`chat.show(screen, photos=[media])`.
`ratio` — отношение сторон (1 — квадрат), `fit`: crop (обрезать по
краям) или fit (вписать целиком).
"""
if fit not in {"crop", "fit"}:
raise ValueError("fit бывает crop или fit")
return {"type": "photo", "media": str(media), "ratio": ratio, "fit": fit}
@staticmethod
def badge(text: str, color: str = "") -> dict:
node = {"type": "badge", "text": text}
if color:
node["color"] = color
return node
@staticmethod
def divider() -> dict:
return {"type": "divider"}
@staticmethod
def progress(value: float, label: str = "", color: str = "") -> dict:
node: dict[str, Any] = {"type": "progress", "value": max(0.0, min(1.0, value)), "label": label}
if color:
node["color"] = color
return node
@staticmethod
def stat(label: str, value: str, delta: float | None = None, suffix: str = "%") -> dict:
node: dict[str, Any] = {"type": "stat", "label": label, "value": value, "suffix": suffix}
if delta is not None:
node["delta"] = round(delta, 2)
return node
# ── данные ───────────────────────────────────────────────
@staticmethod
def ticker(items: Iterable[tuple[str, str, float]]) -> dict:
"""Котировки: (символ, цена, изменение в процентах)."""
return {"type": "ticker", "items": [
{"symbol": s, "price": p, "delta": round(d, 2)} for s, p, d in items]}
@staticmethod
def chart(kind: str, values: Sequence[float] = (), *,
candles: Sequence[Sequence[float]] = (), color: str = "",
height: float = 150, labels: Sequence[str] = ()) -> dict:
if kind not in CHART_KINDS:
raise ValueError(f"вид графика {kind!r} не из {sorted(CHART_KINDS)}")
if kind == "candles" and not candles:
raise ValueError("для свечей нужен candles=[[open, high, low, close], ...]")
if kind != "candles" and not values:
raise ValueError("для графика нужны values")
node: dict[str, Any] = {"type": "chart", "kind": kind, "height": height}
if values:
node["values"] = [round(float(v), 4) for v in values]
if candles:
node["candles"] = [[round(float(x), 4) for x in c[:4]] for c in candles]
if color:
node["color"] = color
if labels:
node["labels"] = list(labels)
return node
# ── торговля ─────────────────────────────────────────────
@staticmethod
def product(title: str, price: str, *, subtitle: str = "", old_price: str = "",
rating: float = 0, reviews: int = 0, emoji: str = "", photo: str = "",
gradient: Sequence[str] = (), badges: Sequence[str | dict] = (),
specs: Sequence[tuple[str, str]] = (), description: str = "",
stock: int | None = None, actions: Sequence[dict] = ()) -> dict:
node: dict[str, Any] = {
"type": "product", "title": title, "price": price, "subtitle": subtitle,
"old_price": old_price, "rating": rating, "reviews": reviews, "emoji": emoji,
"photo": photo,
"badges": list(badges), "specs": [list(s) for s in specs],
"description": description, "actions": list(actions),
}
if gradient:
node["gradient"] = list(gradient)
if stock is not None:
node["stock"] = stock
return node
@staticmethod
def showcase(items: Sequence[dict], title: str = "", columns: int = 2) -> dict:
return {"type": "showcase", "title": title, "columns": max(1, min(3, columns)),
"items": list(items)}
@staticmethod
def item(title: str, price: str, payload: str, *, emoji: str = "",
gradient: Sequence[str] = (), badge: str = "", photo: str = "") -> dict:
node: dict[str, Any] = {"title": title, "price": price, "payload": payload,
"emoji": emoji, "badge": badge, "photo": photo}
if gradient:
node["gradient"] = list(gradient)
return node
# ── кнопки ───────────────────────────────────────────────
@staticmethod
def button(label: str, payload: str = "", *, style: str = "primary", animation: str = "ripple",
icon: str = "", color: str = "", dismiss: str = "", local: bool = False,
set: dict | None = None, inc: dict | Sequence[dict] | None = None, # noqa: A002
toggle: str | Sequence[str] | None = None, enabled_if: str = "",
lists: dict | Sequence[dict] | None = None, sound: str = "") -> dict:
"""Кнопка. `set`/`inc`/`toggle`/`lists` меняют переменные экрана прямо в телефоне.
`local=True` — боту не уходит ничего: вкладка, счётчик, раскрытие. Без
`local` нажатие уходит боту вместе со всеми переменными экрана — их
видно в `chat.values`.
`set={"total": "=qty * price"}` — строка с «=» в начале считается в
момент нажатия; без него записывается как есть.
`lists=ui.push(...)` (или список таких) — добавить, убрать, поменять запись.
`enabled_if="len(trim(name)) >= 2"` — кнопка приглушена и не жмётся,
пока условие ложно.
"""
list_ops = [lists] if isinstance(lists, dict) else list(lists or [])
if not payload and not (set or inc or toggle or list_ops):
raise ValueError("кнопке нужен payload или действие на месте (set/inc/toggle/lists)")
if style not in BUTTON_STYLES:
raise ValueError(f"стиль кнопки {style!r} не из {sorted(BUTTON_STYLES)}")
if animation not in ANIMATIONS:
raise ValueError(f"отклик {animation!r} не из {sorted(ANIMATIONS)}")
if dismiss not in DISMISS:
raise ValueError(f"исчезновение {dismiss!r} не из {sorted(DISMISS - {''})}")
node: dict[str, Any] = {"type": "button", "label": label, "payload": payload,
"style": style, "animation": animation, "icon": icon,
"dismiss": dismiss, "local": local or not payload}
if color:
node["color"] = color
if set:
node["set"] = dict(set)
if inc:
node["inc"] = inc if isinstance(inc, dict) else list(inc)
if toggle:
node["toggle"] = toggle if isinstance(toggle, str) else list(toggle)
if enabled_if:
node["enabled_if"] = enabled_if
if sound:
node["sound"] = _check_sound(sound)
for op in list_ops:
kind = op["op"]
body = {k: v for k, v in op.items() if k != "op"}
node.setdefault(kind, []).append(body)
return node
# ── списки ───────────────────────────────────────────────
@staticmethod
def push(list: str, item: Any, unique: str = "") -> dict: # noqa: A002
"""Добавить запись в список. Поля-строки с «=» — выражения:
ui.push("cart", {"id": "=color", "title": "Наушники", "qty": "=qty"}, unique="id").
С `unique` запись с тем же полем заменяется, а не повторяется."""
op: dict[str, Any] = {"op": "push", "list": list, "item": item}
if unique:
op["unique"] = unique
return op
@staticmethod
def remove(list: str, *, index: Any = None, where: str = "") -> dict: # noqa: A002
"""Убрать записи: по номеру (index="=index" — строка, где нажали) или по условию
(where="it.id == item.id")."""
if index is None and not where:
raise ValueError("ui.remove: нужен index или where")
op: dict[str, Any] = {"op": "remove", "list": list}
if index is not None:
op["index"] = index
if where:
op["where"] = where
return op
@staticmethod
def update(list: str, set: dict, *, index: Any = None, where: str = "") -> dict: # noqa: A002
"""Поменять поля записей; текущая запись в выражениях — `it`:
ui.update("cart", {"qty": "=it.qty + 1"}, index="=index")."""
if index is None and not where:
raise ValueError("ui.update: нужен index или where")
op: dict[str, Any] = {"op": "update", "list": list, "set": dict(set)}
if index is not None:
op["index"] = index
if where:
op["where"] = where
return op
@staticmethod
def clear(list: str) -> dict: # noqa: A002
return {"op": "clear", "list": list}
@staticmethod
def each(list: str, *children: dict, empty: Sequence[dict] = (), as_: str = "item", # noqa: A002
gap: float = 8, columns: int = 1, aspect: float = 0) -> dict:
"""Строки списка: `children` для каждой записи. Запись — `item` (или `as_`),
номер — `index`: ui.text("{item.title} × {item.qty}").
`columns` больше одной — записи идут сеткой: поле игры, календарь,
плитки. `aspect=1` делает ячейку квадратной."""
if not children:
raise ValueError("ui.each без содержимого строки")
node: dict[str, Any] = {"type": "each", "list": list, "as": as_,
"children": list_(children), "gap": gap,
"columns": max(1, min(6, columns))}
if aspect:
node["aspect"] = aspect
if empty:
node["empty"] = list_(empty)
return node
# ── экран как программа ──────────────────────────────────
@staticmethod
def timer(every: float = 1, *, while_: str = "", limit: int = 0, set: dict | None = None, # noqa: A002
inc: dict | Sequence[dict] | None = None, toggle: str | Sequence[str] | None = None,
lists: dict | Sequence[dict] | None = None) -> dict:
"""Тик: раз в `every` секунд меняет переменные экрана сам, без бота.
ui.timer(1, inc=ui.inc("score", 1), while_="running") — счёт растёт,
пока «running» истинно. Тикает, только пока экран на виду.
Чаще пяти раз в секунду нельзя: это уже расход заряда, а не логика.
"""
if every < 0.2:
raise ValueError("тик чаще пяти раз в секунду не нужен даже игре")
node: dict[str, Any] = {"type": "timer", "every": every}
if while_:
node["while"] = while_
if limit:
node["limit"] = int(limit)
if set:
node["set"] = dict(set)
if inc:
node["inc"] = inc if isinstance(inc, dict) else list_(inc)
if toggle:
node["toggle"] = toggle if isinstance(toggle, str) else list_(toggle)
for op in ([lists] if isinstance(lists, dict) else list_(lists or [])):
node.setdefault(op["op"], []).append({k: v for k, v in op.items() if k != "op"})
if not (set or inc or toggle or lists):
raise ValueError("тику нужно, что менять: set / inc / toggle / lists")
return node
# ── холст ────────────────────────────────────────────────
@staticmethod
def canvas(width: float, height: float, shapes: Sequence[dict], *, background: str = "",
animate: bool = False, on_tap: dict | None = None,
on_drag_start: dict | None = None, on_drag: dict | None = None,
on_drag_end: dict | None = None) -> dict:
"""Холст: фигуры по координатам в своей сетке `width` × `height`.
Холст растягивается по ширине сообщения, поэтому в пикселях думать не
нужно. Числа фигур можно задавать выражениями — строкой с «=» в
начале: ui.circle("=50 + 40 * sin(frame)", 50, 6).
`animate=True` — у выражений появляется `frame`: секунды с появления
холста дробным числом. Это и есть плавное движение.
`on_tap=ui.tap(...)` — нажатие; куда попал палец, видно в `tap_x`
и `tap_y` в тех же единицах. `on_drag_start` срабатывает, когда
палец коснулся и повёл, `on_drag` — пока ведёт (те же `tap_x`,
`tap_y`), `on_drag_end` — когда его оторвали.
Рисование пальцем забирает прокрутку чата на площади холста.
"""
if not shapes:
raise ValueError("пустой холст рисовать нечем")
node: dict[str, Any] = {"type": "canvas", "width": width, "height": height,
"shapes": list_(shapes), "animate": animate}
if background:
node["background"] = background
if on_tap:
node["on_tap"] = on_tap
if on_drag_start:
node["on_drag_start"] = on_drag_start
if on_drag:
node["on_drag"] = on_drag
if on_drag_end:
node["on_drag_end"] = on_drag_end
return node
@staticmethod
def swipe(*children: dict, left: dict | None = None, right: dict | None = None,
up: dict | None = None, down: dict | None = None, hold: dict | None = None,
threshold: float = 0.28) -> dict:
"""Смахивание и долгое нажатие. Действия — такие же, как `ui.tap(...)`.
ui.swipe(ui.card(...), left=ui.tap(lists=ui.remove("cart", index="=index")),
right=ui.tap("LIKE:{item.id}"))
Содержимое едет за пальцем и пружинит назад, если жест не доведён;
`threshold` — какую долю ширины надо пройти, чтобы жест засчитали.
Вверх и вниз по умолчанию не ловятся: этим движением прокручивают
переписку. Задали `up` или `down` — прокрутка на этом месте отдаётся
жесту, и это осознанный выбор бота.
"""
if not children:
raise ValueError("ui.swipe без содержимого")
if not (left or right or up or down or hold):
raise ValueError("ui.swipe без единого действия")
node: dict[str, Any] = {"type": "swipe", "children": list_(children),
"threshold": threshold}
for name, value in (("left", left), ("right", right), ("up", up), ("down", down),
("hold", hold)):
if value:
node[name] = value
return node
@staticmethod
def circle(x: Any, y: Any, r: Any, color: str = "", **kw: Any) -> dict:
return _shape("circle", x=x, y=y, r=r, color=color, **kw)
@staticmethod
def rect(x: Any, y: Any, w: Any, h: Any, color: str = "", **kw: Any) -> dict:
return _shape("rect", x=x, y=y, w=w, h=h, color=color, **kw)
@staticmethod
def line(x1: Any, y1: Any, x2: Any, y2: Any, color: str = "", **kw: Any) -> dict:
return _shape("line", x1=x1, y1=y1, x2=x2, y2=y2, color=color, **kw)
@staticmethod
def label(x: Any, y: Any, text: str, color: str = "", **kw: Any) -> dict:
"""Надпись на холсте. Ставится серединой в точку, если не задан align."""
return _shape("text", x=x, y=y, text=text, color=color, **kw)
@staticmethod
def tap(payload: str = "", *, local: bool = True, set: dict | None = None, # noqa: A002
inc: dict | Sequence[dict] | None = None, toggle: str | Sequence[str] | None = None,
lists: dict | Sequence[dict] | None = None, sound: str = "") -> dict:
"""Нажатие по холсту: что поменять на месте и уходит ли оно боту."""
node: dict[str, Any] = {"payload": payload, "local": local or not payload}
if sound:
node["sound"] = _check_sound(sound)
if set:
node["set"] = dict(set)
if inc:
node["inc"] = inc if isinstance(inc, dict) else list_(inc)
if toggle:
node["toggle"] = toggle if isinstance(toggle, str) else list_(toggle)
for op in ([lists] if isinstance(lists, dict) else list_(lists or [])):
node.setdefault(op["op"], []).append({k: v for k, v in op.items() if k != "op"})
if not (payload or set or inc or toggle or lists or sound):
raise ValueError("нажатию по холсту нужно действие, звук или payload")
return node
@staticmethod
def clock(format: str = "time", *, label: str = "", offset: int | None = None) -> dict: # noqa: A002
"""Часы телефона: time (ЧЧ:ММ), seconds, date, full.
`offset` — сдвиг от Гринвича в минутах, если показываете время другого города."""
if format not in CLOCK_FORMATS:
raise ValueError(f"формат часов {format!r} не из {sorted(CLOCK_FORMATS)}")
node: dict[str, Any] = {"type": "clock", "format": format, "label": label}
if offset is not None:
node["offset"] = int(offset)
return node
# ── ввод ─────────────────────────────────────────────────
@staticmethod
def input(var: str, label: str = "", placeholder: str = "", kind: str = "text", # noqa: A002
max: int = 120) -> dict: # noqa: A002
"""Поле ввода. Написанное уходит боту с нажатием кнопки — в `chat.values[var]`.
kind: text, multiline, number, phone, email. Поля для паролей нет намеренно."""
if kind not in INPUT_KINDS:
raise ValueError(f"вид поля {kind!r} не из {sorted(INPUT_KINDS)}")
return {"type": "input", "var": var, "label": label, "placeholder": placeholder,
"kind": kind, "max": int(max)}
@staticmethod
def inc(var: str, by: float = 1, min: float | None = None, # noqa: A002
max: float | None = None) -> dict: # noqa: A002
"""Прибавить к переменной (с границами): ui.button("+", inc=ui.inc("qty", 1, max=10))."""
node: dict[str, Any] = {"var": var, "by": by}
if min is not None:
node["min"] = min
if max is not None:
node["max"] = max
return node
# ── логика на месте ──────────────────────────────────────
@staticmethod
def when(condition: str, *nodes: dict) -> dict:
"""Показывать, пока выражение истинно: ui.when("tab == 'specs'", ...).
Выражения: числа, 'строки', переменные экрана, + - * / %,
== != < > <= >=, && || !, скобки. Никакого кода — только калькулятор.
"""
if not nodes:
raise ValueError("ui.when без содержимого")
node = dict(nodes[0]) if len(nodes) == 1 else {"type": "column", "children": list(nodes), "gap": 8}
node["show_if"] = condition
return node
@staticmethod
def tabs(var: str, options: Sequence[str | tuple[str, str]]) -> dict:
"""Вкладки: переключают переменную `var` на месте. options — значения или (значение, подпись)."""
return {"type": "tabs", "var": var, "options": _options(options)}
@staticmethod
def choice(var: str, options: Sequence[str | tuple[str, str]], label: str = "") -> dict:
"""Выбор одного варианта плашками: размер, цвет, тариф."""
return {"type": "choice", "var": var, "label": label, "options": _options(options)}
@staticmethod
def toggle(var: str, label: str, hint: str = "") -> dict:
return {"type": "toggle", "var": var, "label": label, "hint": hint}
@staticmethod
def stepper(var: str, label: str = "", min: float = 0, max: float = 99, # noqa: A002
step: float = 1) -> dict:
"""Счётчик «− N +»."""
if max < min:
raise ValueError("у счётчика max меньше min")
return {"type": "stepper", "var": var, "label": label, "min": min, "max": max, "step": step}
@staticmethod
def countdown(seconds: int, label: str = "", on_end: dict | None = None) -> dict:
"""Обратный отсчёт в телефоне. on_end — что поменять по окончании: {"set": {...}}."""
node: dict[str, Any] = {"type": "countdown", "seconds": int(seconds), "label": label}
if on_end:
node["on_end"] = on_end
return node
@staticmethod
def lottie(data: "dict | str | Path", *, loop: bool = True, speed: float = 1.0,
height: float = 160, play_if: str = "") -> dict:
"""Анимация Lottie: словарь, строка JSON или путь к файлу .json.
Картинки внутри должны быть вшиты (base64): по ссылкам приложение
ничего не загружает.
"""
if isinstance(data, Path) or (isinstance(data, str) and data.endswith(".json")
and not data.lstrip().startswith("{")):
data = json.loads(Path(data).read_text(encoding="utf-8"))
if isinstance(data, str):
data = json.loads(data)
if not isinstance(data, dict) or "layers" not in data:
raise ValueError("это не Lottie: нет слоёв (layers)")
size = len(json.dumps(data, separators=(",", ":")))
if size > 400_000:
raise ValueError(f"анимация {size // 1024} КБ — больше 390 КБ приложение не примет")
return {"type": "lottie", "data": data, "loop": loop, "speed": speed,
"height": height, "play_if": play_if}
@staticmethod
def buttons(items: Sequence[dict], columns: int = 2) -> dict:
return {"type": "buttons", "columns": max(1, min(3, columns)), "items": list(items)}
# ── экран ────────────────────────────────────────────────
@staticmethod
def screen(id: str, body: Sequence[dict], *, accent: str = "#22D3EE", # noqa: A002
background: str = "#0B1B2A", surface: str = "#10243A", radius: float = 18,
enter: str = "rise", state: dict | None = None,
force: Sequence[str] = ()) -> str:
"""Готовый текст сообщения. `id` — чтобы потом обновлять экран на месте."""
if enter not in ENTER:
raise ValueError(f"появление {enter!r} не из {sorted(ENTER)}")
if not body:
raise ValueError("пустой экран клиент не покажет")
spec = {
"id": id,
"theme": {"accent": accent, "background": background, "surface": surface,
"radius": radius},
"enter": enter,
"body": list(body),
}
if state:
# Переменные экрана: их меняют вкладки, счётчики и кнопки на месте.
spec["state"] = dict(state)
if force:
# Эти переменные новая версия экрана ставит поверх выбора человека:
# например, корзину, которую бот пересобрал.
spec["force"] = [str(name) for name in force]
return UI_PREFIX + json.dumps(spec, ensure_ascii=False, separators=(",", ":"))
list_ = list # внутри класса ui имя list занято параметрами построителей
#: Поля фигуры, которые холст понимает: остальное он просто не читает.
SHAPE_FIELDS = {"x", "y", "r", "w", "h", "x1", "y1", "x2", "y2", "text", "size", "align",
"color", "alpha", "stroke", "width", "radius"}
def _shape(kind: str, *, list: str = "", as_: str = "item", show_if: str = "", # noqa: A002
**fields: Any) -> dict:
"""Фигура холста. `list` повторяет её для каждой записи списка."""
unknown = set(fields) - SHAPE_FIELDS
if unknown:
raise ValueError(f"холст не знает полей {sorted(unknown)}")
node: dict[str, Any] = {"kind": kind}
node.update({key: value for key, value in fields.items() if value != ""})
if list:
node["list"] = list
node["as"] = as_
if show_if:
node["show_if"] = show_if
return node
def _options(options: Sequence[str | tuple[str, str]]) -> list[dict]:
result = []
for option in options:
value, label = (option, option) if isinstance(option, str) else option
result.append({"value": str(value), "label": str(label)})
if not result:
raise ValueError("нужен хотя бы один вариант")
return result
# ─────────────────────────────────────────────────────────────
# Бот как программа
# ─────────────────────────────────────────────────────────────
class Chat:
"""Собеседник: куда отвечать и что о нём помнить между нажатиями."""
def __init__(self, bot: "Bot", chat_id: str) -> None:
self.bot = bot
self.id = chat_id
#: Память бота об этом собеседнике: корзина, шаг анкеты, что угодно.
self.state: dict[str, Any] = bot.states.setdefault(chat_id, {})
#: Переменные экрана в момент нажатия: что человек выбрал на месте.
self.values: dict[str, Any] = {}
#: Строка списка, где нажата кнопка: {"item": {...}, "index": 2}.
self.scope: dict[str, Any] = {}
#: Номер программы, если это её вызов: ей же уйдут события.
self.app: str = ""
#: Номер вызова программы: по нему ответ можно прислать позже.
self.call: Any = None
def show(self, screen: str, photos: Sequence[str] = (), sounds: Sequence[str] = (),
apps: Sequence[str] = ()) -> None:
"""Показать экран. `photos` и `sounds` — номера файлов, которые на нём есть.
Прикрепление и есть выдача доступа: без него сервер не отдаст файл
получателю — вместо фотографии останется пустая плашка, а свой звук
промолчит.
"""
self.bot.send(self.id, screen,
photos=list_(photos) + list_(sounds) + list_(apps))
def say(self, text: str) -> None:
self.bot.send(self.id, text)
def app_event(self, name: str, data: Any = None, app: str = "") -> None:
"""Событие открытой программе: бот шлёт его сам, без вопроса.
Программа ловит его через `LibyraWebApp.on('имя', ...)`. Если она
сейчас закрыта, событие просто не дойдёт: очереди у него нет — для
важного есть обычное сообщение боту в переписку.
"""
target = app or self.app
if not target:
return
self.bot.send(self.id, APP_PROTO_PREFIX + json.dumps(
{"app": target, "e": str(name)[:64],
"p": json.dumps(data if data is not None else {}, ensure_ascii=False)},
ensure_ascii=False,
))
def open_program(self, name: str, *, text: str = "", label: str = "Запустить") -> None:
"""Прислать карточку Python-программы из `@bot.program` или `bot.script`.
Окно программы собирается и загружается само при первом показе.
"""
node, media = self.bot.program_card(name, label=label)
body = [ui.text(text)] if text else []
self.show(ui.screen("program-" + name, body + [node]), apps=[media])
def live(self, key: str, render: Callable[[int], str | None], every: float = 2.0,
times: int = 60) -> None:
"""Обновлять экран на месте: `render(шаг)` возвращает экран с тем же id.
Вернёт None — обновление останавливается. Повторный вызов с тем же
`key` останавливает прежнее: иначе два потока гнали бы один график.
"""
self.bot.start_live(self, key, render, every, times)
def stop_live(self, key: str) -> None:
self.bot.stop_live(self.id, key)
class Bot:
def __init__(self, token: str, server: str = DEFAULT_SERVER) -> None:
self.token = token
self.server = server.rstrip("/")
# Мимо системного прокси: на машине разработчика он для туннеля SSH,
# и TLS до сервера через него не встаёт.
self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
self.states: dict[str, dict[str, Any]] = {}
self.handlers: list[tuple[str, Callable[[Chat, str], None]]] = []
self.start_handler: Callable[[Chat], None] | None = None
self.text_handler: Callable[[Chat, str], None] | None = None
self.app_handler: Callable[[Chat, str], None] | None = None
#: Методы, которые программа может позвать: имя → обработчик.
self.methods: dict[str, Callable[[Chat, Any], Any]] = {}
self.lives: dict[tuple[str, str], threading.Event] = {}
self.stop = threading.Event()
self.me: dict[str, Any] = {}
#: Python-программы: имя → описание; сеансы — по (собеседник, имя).
self.programs: dict[str, ProgramSpec] = {}
self.sessions: dict[tuple[str, str], ProgramSession] = {}
self.program_apps: dict[str, str] = {}
#: Сколько программ может работать сразу: каждая — поток или процесс
#: на машине бота, а сообщений у бота всего 60 в минуту на всех.
self.max_programs = 8
# ── регистрация обработчиков ─────────────────────────────
def start(self, handler: Callable[[Chat], None]) -> Callable[[Chat], None]:
"""Первое сообщение и /start."""
self.start_handler = handler
return handler
def text(self, handler: Callable[[Chat, str], None]) -> Callable[[Chat, str], None]:
"""Любой текст, если не задан — показывается стартовый экран."""
self.text_handler = handler
return handler
def method(self, name: str) -> Callable:
"""Метод, который зовёт программа: `await LibyraWebApp.call('имя', {...})`.
Что вернёт обработчик, то и придёт программе ответом — словарь,
список, число. Ошибку не прячем: программа получит {"error": "..."}
и сможет её показать, а не молча ждать.
"""
def register(handler: Callable[[Chat, Any], Any]) -> Callable[[Chat, Any], Any]:
self.methods[name] = handler
return handler
return register
def app(self, handler: Callable[[Chat, str], None]) -> Callable[[Chat, str], None]:
"""Данные из мини-приложения: то, что оно отправило через `sendData`."""
self.app_handler = handler
return handler
# ── Python-программы ─────────────────────────────────────
def program(self, name: str, *, title: str = "", subtitle: str = "", icon: str = "🐍",
fullscreen: bool = False, idle: float = 600) -> Callable:
"""Python-программа в окне у человека: функция получает `io`.
@bot.program("guess", title="Угадай число")
def guess(io):
secret = random.randint(1, 100)
while True:
answer = io.input("Ваше число: ")
...
Внутри работают и обычные `print()` и `input()` — уже написанную
консольную программу переделывать не нужно. Каждый собеседник
получает свой сеанс в отдельном потоке. `idle` — сколько секунд
ждать, если окно закрыли: потом программа останавливается.
"""
def register(target: Callable[[ProgramIO], Any]) -> Callable[[ProgramIO], Any]:
self._add_program(ProgramSpec(name, target=target, title=title, subtitle=subtitle,
icon=icon, fullscreen=fullscreen, idle=idle))
return target
return register
def script(self, name: str, path: "str | Path", *args: Any, title: str = "",
subtitle: str = "", icon: str = "🐍", fullscreen: bool = False,
idle: float = 600) -> None:
"""Готовый .py-файл как программа: свой процесс на каждого собеседника.
Запускается тем же Python, что и бот, из папки скрипта. Что скрипт
печатает — видно в окне, что человек вводит — приходит в `input()`.
Токен бота в окружение скрипта не попадает.
"""
script = Path(path).resolve()
if not script.is_file():
raise FileNotFoundError(f"нет файла программы: {script}")
self._add_program(ProgramSpec(name, script=script, args=[str(a) for a in args],
title=title or script.stem, subtitle=subtitle, icon=icon,
fullscreen=fullscreen, idle=idle))
def _add_program(self, spec: ProgramSpec) -> None:
self.programs[spec.name] = spec
self.program_apps.pop(spec.name, None)
for method, handler in (("py.start", self._py_start), ("py.wait", self._py_wait),
("py.input", self._py_input), ("py.stop", self._py_stop)):
self.methods.setdefault(method, handler)
def program_card(self, name: str, *, label: str = "Запустить") -> tuple[dict, str]:
"""Карточка программы и номер её окна. Окно загружается один раз."""
spec = self.programs.get(name)
if spec is None:
raise ValueError(f"программа {name!r} не зарегистрирована: @bot.program или bot.script")
media = self.program_apps.get(name)
if not media:
media = self._upload_app_bytes(_program_bundle(spec), f"program-{name}.zip")
self.program_apps[name] = media
return ui.app(media, spec.title, subtitle=spec.subtitle, icon=spec.icon, label=label,
fullscreen=spec.fullscreen), media
def _program_session(self, chat: Chat, params: Any) -> "ProgramSession | None":
params = params if isinstance(params, dict) else {}
session = self.sessions.get((chat.id, str(params.get("program") or "")))
if session is None or session.id != str(params.get("session") or ""):
return None
session.app = chat.app or session.app
return session
def _py_start(self, chat: Chat, params: Any) -> Any:
params = params if isinstance(params, dict) else {}
name = str(params.get("program") or "")
spec = self.programs.get(name)
if spec is None:
return {"error": f"нет программы {name}"}
key = (chat.id, name)
old = self.sessions.get(key)
if old is not None and not old.finished.is_set() and not params.get("restart"):
# Окно открыли снова: программа та же, показываем, что она успела.
old.app = chat.app or old.app
old.touch()
return old.snapshot(0)
if old is not None:
old.stop("restart")
now = time.monotonic()
for other_key, other in list(self.sessions.items()):
if other.finished.is_set() and now - other.ended_at > 1800:
del self.sessions[other_key]
if sum(1 for s in self.sessions.values() if not s.finished.is_set()) >= self.max_programs:
return {"error": "Сейчас запущено слишком много программ. Попробуйте через минуту."}
session = ProgramSession(self, spec, chat.id, chat.app)
self.sessions[key] = session
session.start()
LOG.info(" программа %s запущена для %s…", name, chat.id[:8])
return session.snapshot(0)
def _py_wait(self, chat: Chat, params: Any) -> Any:
session = self._program_session(chat, params)
if session is None:
return {"gone": True}
return session.wait(chat.call, _int(params.get("after")))
def _py_input(self, chat: Chat, params: Any) -> Any:
session = self._program_session(chat, params)
if session is None:
return {"gone": True}
session.feed(str(params.get("line") or ""))
return session.wait(chat.call, _int(params.get("after")))
def _py_stop(self, chat: Chat, params: Any) -> Any:
session = self._program_session(chat, params)
if session is not None:
session.stop("user")
return {"ok": True}
def on(self, pattern: str) -> Callable:
"""Нажатие с таким payload. Можно со звёздочкой: `on("BUY:*")`."""
def register(handler: Callable[[Chat, str], None]) -> Callable[[Chat, str], None]:
self.handlers.append((pattern, handler))
return handler
return register
# ── сеть ─────────────────────────────────────────────────
def call(self, method: str, path: str, payload: dict | None = None) -> dict:
body = None if payload is None else json.dumps(payload).encode("utf-8")
request = urllib.request.Request(self.server + path, data=body, method=method)
request.add_header("Authorization", "Bearer " + self.token)
request.add_header("Content-Type", "application/json")
with self.opener.open(request, timeout=30) as response:
return json.loads(response.read().decode("utf-8") or "{}")
def upload_app(self, source: "str | Path") -> str:
"""Кладёт мини-приложение на сервер и возвращает его номер.
Принимает папку (соберём zip сами) или готовый .zip. Внутри должен
быть `index.html` в корне и всё, что ему нужно: шрифты, картинки,
библиотеки. Ссылок наружу быть не должно — в телефоне у окна сети
нет, и они просто не загрузятся.
До 5 МБ в архиве и 20 МБ в распакованном виде.
"""
path = Path(source)
if path.is_dir():
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
for file in sorted(path.rglob("*")):
if file.is_file():
archive.write(file, file.relative_to(path).as_posix())
data = buffer.getvalue()
name = path.name + ".zip"
else:
data = path.read_bytes()
name = path.name
return self._upload_app_bytes(data, name)
def _upload_app_bytes(self, data: bytes, name: str) -> str:
if len(data) > 5 * 1024 * 1024:
raise ValueError(f"набор {len(data) // 1024} КБ — больше 5 МБ сервер не примет")
answer = self.call("POST", "/api/bot/uploadApp", {
"app_base64": base64.b64encode(data).decode("ascii"),
"filename": name,
})
media = str(answer.get("app_id") or "")
if not media:
raise RuntimeError(f"сервер не принял набор: {answer}")
LOG.info("мини-приложение загружено: %s (%d КБ)", media[:8], len(data) // 1024)
return media
def upload_sound(self, source: "str | bytes | Path") -> str:
"""Кладёт короткий звук на сервер и возвращает его номер.
Принимаются mp3, ogg, wav и m4a — по началу файла. До 2 МБ: это
сигнал на секунду-другую, а не музыка.
"""
data = source if isinstance(source, bytes) else Path(source).read_bytes()
if len(data) > 2 * 1024 * 1024:
raise ValueError(f"звук {len(data) // 1024} КБ — больше 2 МБ сервер не примет")
name = "sound.mp3" if isinstance(source, bytes) else Path(source).name
answer = self.call("POST", "/api/bot/uploadSound", {
"sound_base64": base64.b64encode(data).decode("ascii"),
"filename": name,
})
media = str(answer.get("audio_id") or "")
if not media:
raise RuntimeError(f"сервер не принял звук: {answer}")
LOG.info("звук загружен: %s (%d КБ)", media[:8], len(data) // 1024)
return media
def upload_photo(self, source: "str | bytes | Path") -> str:
"""Кладёт картинку на сервер и возвращает её номер.
Принимает путь к файлу или сами байты. Больше трёх мегабайт сервер
не примет — это переписка, а не фотохостинг.
"""
data = source if isinstance(source, bytes) else Path(source).read_bytes()
if len(data) > 3 * 1024 * 1024:
raise ValueError(f"картинка {len(data) // 1024} КБ — больше 3 МБ сервер не примет")
name = "photo.jpg" if isinstance(source, bytes) else Path(source).name
answer = self.call("POST", "/api/bot/uploadPhoto", {
"photo_base64": base64.b64encode(data).decode("ascii"),
"filename": name,
})
media = str(answer.get("image_id") or "")
if not media:
raise RuntimeError(f"сервер не принял картинку: {answer}")
LOG.info("картинка загружена: %s (%d КБ)", media[:8], len(data) // 1024)
return media
def send(self, chat_id: str, text: str, photos: Sequence[str] = ()) -> None:
try:
payload = {"to_pk": chat_id, "chat_id": chat_id, "text": text}
if photos:
payload["photo_ids"] = [str(x) for x in photos][:8]
answer = self.call("POST", "/api/bot/sendMessage", payload)
if answer.get("success") is False:
LOG.warning("Сервер не принял ответ для %s…: %s", chat_id[:8], answer.get("error"))
else:
LOG.info("→ %s… %s", chat_id[:8], _describe(text))
except urllib.error.HTTPError as error:
LOG.warning("Не отправлено %s…: HTTP %s", chat_id[:8], error.code)
def reply(self, chat_id: str, app: str, call_id: Any, result: Any) -> None:
"""Ответ программе на вызов `call_id` — в том числе отложенный."""
self.send(chat_id, APP_PROTO_PREFIX + json.dumps(
{"app": app, "id": call_id, "r": json.dumps(result, ensure_ascii=False)},
ensure_ascii=False,
))
# ── живые экраны ─────────────────────────────────────────
def start_live(self, chat: Chat, key: str, render: Callable[[int], str | None],
every: float, times: int) -> None:
self.stop_live(chat.id, key)
stop = threading.Event()
self.lives[(chat.id, key)] = stop
def loop() -> None:
for step in range(times):
if stop.is_set() or self.stop.is_set():
return
try:
screen = render(step)
except Exception: # noqa: BLE001
LOG.exception("Живой экран %s упал", key)
return
if screen is None:
return
self.send(chat.id, screen)
stop.wait(every)
threading.Thread(target=loop, name=f"live-{key}", daemon=True).start()
def stop_live(self, chat_id: str, key: str) -> None:
old = self.lives.pop((chat_id, key), None)
if old:
old.set()
# ── разбор входящих ──────────────────────────────────────
def dispatch(self, chat_id: str, text: str) -> None:
chat = Chat(self, chat_id)
LOG.info("← %s… %s", chat_id[:8], _describe(text))
if text.startswith(CLICK_PREFIX):
try:
click = json.loads(text[len(CLICK_PREFIX):])
payload = str(click.get("payload") or "")
chat.values = click.get("state") if isinstance(click.get("state"), dict) else {}
chat.scope = click.get("scope") if isinstance(click.get("scope"), dict) else {}
if chat.values:
# Что человек выбрал на экране — видно в журнале сразу.
LOG.info(" выбор: %s", ", ".join(f"{k}={v}" for k, v in chat.values.items()))
except ValueError:
# Молча выходить нельзя: так нажатие пропадает без следа.
LOG.warning("Нажатие не разобрано: %r", text[:120])
return
for pattern, handler in self.handlers:
if fnmatch.fnmatchcase(payload, pattern):
handler(chat, payload)
return
LOG.info("Нажатие %r без обработчика", payload)
return
if text.startswith(APP_PROTO_PREFIX):
# Вызов из программы: отвечаем тем же номером, чтобы она поняла,
# на какой именно вопрос пришёл ответ.
try:
body = json.loads(text[len(APP_PROTO_PREFIX):])
except ValueError:
LOG.warning("Вызов программы не разобран")
return
app_id = str(body.get("app") or "")
call_id = body.get("id")
name = str(body.get("m") or "")
chat.app = app_id
chat.call = call_id
raw_params = body.get("p")
try:
params = json.loads(raw_params) if isinstance(raw_params, str) else raw_params
except ValueError:
params = raw_params
LOG.info(" вызов программы: %s(%s)", name, str(params)[:80])
handler = self.methods.get(name)
if handler is None:
result: Any = {"error": f"нет метода {name}"}
else:
try:
result = handler(chat, params)
except Exception as error: # noqa: BLE001
LOG.exception("Метод %s упал", name)
result = {"error": str(error)[:200]}
if call_id is not None and result is not DEFER:
self.reply(chat.id, app_id, call_id, result)
return
if text.startswith(APP_DATA_PREFIX):
# Программа прислала данные. Конверт разбираем сами: боту важно
# содержимое, а не то, с телефона оно пришло или из окна.
try:
envelope = json.loads(text[len(APP_DATA_PREFIX):])
except ValueError:
LOG.warning("Данные программы не разобраны")
return
payload = str(envelope.get("data") or "")
LOG.info(" из программы: %s", payload[:120])
if self.app_handler:
self.app_handler(chat, payload)
elif self.text_handler:
self.text_handler(chat, payload)
return
if text.strip().lower() in ("/start", "start", "старт") or self.text_handler is None:
if self.start_handler:
self.start_handler(chat)
return
self.text_handler(chat, text)
def run(self) -> None:
answer = self.call("GET", "/api/bot/getMe")
self.me = answer.get("result") if isinstance(answer.get("result"), dict) else answer
LOG.info("Бот «%s», номер %s — слушаю", self.me.get("username"), self.me.get("public_id"))
offset, pause = 0, 1.0
while not self.stop.is_set():
try:
answer = self.call("GET", "/api/bot/getUpdates?offset=%d" % offset)
result = answer.get("result", answer)
updates = result if isinstance(result, list) else result.get("updates", [])
if updates:
LOG.info("получено обновлений: %d", len(updates))
for update in updates:
# Номер обновления бывает строкой (UUID): int() от неё
# уронил бы всю пачку разом.
raw_number = update.get("update_id") or 0
number = int(raw_number) if str(raw_number).isdigit() else 0
message = update.get("message") if isinstance(update.get("message"), dict) else update
chat_id = str(message.get("from_pk") or message.get("chat_id") or "").strip()
text = str(message.get("text") or "")
if chat_id and text:
try:
self.dispatch(chat_id, text)
except Exception: # noqa: BLE001
# Ошибка в одном обработчике не должна останавливать бота.
LOG.exception("Обработчик упал")
else:
# Без отправителя или текста: пишем, какие в нём поля, —
# по-другому такое обновление не отладить.
LOG.warning("Обновление без текста или отправителя: поля %s",
sorted(message.keys()))
if number:
offset = max(offset, number + 1)
pause = 1.0
self.stop.wait(1.5 if not updates else 0.15)
except urllib.error.HTTPError as error:
LOG.error("Сервер ответил %s", error.code)
self.stop.wait(pause)
pause = min(pause * 2, 60)
except Exception: # noqa: BLE001
LOG.exception("Сбой связи")
self.stop.wait(3.0)
# ─────────────────────────────────────────────────────────────
# Python-программы в окне
# ─────────────────────────────────────────────────────────────
#
# Программа работает на машине бота, у человека — только окно-терминал.
# Вывод не толкается в окно событием на каждый print: у бота всего 60
# сообщений в минуту на всех. Окно само спрашивает «что нового после N»
# (py.wait), бот держит вопрос до 24 секунд и отвечает одной пачкой, как
# только программа что-то напечатала или ждёт ввода. Пока окно закрыто,
# ничего не отправляется вовсе.
#: Обработчик вызова вернул это — ответ придёт позже, из сеанса.
DEFER = object()
_PROGRAM_NAME = re.compile(r"[a-z0-9][a-z0-9_-]{0,39}")
class ProgramClosed(Exception):
"""Окно закрыли, нажали «Стоп» или сеанс простоял: ждать ввода некому."""
class ProgramSpec:
"""Что запускать: функцию из `@bot.program` или файл из `bot.script`."""
def __init__(self, name: str, *, target: Callable[["ProgramIO"], Any] | None = None,
script: Path | None = None, args: Sequence[str] = (), title: str = "",
subtitle: str = "", icon: str = "🐍", fullscreen: bool = False,
idle: float = 600) -> None:
if not _PROGRAM_NAME.fullmatch(name):
raise ValueError(f"имя программы {name!r}: латиница, цифры, - и _, до 40 знаков")
if (target is None) == (script is None):
raise ValueError("программе нужна функция или файл — что-то одно")
if idle < 30:
raise ValueError("idle меньше 30 секунд: программа остановится, пока человек думает")
self.name = name
self.target = target
self.script = script
self.args = list_(args)
self.title = title or name
self.subtitle = subtitle
self.icon = icon
self.fullscreen = fullscreen
self.idle = float(idle)
class ProgramIO:
"""Ввод-вывод программы: как print и input, только в окне у человека."""
def __init__(self, session: "ProgramSession") -> None:
self._session = session
#: Собеседник: `io.chat.state` — память о нём, `io.chat.say(...)` — сообщение.
self.chat = Chat(session.bot, session.chat_id)
def print(self, *values: Any, sep: str = " ", end: str = "\n") -> None:
self._session.write(sep.join(str(v) for v in values) + end)
def write(self, text: str) -> None:
self._session.write(str(text))
def input(self, prompt: str = "") -> str:
return self._session.read_line(str(prompt))
def choose(self, prompt: str, options: Sequence[Any]) -> str:
"""Вопрос с кнопками под строкой ввода; написать ответ руками тоже можно."""
labels = [str(o)[:40] for o in options][:12]
if not labels:
raise ValueError("io.choose без вариантов")
self._session.emit("buttons", o=labels)
return self._session.read_line(str(prompt))
def clear(self) -> None:
self._session.emit("clear")
def title(self, text: str) -> None:
self._session.emit("title", t=str(text)[:80])
def sleep(self, seconds: float) -> None:
"""Пауза, которую прерывает «Стоп»."""
if self._session.stopped.wait(seconds):
raise ProgramClosed()
@property
def stopped(self) -> bool:
return self._session.stopped.is_set()
class ProgramSession:
"""Один запуск программы для одного собеседника."""
HISTORY = 256 * 1024 # сколько вывода помнить для повторного открытия окна
PART = 8_000 # символов в одном ответе: конверт моста — до 32 000
HOLD = 24.0 # сколько держать py.wait: мост ждёт ответ 30 секунд
BATCH = 0.25 # сколько копить вывод перед ответом
def __init__(self, bot: Bot, spec: ProgramSpec, chat_id: str, app: str) -> None:
self.bot, self.spec, self.chat_id, self.app = bot, spec, chat_id, app
self.id = secrets.token_hex(6)
self.cond = threading.Condition()
self.items: list[dict] = []
self.seq = 0 # номер последней записи вывода
self.size = 0
self.trimmed = 0 # последняя запись, выпавшая из истории
self.delivered = 0 # последняя запись, ушедшая в окно
self.unsent_at: float | None = None
self.state = "running"
self.code: int | None = None
self.why = ""
self.lines: "queue.Queue[str | None]" = queue.Queue()
self.stopped = threading.Event()
self.finished = threading.Event()
self.pending: tuple[Any, int, float] | None = None
self.last_reply = 0.0
self.seen = time.monotonic()
self.ended_at = 0.0
self.proc: subprocess.Popen | None = None
def start(self) -> None:
run = self._run_script if self.spec.script else self._run_function
threading.Thread(target=run, name=f"program-{self.spec.name}", daemon=True).start()
threading.Thread(target=self._replies, name=f"program-{self.spec.name}-out",
daemon=True).start()
def touch(self) -> None:
with self.cond:
self.seen = time.monotonic()
# ── вывод ────────────────────────────────────────────────
def emit(self, kind: str, **fields: Any) -> None:
with self.cond:
last = self.items[-1] if self.items else None
text = fields.get("t", "")
if (kind == "out" and last is not None and last["k"] == "out"
and last["seq"] > self.delivered and len(last["t"]) + len(text) <= self.PART // 2):
last["t"] += text # окно эту запись ещё не видело — дописываем
self.size += len(text)
else:
self.seq += 1
item = {"seq": self.seq, "k": kind, **fields}
self.items.append(item)
self.size += _weight(item)
while self.size > self.HISTORY and len(self.items) > 1:
old = self.items.pop(0)
self.size -= _weight(old)
self.trimmed = old["seq"]
if self.unsent_at is None:
self.unsent_at = time.monotonic()
self.cond.notify_all()
def _out(self, text: str) -> None:
for start in range(0, len(text), self.PART // 2):
self.emit("out", t=text[start:start + self.PART // 2])
def write(self, text: str) -> None:
if self.stopped.is_set():
raise ProgramClosed()
if text:
self._out(text)
def _set_state(self, state: str) -> None:
if self.state != state:
self.state = state
self.emit("state", s=state)
def snapshot(self, after: int) -> dict:
"""Что окно ещё не видело. С нуля — хвост истории, а не начало."""
with self.cond:
fresh = [item for item in self.items if item["seq"] > after]
picked: list[dict] = []
total = 0
# С нуля окно открывают заново: важнее конец вывода, а не начало.
for item in (reversed(fresh) if after == 0 else fresh):
if picked and total + _weight(item) > self.PART:
break
picked.append(item)
total += _weight(item)
if after == 0:
picked.reverse()
more = after != 0 and len(picked) < len(fresh)
skipped = after == 0 and len(picked) < len(fresh)
if picked:
self.delivered = max(self.delivered, picked[-1]["seq"])
self.unsent_at = None
result: dict[str, Any] = {
"session": self.id, "title": self.spec.title, "items": picked,
# Пока не всё отдано, окно должно спрашивать дальше.
"state": "running" if more and self.state == "done" else self.state,
"truncated": skipped or (0 < after < self.trimmed) or (after == 0 and self.trimmed > 0),
}
if self.state == "done" and not more:
result["code"] = self.code
return result
# ── ожидание окна ────────────────────────────────────────
def wait(self, call_id: Any, after: int) -> Any:
"""py.wait: ответ придёт, когда будет что показать (или через 24 с)."""
with self.cond:
self.seen = time.monotonic()
if self.finished.is_set() and self.pending is None:
return self.snapshot(after)
if call_id is None:
return self.snapshot(after)
# Прежний вопрос окно уже бросило: отвечать на него — лишнее сообщение.
self.pending = (call_id, after, time.monotonic() + self.HOLD)
self.cond.notify_all()
return DEFER
def _gap(self) -> float:
"""Пауза между ответами: делим 50 сообщений в минуту на живые программы."""
live = sum(1 for s in list(self.bot.sessions.values()) if not s.finished.is_set())
return max(0.3, live * 60 / 50)
def _replies(self) -> None:
while not self.bot.stop.is_set():
with self.cond:
now = time.monotonic()
if self.pending is None:
if self.finished.is_set():
return
if now - self.seen > self.spec.idle:
self.stop("idle") # окно давно не спрашивало — его закрыли
self.cond.wait(1.0)
continue
call_id, after, deadline = self.pending
ready = now >= deadline or (self.finished.is_set() and self.seq <= after)
if self.seq > after:
# Окно переспрашивает уже отданное — копить нечего.
since = self.unsent_at if self.unsent_at is not None else now - self.BATCH
at = max(since + self.BATCH, self.last_reply + self._gap())
ready = ready or now >= at
if not ready:
self.cond.wait(min(at, deadline) - now)
continue
if not ready:
self.cond.wait(min(deadline - now, 1.0))
continue
result = self.snapshot(after)
self.pending = None
self.last_reply = now
self.bot.reply(self.chat_id, self.app, call_id, result)
# ── ввод ─────────────────────────────────────────────────
def feed(self, line: str) -> None:
line = line.replace("\r", "").replace("\n", " ")[:4000]
with self.cond:
self.seen = time.monotonic()
if self.finished.is_set():
return
self.emit("echo", t=line)
self.lines.put(line)
def read_line(self, prompt: str = "") -> str:
self.write(prompt)
self._set_state("input")
try:
while True:
if self.stopped.is_set():
raise ProgramClosed()
try:
line = self.lines.get(timeout=1.0)
except queue.Empty:
continue
if line is None:
raise ProgramClosed()
return line
finally:
if not self.stopped.is_set():
self._set_state("running")
def stop(self, why: str) -> None:
if self.stopped.is_set():
return
self.why = self.why or why
self.stopped.set()
self.lines.put(None)
proc = self.proc
if proc is not None and proc.poll() is None:
proc.kill()
LOG.info(" программа %s остановлена (%s)", self.spec.name, why)
def _finish(self, code: int | None) -> None:
self.code = code
self.emit("end", code=code, why=self.why)
self._set_state("done")
self.ended_at = time.monotonic()
self.finished.set()
with self.cond:
self.cond.notify_all()
# ── запуск ───────────────────────────────────────────────
def _run_function(self) -> None:
code: int | None = 0
_console.attach(self)
try:
self.spec.target(ProgramIO(self)) # type: ignore[misc]
except ProgramClosed:
code = None
except SystemExit as exit_:
code = exit_.code if isinstance(exit_.code, int) else 0
except Exception as error: # noqa: BLE001
# Полный след — в журнал бота; человеку — одна строка без путей машины.
LOG.exception("Программа %s упала", self.spec.name)
self.emit("error", t=f"{type(error).__name__}: {error}"[:500])
code = 1
finally:
_console.detach()
self._finish(code)
def _run_script(self) -> None:
script = self.spec.script
assert script is not None
token = self.bot.token
env = {k: v for k, v in os.environ.items() if not (token and v == token)}
env.update(PYTHONUNBUFFERED="1", PYTHONIOENCODING="utf-8",
LIBYRA_PROGRAM=self.spec.name)
try:
self.proc = subprocess.Popen(
[sys.executable, "-u", str(script), *self.spec.args], cwd=str(script.parent),
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=env,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
except OSError as error:
self.emit("error", t=f"Программа не запустилась: {error}")
self._finish(None)
return
threading.Thread(target=self._feed_script, name=f"program-{self.spec.name}-in",
daemon=True).start()
decoder = codecs.getincrementaldecoder("utf-8")("replace")
stdout = self.proc.stdout
assert stdout is not None
while True:
chunk = stdout.read1(4096) if hasattr(stdout, "read1") else stdout.read(4096)
if not chunk:
break
self._out(decoder.decode(chunk))
self._out(decoder.decode(b"", final=True))
code = self.proc.wait()
self._finish(None if self.stopped.is_set() else code)
def _feed_script(self) -> None:
proc = self.proc
assert proc is not None and proc.stdin is not None
while proc.poll() is None:
try:
line = self.lines.get(timeout=1.0)
except queue.Empty:
continue
if line is None:
return
try:
proc.stdin.write((line + "\n").encode("utf-8"))
proc.stdin.flush()
except OSError:
return
def _weight(item: dict) -> int:
return 32 + len(item.get("t", "")) + sum(len(o) for o in item.get("o", ()))
def _int(value: Any) -> int:
try:
return max(0, int(value))
except (TypeError, ValueError):
return 0
class _Console:
"""print и input внутри программы идут в её окно; у остальных потоков — как были."""
def __init__(self) -> None:
self.local = threading.local()
self.lock = threading.Lock()
self.installed = False
def current(self) -> ProgramSession | None:
return getattr(self.local, "session", None)
def attach(self, session: ProgramSession) -> None:
with self.lock:
if not self.installed:
sys.stdout = _Stream(sys.stdout, self)
sys.stdin = _Stream(sys.stdin, self)
self.installed = True
self.local.session = session
def detach(self) -> None:
self.local.session = None
class _Stream:
"""Подмена sys.stdout и sys.stdin: решает по потоку, чей это вывод."""
def __init__(self, original: Any, console: _Console) -> None:
self._original = original
self._console = console
def write(self, text: str) -> int:
session = self._console.current()
if session is None:
return self._original.write(text) if self._original else len(text)
session.write(text)
return len(text)
def flush(self) -> None:
if self._console.current() is None and self._original:
self._original.flush()
def readline(self, size: int = -1) -> str:
session = self._console.current()
if session is None:
return self._original.readline(size) if self._original else ""
return session.read_line() + "\n"
def read(self, size: int = -1) -> str:
return self.readline(size)
def fileno(self) -> int:
# Без номера файла input() пишет приглашение в sys.stdout и читает
# sys.stdin.readline() — то есть идёт через нас, а не мимо в консоль.
if self._console.current() is not None or not self._original:
raise io.UnsupportedOperation("fileno")
return self._original.fileno()
def isatty(self) -> bool:
if self._console.current() is not None or not self._original:
return False
return self._original.isatty()
def __getattr__(self, name: str) -> Any:
return getattr(self._original, name)
_console = _Console()
def _program_bundle(spec: ProgramSpec) -> bytes:
"""Окно-терминал программы: один index.html, без сети и чужих файлов.
Дата в архиве постоянная: тот же код даёт тот же отпечаток, и паспорт
не спрашивается заново после каждого перезапуска бота.
"""
config = json.dumps({"name": spec.name, "title": spec.title, "icon": spec.icon},
ensure_ascii=False).replace("
Программа
"""
def _describe(text: str) -> str:
"""Кратко для журнала: что за сообщение, без полного JSON."""
if text.startswith(UI_PREFIX):
try:
return "экран «%s»" % json.loads(text[len(UI_PREFIX):]).get("id", "?")
except ValueError:
return "экран (не разобран)"
if text.startswith(CLICK_PREFIX):
try:
return "нажатие %r" % json.loads(text[len(CLICK_PREFIX):]).get("payload")
except ValueError:
return "нажатие (не разобрано)"
return "текст %r" % text[:40]
def self_check() -> None:
"""Каждый построитель даёт то, что понимает клиент; опечатки ловятся сразу."""
screen = ui.screen("check", [
ui.hero("Заголовок", gradient=["#22D3EE", "#7C5CFF"]),
ui.text("**жирный** и _курсив_"),
ui.row(ui.stat("Цена", "100", 1.5), ui.stat("Объём", "2 000", -0.4)),
ui.ticker([("BTC", "64 000", 1.2)]),
ui.chart("candles", candles=[[1, 2, 0.5, 1.5], [1.5, 2.2, 1.2, 1.3]]),
ui.chart("area", [1, 3, 2, 4]),
ui.showcase([ui.item("Товар", "990 ₽", "P:1", emoji="🎧")]),
ui.product("Товар", "990 ₽", actions=[ui.button("Купить", "BUY", animation="glow")]),
ui.buttons([ui.button("Песок", "X", dismiss="sand")]),
])
assert screen.startswith(UI_PREFIX)
spec = json.loads(screen[len(UI_PREFIX):])
assert spec["id"] == "check" and len(spec["body"]) == 9
logic = ui.screen("logic", [
ui.tabs("tab", [("about", "О товаре"), ("specs", "Характеристики")]),
ui.when("tab == 'specs'", ui.text("Вес 248 г")),
ui.choice("color", ["Чёрный", "Белый"], label="Цвет"),
ui.stepper("qty", "Количество", min=1, max=5),
ui.toggle("gift", "Упаковать в подарок"),
ui.text("Итого: {qty * price} ₽"),
ui.countdown(30, "До конца акции", on_end={"set": {"expired": True}}),
ui.button("+1", inc=ui.inc("qty", 1, max=5), local=True),
ui.lottie({"v": "5.7.4", "fr": 30, "ip": 0, "op": 30, "w": 10, "h": 10, "layers": []}),
], state={"tab": "about", "qty": 1, "price": 990, "color": "Чёрный"})
logic_spec = json.loads(logic[len(UI_PREFIX):])
assert logic_spec["state"]["qty"] == 1 and logic_spec["body"][1]["show_if"] == "tab == 'specs'"
assert logic_spec["body"][7]["local"] is True
form = ui.screen("form", [
ui.input("name", "Имя", "Как к вам обращаться"),
ui.input("phone", "Телефон", kind="phone"),
ui.each("cart", ui.text("{item.title} × {item.qty}"),
ui.button("Удалить", local=True, lists=ui.remove("cart", index="=index")),
empty=[ui.caption("Пусто")]),
ui.button("+ чехол", local=True, lists=ui.push("cart", {"id": "c", "title": "Чехол", "qty": 1},
unique="id")),
ui.button("Отправить", "SEND", enabled_if="len(trim(name)) >= 2"),
], state={"name": "", "phone": "", "cart": []}, force=["cart"])
form_spec = json.loads(form[len(UI_PREFIX):])
assert form_spec["body"][2]["children"][1]["remove"] == [{"list": "cart", "index": "=index"}]
assert form_spec["body"][3]["push"][0]["unique"] == "id" and form_spec["force"] == ["cart"]
assert form_spec["body"][4]["enabled_if"]
program = ui.screen("game", [
ui.clock("seconds", label="Сейчас"),
ui.timer(1, inc=ui.inc("score", 1), while_="running"),
ui.each("board", ui.button("{cell.v}", local=True, enabled_if="cell.v == ''",
lists=ui.update("board", {"v": "=mark"}, index="=index")),
as_="cell", columns=3, aspect=1),
ui.text("Очки: {score}"),
], state={"score": 0, "running": True, "mark": "×",
"board": [{"id": i, "v": ""} for i in range(9)]})
program_spec = json.loads(program[len(UI_PREFIX):])
assert program_spec["body"][1]["while"] == "running"
assert program_spec["body"][2]["columns"] == 3 and program_spec["body"][2]["aspect"] == 1
paint = ui.screen("paint", [
ui.canvas(100, 60, [
ui.circle("=50 + 40 * sin(frame)", "=30 + 20 * sin(frame * 1.7)", 5, "#22D3EE"),
ui.line(0, 59, 100, 59, "#7C5CFF", width=1),
ui.label(50, 8, "{len(dots)} точек", "#E6F1FF", size=5),
ui.circle("=item.x", "=item.y", 3, "#FFD166", list="dots"),
], background="#0B1B2A", animate=True,
on_tap=ui.tap(lists=ui.push("dots", {"id": "=now()", "x": "=tap_x", "y": "=tap_y"}))),
ui.button("Очистить", local=True, lists=ui.clear("dots")),
], state={"dots": []})
paint_spec = json.loads(paint[len(UI_PREFIX):])
assert paint_spec["body"][0]["animate"] is True
assert paint_spec["body"][0]["shapes"][3]["list"] == "dots"
assert paint_spec["body"][0]["on_tap"]["push"][0]["item"]["x"] == "=tap_x"
gestures = ui.screen("gestures", [
ui.swipe(
ui.card(ui.text("{item.title}")),
left=ui.tap(lists=ui.remove("cards", index="=index")),
right=ui.tap("LIKE:{item.id}"),
hold=ui.tap(set={"note": "=item.title"}),
),
ui.canvas(100, 50, [ui.line("=item.x1", "=item.y1", "=item.x2", "=item.y2",
"#22D3EE", list="strokes", width=1.5)],
on_drag=ui.tap(set={"lx": "=tap_x", "ly": "=tap_y"},
lists=ui.push("strokes", {"id": "=now()", "x1": "=lx", "y1": "=ly",
"x2": "=tap_x", "y2": "=tap_y"}))),
], state={"cards": [], "strokes": [], "lx": 0, "ly": 0, "note": ""})
gesture_spec = json.loads(gestures[len(UI_PREFIX):])
assert gesture_spec["body"][0]["left"]["remove"][0]["index"] == "=index"
assert gesture_spec["body"][0]["right"]["payload"] == "LIKE:{item.id}"
assert gesture_spec["body"][1]["on_drag"]["push"][0]["item"]["x1"] == "=lx"
photos = ui.screen("photos", [
ui.photo("00000000-0000-0000-0000-000000000000", ratio=1.4),
ui.product("Наушники", "7 990 ₽", photo="00000000-0000-0000-0000-000000000000"),
ui.showcase([ui.item("Чехол", "990 ₽", "P:2", photo="00000000-0000-0000-0000-000000000000")]),
])
photo_spec = json.loads(photos[len(UI_PREFIX):])
assert photo_spec["body"][0]["media"].startswith("00000000")
assert photo_spec["body"][1]["photo"] and photo_spec["body"][2]["items"][0]["photo"]
sounds = ui.screen("sounds", [
ui.sound("alarm", play_if="left == 0"),
ui.button("Купить", "BUY", sound="coin"),
ui.button("Удалить", local=True, set={"x": 1}, sound="error"),
], state={"left": 3})
sound_spec = json.loads(sounds[len(UI_PREFIX):])
assert sound_spec["body"][0]["play"] == "alarm" and sound_spec["body"][0]["play_if"] == "left == 0"
assert sound_spec["body"][1]["sound"] == "coin"
assert ui.sound("media:abc")["play"] == "media:abc"
bot_probe = Bot(token="проверка")
@bot_probe.method("ledger.total")
def _total(chat, params): # noqa: ANN001, ANN202
return {"total": sum(params.get("rows", []))}
assert "ledger.total" in bot_probe.methods
assert bot_probe.methods["ledger.total"](None, {"rows": [1, 2]}) == {"total": 3}
program = ui.screen("app", [
ui.app("00000000-0000-0000-0000-000000000000", "Счётная", subtitle="Своя бухгалтерия"),
])
app_spec = json.loads(program[len(UI_PREFIX):])
assert app_spec["body"][0]["media"].startswith("00000000")
assert app_spec["body"][0]["label"] == "Открыть"
for bad in (lambda: ui.button("x"),
lambda: ui.sound("бум"),
lambda: ui.button("x", "Y", sound="взрыв"),
lambda: ui.photo("x", fit="stretch"),
lambda: ui.swipe(ui.text("x")),
lambda: ui.swipe(left=ui.tap("A")),
lambda: ui.canvas(10, 10, []),
lambda: ui.circle(1, 1, 1, glow=True),
lambda: ui.timer(0.05, set={"a": 1}),
lambda: ui.timer(1),
lambda: ui.clock("часики"),
lambda: ui.input("x", kind="password"),
lambda: ui.remove("cart"),
lambda: ui.lottie({"v": "5.7.4"}),
lambda: ui.stepper("q", min=5, max=1),
lambda: ui.button("x", "y", animation="spin"),
lambda: ui.text("x", "giant"),
lambda: ui.chart("pie", [1]),
lambda: ui.screen("x", [])):
try:
bad()
except ValueError:
continue
raise AssertionError("опечатка не поймана")
_check_programs()
print("libyra_ui: проверка пройдена")
def _check_programs() -> None:
"""Python-программы: функция и скрипт, ввод, кнопки, остановка — без сети."""
import tempfile
probe = Bot(token="проверка-программ")
replies: list[tuple[Any, dict]] = []
probe.reply = lambda chat_id, app, call_id, result: replies.append((call_id, result)) # type: ignore[method-assign]
def until(check: Callable[[], Any], what: str) -> None:
deadline = time.monotonic() + 10
while not check():
if time.monotonic() > deadline:
raise AssertionError(f"программа не дошла до: {what}")
time.sleep(0.02)
def answer(call_id: Any) -> dict:
until(lambda: any(c == call_id for c, _ in replies), f"ответ на вызов {call_id}")
return next(r for c, r in replies if c == call_id)
def text(result: dict) -> str:
return "".join(i.get("t", "") for i in result["items"] if i["k"] in ("out", "echo"))
@probe.program("hello", title="Привет")
def _hello(io_: ProgramIO) -> None:
name = input("Имя? ") # обычный input — через окно
print("Привет,", name) # обычный print — в окно
io_.print("выбор:", io_.choose("Ещё?", ["да", "нет"]))
with tempfile.TemporaryDirectory() as folder:
script = Path(folder) / "echo.py"
script.write_text("import os\nprint('токен:', 'проверка-программ' in os.environ.values())\n"
"print('эхо:', input('> '))\nraise SystemExit(3)\n", encoding="utf-8")
probe.script("echo", script, title="Эхо")
assert {"py.start", "py.wait", "py.input", "py.stop"} <= set(probe.methods)
bundle = _program_bundle(probe.programs["hello"])
assert bundle == _program_bundle(probe.programs["hello"]), "отпечаток окна плавает"
html = zipfile.ZipFile(io.BytesIO(bundle)).read("index.html").decode("utf-8")
assert '"name": "hello"' in html and "__PROGRAM__" not in html and "http" not in html
chat = Chat(probe, "pk-check")
chat.app = "app-check"
session_id = probe._py_start(chat, {"program": "hello"})["session"]
session = probe.sessions[("pk-check", "hello")]
until(lambda: session.state == "input", "ввод имени")
chat.call = 1
assert probe._py_wait(chat, {"program": "hello", "session": session_id, "after": 0}) is DEFER
first = answer(1)
assert "Имя? " in text(first) and first["state"] == "input"
chat.call = 2
probe._py_input(chat, {"program": "hello", "session": session_id,
"after": first["items"][-1]["seq"], "line": "Оля"})
second = answer(2)
until(lambda: session.state == "input", "выбор кнопкой")
if not any(i["k"] == "buttons" for i in second["items"]):
chat.call = 3
probe._py_wait(chat, {"program": "hello", "session": session_id,
"after": second["items"][-1]["seq"]})
second = answer(3)
assert "Привет, Оля" in text(second)
assert any(i["k"] == "buttons" and i["o"] == ["да", "нет"] for i in second["items"])
chat.call = 4
probe._py_input(chat, {"program": "hello", "session": session_id,
"after": session.seq, "line": "да"})
until(session.finished.is_set, "конец программы")
assert "выбор: да" in text(answer(4))
chat.call = None
final = probe._py_wait(chat, {"program": "hello", "session": session_id, "after": 0})
assert "выбор: да" in text(final) and final["state"] == "done" and final["code"] == 0
assert probe._py_wait(chat, {"program": "hello", "session": "чужой"}) == {"gone": True}
echo_id = probe._py_start(chat, {"program": "echo"})["session"]
echo = probe.sessions[("pk-check", "echo")]
until(lambda: "> " in text(echo.snapshot(0)), "приглашение скрипта")
chat.call = 5
probe._py_input(chat, {"program": "echo", "session": echo_id, "after": 0, "line": "тест"})
until(echo.finished.is_set, "конец скрипта")
done = echo.snapshot(0)
assert "токен: False" in text(done), "токен бота утёк в окружение скрипта"
assert "эхо: тест" in text(done) and done["code"] == 3
chat.call = None
probe._py_start(chat, {"program": "hello", "restart": True})
stopping = probe.sessions[("pk-check", "hello")]
until(lambda: stopping.state == "input", "повторный запуск")
probe._py_stop(chat, {"program": "hello", "session": stopping.id})
until(stopping.finished.is_set, "остановка")
assert any(i["k"] == "end" and i["why"] == "user" for i in stopping.items)
probe.stop.set()
for bad in (lambda: ProgramSpec("Плохое имя", target=print),
lambda: ProgramSpec("x", target=print, idle=5),
lambda: ProgramSpec("x")):
try:
bad()
except ValueError:
continue
raise AssertionError("ошибка в описании программы не поймана")
if __name__ == "__main__":
self_check()