admin
87e421ae73
Статус долгой операции больше не оставляет в чате отдельное сообщение: в приватном чате он стримится живым драфтом (sendMessageDraft), пока идет классификация, генерация, чтение скриншотов или распознавание речи. - activity_phrases.py: пулы фраз по типу операции (generation, training, images, voice, ideas), случайный интервал 3-8 сек (ACTIVITY_MIN_SECONDS / ACTIVITY_MAX_SECONDS), фраза не повторяет предыдущую - OperationStatus: драфт со стабильным draft_id, фоновая ротация фраз, finish() отправляет финальный статус реальным сообщением, retarget() меняет пул на лету (generate_ideas), предохранитель ACTIVITY_MAX_ROTATIONS - в группах и топиках драфты недоступны: прежний answer + edit_text, ротация фраз отключена; при ошибке драфта автоматический откат - Dockerfile копирует activity_phrases.py и support.py, иначе контейнер падает на импорте Проверено: 172 теста OK,其中包括 14 новых в tests/test_activity_phrases.py. Тесты нашли и закрыли баг: ротация запускалась и в группах, где каждый тик был бы реальным редактированием сообщения. VERSION: 0.1.0+48 -> 0.1.0+49
122 lines
5.6 KiB
Python
122 lines
5.6 KiB
Python
"""Пулы фраз, которыми бот обозначает ход долгой операции.
|
|
|
|
Фразы показываются в живом драфте (``sendMessageDraft``) в приватном чате и
|
|
сменяют друг друга случайным образом через случайные промежутки времени, пока
|
|
операция не завершилась. Ни одна фраза не обещает конкретного результата — это
|
|
только обозначение процесса.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import random
|
|
|
|
GENERATION_PHRASES: tuple[str, ...] = (
|
|
"⏳ Разбираю запрос и подбираю формулировки…",
|
|
"⏳ Собираю фактуру из того, что ты прислал…",
|
|
"⏳ Придумываю три разных захода к теме…",
|
|
"⏳ Проверяю, что не придумываю лишних фактов…",
|
|
"⏳ Подбираю более точное начало поста…",
|
|
"⏳ Слежу за длиной и читаемостью каждого варианта…",
|
|
"⏳ Убираю канцелярит и лишние вводные…",
|
|
"⏳ Собираю три варианта и проверяю орфографию…",
|
|
)
|
|
|
|
TRAINING_PHRASES: tuple[str, ...] = (
|
|
"⏳ Читаю присланные материалы…",
|
|
"⏳ Выделяю особенности твоей подачи…",
|
|
"⏳ Сопоставляю новые примеры с прежним профилем…",
|
|
"⏳ Уточняю портрет аудитории…",
|
|
"⏳ Отмечаю типичные темы и повороты…",
|
|
"⏳ Собираю обновлённый профиль аккаунта…",
|
|
"⏳ Проверяю, что ничего не потерялось…",
|
|
"⏳ Упаковываю вывод в компактный профиль…",
|
|
)
|
|
|
|
IMAGES_PHRASES: tuple[str, ...] = (
|
|
"⏳ Скачиваю изображения…",
|
|
"⏳ Распознаю тексты на скриншотах…",
|
|
"⏳ Отделяю заголовки от основного текста…",
|
|
"⏳ Сверяю извлечённые посты между собой…",
|
|
"⏳ Ищу повторяющиеся приёмы в текстах…",
|
|
"⏳ Проверяю, что тексты прочитаны полностью…",
|
|
"⏳ Готовлю варианты поста по увиденному…",
|
|
)
|
|
|
|
VOICE_PHRASES: tuple[str, ...] = (
|
|
"⏳ Скачиваю аудиосообщение…",
|
|
"⏳ Перевожу речь в текст…",
|
|
"⏳ Расставляю знаки препинания в расшифровке…",
|
|
"⏳ Уточняю границы фраз…",
|
|
"⏳ Проверяю расшифровку на слух…",
|
|
"⏳ Разбираю, что именно ты просишь…",
|
|
)
|
|
|
|
IDEAS_PHRASES: tuple[str, ...] = (
|
|
"⏳ Ищу свежий угол для темы…",
|
|
"⏳ Проверяю, что идея не повторяет прошлые…",
|
|
"⏳ Подбираю формулировку заголовка…",
|
|
"⏳ Оцениваю пользу идеи для твоей аудитории…",
|
|
"⏳ Формулирую, зачем эта тема аккаунту…",
|
|
"⏳ Отбираю самый сильный вариант…",
|
|
)
|
|
|
|
DEFAULT_PHRASES: tuple[str, ...] = (
|
|
"⏳ Работаю над запросом…",
|
|
"⏳ Привожу результат к нужному виду…",
|
|
"⏳ Почти готова, проверяю детали…",
|
|
)
|
|
|
|
ACTIVITY_PHRASES: dict[str, tuple[str, ...]] = {
|
|
"generation": GENERATION_PHRASES,
|
|
"training": TRAINING_PHRASES,
|
|
"images": IMAGES_PHRASES,
|
|
"voice": VOICE_PHRASES,
|
|
"ideas": IDEAS_PHRASES,
|
|
}
|
|
|
|
|
|
def _positive_float(name: str, default: float) -> float:
|
|
raw = os.getenv(name, "").strip()
|
|
if not raw:
|
|
return default
|
|
try:
|
|
value = float(raw)
|
|
except ValueError:
|
|
return default
|
|
return value if value > 0 else default
|
|
|
|
|
|
def _positive_int(name: str, default: int) -> int:
|
|
raw = os.getenv(name, "").strip()
|
|
if not raw:
|
|
return default
|
|
try:
|
|
value = int(raw)
|
|
except ValueError:
|
|
return default
|
|
return value if value > 0 else default
|
|
|
|
|
|
ACTIVITY_MIN_SECONDS = _positive_float("ACTIVITY_MIN_SECONDS", 3.0)
|
|
ACTIVITY_MAX_SECONDS = _positive_float("ACTIVITY_MAX_SECONDS", 8.0)
|
|
if ACTIVITY_MIN_SECONDS > ACTIVITY_MAX_SECONDS:
|
|
ACTIVITY_MIN_SECONDS, ACTIVITY_MAX_SECONDS = ACTIVITY_MAX_SECONDS, ACTIVITY_MIN_SECONDS
|
|
|
|
# Предохранитель: если операция зависла и finish() не вызвался, ротация
|
|
# всё равно прекратится и не будет бесконечно долбить драфт.
|
|
ACTIVITY_MAX_ROTATIONS = _positive_int("ACTIVITY_MAX_ROTATIONS", 12)
|
|
|
|
|
|
def random_interval() -> float:
|
|
"""Случайная пауза между сменами фраз, в секундах."""
|
|
return random.uniform(ACTIVITY_MIN_SECONDS, ACTIVITY_MAX_SECONDS)
|
|
|
|
|
|
def next_phrase(kind: str | None = None, exclude: str | None = None) -> str:
|
|
"""Случайная фраза из пула операции, не повторяющая только что показанную."""
|
|
pool = ACTIVITY_PHRASES.get(kind or "", DEFAULT_PHRASES)
|
|
if exclude is None or len(pool) < 2:
|
|
return random.choice(pool)
|
|
variants = tuple(phrase for phrase in pool if phrase != exclude)
|
|
return random.choice(variants)
|