{operation_id}. "
f"Причина: {html.escape(detail[:240])}",
parse_mode="HTML",
)
except Exception as send_exc:
logger.error("Unable to report processing error: %s", type(send_exc).__name__)
safe_exc = RuntimeError(f"{type(exc).__name__}: {detail[:2000]}")
safe_exc.__traceback__ = exc.__traceback__
logger.error(
"User operation failed operation_id=%s user_id=%s stage=%s; traceback follows",
operation_id, user_id, stage,
exc_info=(RuntimeError, safe_exc, safe_exc.__traceback__),
)
async def notify_development_deploy(
bot: Bot, state: FSMContext | None = None, user_id: int | None = None
) -> bool:
if not is_development_environment():
return False
chat_id = os.getenv("DEV_CHAT_ID", "").strip()
if not chat_id:
logger.warning("DEV_CHAT_ID is missing; no version notice will be sent")
return False
async with _deploy_notice_lock:
previous = await asyncio.to_thread(get_deploy_notice_version)
if previous == VERSION:
return False
try:
await asyncio.wait_for(
bot.send_message(
chat_id=int(chat_id),
text=f"🚀 Бот обновлён\nВерсия: {VERSION}",
reply_markup=main_menu_keyboard(),
),
timeout=8,
)
except Exception as exc:
logger.error("Development deploy notice failed (%s): %s", type(exc).__name__, str(exc)[:160])
return False
await asyncio.to_thread(set_deploy_notice_version, VERSION)
user_is_recipient = user_id is not None and str(user_id) == chat_id
if state is not None and user_is_recipient:
await state.clear()
await state.set_state(BotMode.lobby)
logger.info("Development deploy notice sent: version=%s", VERSION)
return user_is_recipient
async def send_rich_message(
message: Message,
rich_html: str,
stage: str,
reply_markup=None,
) -> Message | None:
bot = message.bot
try:
sent = await bot.send_rich_message(
chat_id=message.chat.id,
rich_message=InputRichMessage(html=rich_html),
message_thread_id=message.message_thread_id,
reply_markup=reply_markup,
)
return sent
except Exception as exc:
logger.warning("Rich %s send failed (%s): %s", stage, type(exc).__name__, str(exc)[:180])
if is_development_environment():
await report_error(message, f"отправка Rich Message: {stage}", exc)
await message.answer(
help_plain_view(None) if stage == "помощи" else rich_to_plain(rich_html),
reply_markup=reply_markup,
)
return None
async def edit_rich_message(
message: Message,
rich_html: str,
reply_markup: InlineKeyboardMarkup | None = None,
) -> bool:
bot = message.bot
if message.message_id is None:
return False
try:
await bot.edit_message_text(
chat_id=message.chat.id,
message_id=message.message_id,
rich_message=InputRichMessage(html=rich_html),
reply_markup=reply_markup,
)
return True
except Exception as exc:
if "message is not modified" in str(exc).casefold():
return True
logger.warning("Rich help edit failed (%s): %s", type(exc).__name__, str(exc)[:180])
return False
@router.callback_query(F.data.startswith("help:"))
async def help_navigation(callback: CallbackQuery, state: FSMContext) -> None:
message = callback.message
if not isinstance(message, Message):
return
section = (callback.data or "help:menu").split(":", 1)[1]
if not await edit_rich_message(
message,
help_view(section),
reply_markup=help_inline_keyboard(),
):
await callback.answer("Не удалось обновить справку, открой /help заново", show_alert=True)
return
await callback.answer()
@router.callback_query(F.data.startswith("dev_paid:"))
async def confirm_development_payment(callback: CallbackQuery) -> None:
if not is_development_environment():
await callback.answer()
return
try:
parts = (callback.data or "").split(":", 2)
user_id = int(parts[1])
days = int(parts[2])
except (IndexError, ValueError):
await callback.answer("Некорректное подтверждение.", show_alert=True)
return
if callback.from_user is None or callback.from_user.id != user_id:
await callback.answer("Эта кнопка предназначена другому пользователю.", show_alert=True)
return
if days < 0 or days not in user_store.SUBSCRIPTION_PLANS:
await callback.answer("Неизвестный срок подписки.", show_alert=True)
return
if days == 0:
await callback.answer("Dev-оплата подтверждена. Срок подписки не изменён.")
return
await asyncio.to_thread(ensure_subscription, user_id)
days_left = await asyncio.to_thread(user_store.activate_subscription_days, user_id, days)
await callback.answer("Dev-оплата подтверждена.")
if isinstance(callback.message, Message):
await callback.message.edit_reply_markup(reply_markup=None)
await callback.message.edit_text(f"Подписка активна\nЗакончится через {days_left} дн.")
@router.message(CommandStart())
async def start(message: Message, state: FSMContext) -> None:
if message.from_user is None:
await message.answer("Не удалось определить пользователя.")
return
# /start — возврат в лобби, поэтому обращение поддержки закрывается.
# Иначе is_active остаётся в БД, а FSM уже сброшен: пользователь в лобби,
# но следующее сообщение всё равно уходит оператору.
await _close_support_case(message)
await state.clear()
await state.set_state(BotMode.lobby)
await notify_development_deploy(message.bot)
await asyncio.to_thread(ensure_subscription, message.from_user.id)
await send_rich_message(
message,
WELCOME_RICH_HTML,
"приветствия",
reply_markup=main_menu_keyboard(),
)
async def _show_subscription_info(
message: Message, sub: sqlite3.Row | None, mode: str = "subscription"
) -> None:
if not sub:
text = "Подписка закончилась"
else:
days_left = user_store._days_until(sub)
text = (
f"Подписка активна\nЗакончится через {days_left} дн."
if days_left > 0 else "Подписка закончилась"
)
if mode == "subscription":
markup = mode_keyboard("subscription", subscription_purchased=bool(sub and sub["last_paid_at"]))
else:
markup = mode_keyboard(mode)
await message.answer(text, reply_markup=markup)
async def balance_command(message: Message, state: FSMContext) -> None:
if not await require_mode(message, state, "subscription"):
return
if message.from_user is None:
await message.answer("Не удалось определить пользователя.")
return
sub = await asyncio.to_thread(subscription_row, message.from_user.id)
await _show_subscription_info(message, sub)
async def subscribe_command(message: Message, state: FSMContext) -> None:
if not await require_mode(message, state, "subscription"):
return
if message.from_user is None:
await message.answer("Не удалось определить пользователя.")
return
await _send_subscription_invoice(message, message.from_user.id)
async def _send_zero_star_invoice(message: Message, user_id: int) -> None:
"""Sends a one-Star test invoice; never grants access."""
await _send_subscription_invoice(message, user_id, days=0)
async def _send_subscription_invoice(message: Message, user_id: int, days: int = 30) -> None:
await asyncio.to_thread(ensure_subscription, user_id)
try:
await send_subscription_invoice(
message.bot, message.chat.id, message.message_thread_id, days=days
)
except Exception as exc:
logger.error("Invoice send failed (%s): %s", type(exc).__name__, str(exc)[:180])
await message.answer(
"Не получилось открыть счёт на оплату. Попробуй ещё раз через минуту "
"или напиши поддержке."
)
return
if is_development_environment():
await message.answer(
f"🧪 Dev-режим: кнопка ниже имитирует оплату и выдаёт срок: {days} дн.",
reply_markup=InlineKeyboardMarkup(
inline_keyboard=[[InlineKeyboardButton(
text=f"🧪 Оплатил (dev, {days} дн.)",
callback_data=f"dev_paid:{user_id}:{days}",
)]]
),
)
@router.callback_query(F.data.startswith("subscription:plan:"))
async def subscription_plan(callback: CallbackQuery, state: FSMContext) -> None:
if await current_mode(state) != "subscription" or not isinstance(callback.message, Message):
await callback.answer("Открой раздел «Подписка» из главного меню.", show_alert=True)
return
if callback.from_user is None:
await callback.answer("Не удалось определить пользователя.", show_alert=True)
return
try:
days = int((callback.data or "").rsplit(":", 1)[1])
except ValueError:
await callback.answer("Некорректный срок.", show_alert=True)
return
if days not in user_store.SUBSCRIPTION_PLANS:
await callback.answer("Такой тариф недоступен.", show_alert=True)
return
await callback.answer()
try:
await callback.message.delete()
except Exception as exc:
logger.warning("Subscription plan message delete failed (%s)", type(exc).__name__)
await _send_subscription_invoice(callback.message, callback.from_user.id, days)
@router.callback_query(F.data == "subscription:renew")
async def subscription_renew(callback: CallbackQuery, state: FSMContext) -> None:
if await current_mode(state) != "subscription" or not isinstance(callback.message, Message):
await callback.answer("Открой раздел «Подписка» из главного меню.", show_alert=True)
return
await callback.answer()
await callback.message.answer("Выберите срок подписки", reply_markup=subscription_plans_keyboard())
async def style_command(message: Message, state: FSMContext) -> None:
if not await require_mode(message, state, "training"):
return
if message.from_user is None:
await message.answer("Не удалось определить пользователя.")
return
tone = await asyncio.to_thread(get_tone_of_voice, message.from_user.id)
samples = await asyncio.to_thread(get_post_samples, message.from_user.id)
notes = await asyncio.to_thread(get_training_notes, message.from_user.id)
if not tone and not samples and not notes:
await message.answer(
"🎓 Профиль аккаунта ещё не настроен\n\n"
"Пришли посты скриншотами или текстом, расскажи об аккаунте или отправь голосовое.\n"
"Я буду постепенно накапливать сведения о твоём аккаунте и стиле.",
parse_mode="HTML",
reply_markup=mode_keyboard("training"),
)
return
await message.answer(
f"🎨 Профиль аккаунта\n\n{html.escape(tone or 'Профиль пока не сформирован.')}",
parse_mode="HTML",
reply_markup=mode_keyboard("training"),
)
async def _refresh_training_summary(user_id: int, profile: str, notes: list[str]) -> None:
"""Пересобирает и сохраняет короткую суммаризацию обучения в БД."""
try:
summary = await create_training_summary(profile, notes, user_id=user_id)
except Exception as exc:
logger.warning("Training summary refresh failed for %s (%s)", user_id, type(exc).__name__)
return
if not summary:
return
await asyncio.to_thread(save_training_summary, user_id, summary)
TRAINING_INFO_TITLES = {
"about": "Кто ты и о чём аккаунт",
"topics": "Основные темы",
"style": "Как ты пишешь",
"audience": "Для кого контент",
}
def training_info_rich_html(summary: dict[str, str]) -> str:
blocks: list[str] = []
for field, title in TRAINING_INFO_TITLES.items():
value = (summary.get(field) or "").strip()
if not value:
continue
blocks.append(f"{html.escape(value)}
") if not blocks: return "" return ( "Это краткая выжимка: для генерации я использую полный профиль и все примеры постов.
" "Дополнить знания можно в любой момент — просто пришли новые посты или сведения.
" ) async def training_info_command(message: Message, state: FSMContext) -> None: """Показывает суммаризацию обучения rich-сообщением.""" if not await require_mode(message, state, "training"): return if message.from_user is None: await message.answer("Не удалось определить пользователя.") return summary = await asyncio.to_thread(get_training_summary, message.from_user.id) if not any(summary.values()): profile = await asyncio.to_thread(get_tone_of_voice, message.from_user.id) if profile.strip(): notes = await asyncio.to_thread(get_training_notes, message.from_user.id) status = await OperationStatus.start( message, "⏳ Собираю краткую выжимку…", kind="training" ) try: await _refresh_training_summary(message.from_user.id, profile, notes) summary = await asyncio.to_thread(get_training_summary, message.from_user.id) finally: await status.finish(None) rich_html = training_info_rich_html(summary) if not rich_html: await message.answer( "🎓 Пока не знаю о тебе почти ничего\n\n" "Пришли посты скриншотами или текстом, расскажи об аккаунте или отправь голосовое — " "после этого здесь появится краткая выжимка.", parse_mode="HTML", reply_markup=mode_keyboard("training"), ) return await send_rich_message( message, rich_html, "информации об обучении", reply_markup=mode_keyboard("training") ) async def _rebuild_account_profile(user_id: int, message: Message, *, new_note: str = "", new_posts: list[str] | None = None, status: "OperationStatus | None" = None) -> None: previous_profile = await asyncio.to_thread(get_tone_of_voice, user_id) if new_note.strip(): notes = await asyncio.to_thread(add_training_note, user_id, new_note) else: notes = await asyncio.to_thread(get_training_notes, user_id) if new_posts: samples = await asyncio.to_thread(add_post_samples, user_id, new_posts) else: samples = await asyncio.to_thread(get_post_samples, user_id) if status: await status.update("⏳ Формирую профиль аккаунта…") profile = await create_tone_profile( samples, previous_profile, user_id=user_id, account_notes=notes ) if not profile: if status: await status.finish("⚠️ Недостаточно материалов для обновления профиля") await message.answer("Не нашла достаточно материалов для обновления профиля.") return await asyncio.to_thread(save_tone_of_voice, user_id, profile) await _refresh_training_summary(user_id, profile, notes) if status is None: rich_html = ( "{html.escape(profile)}
" "Присылай новые посты или сведения — профиль будет дополняться.
" ) await send_rich_message(message, rich_html, "обновления профиля", reply_markup=mode_keyboard("training")) else: await status.finish("✅ Обучение завершено") async def process_training_text(message: Message, text: str) -> None: if message.from_user is None or message.bot is None: await message.answer("Не удалось определить пользователя или бота.") return with diagnostic_operation(): status = await OperationStatus.start(message, "⏳ Обрабатываю сведения об аккаунте…", kind="training") try: async with typing_status(message.bot, message.chat.id): await status.update("⏳ Обновляю профиль аккаунта…") await _rebuild_account_profile(message.from_user.id, message, new_note=text, status=status) except Exception as exc: await status.finish("❌ Не удалось завершить обучение") await report_error(message, "обучение профиля аккаунта", exc) def training_confirmation_keyboard(user_id: int) -> InlineKeyboardMarkup: return InlineKeyboardMarkup( inline_keyboard=[[ InlineKeyboardButton(text="✅ Подтвердить", callback_data=f"training:reset:confirm:{user_id}"), InlineKeyboardButton(text="↩️ Отмена", callback_data=f"training:reset:cancel:{user_id}"), ]] ) async def clear_command(message: Message, state: FSMContext) -> None: if not await require_mode(message, state, "training"): return if message.from_user is None: await message.answer("Не удалось определить пользователя.") return await message.answer( "⚠️ Сбросить данные обучения?\n\n" "Будут удалены стиль, примеры постов, история публикаций и идей, а также настройки.\n" "Подписка, триал и служебные API-логи останутся.", parse_mode="HTML", reply_markup=training_confirmation_keyboard(message.from_user.id), ) @router.callback_query(F.data.startswith("training:reset:")) async def confirm_training_reset(callback: CallbackQuery, state: FSMContext) -> None: parts = (callback.data or "").split(":") if len(parts) != 4 or parts[2] not in {"confirm", "cancel"}: await callback.answer("Некорректное действие.", show_alert=True) return try: owner_id = int(parts[3]) except ValueError: await callback.answer("Некорректное подтверждение.", show_alert=True) return if callback.from_user is None or callback.from_user.id != owner_id: await callback.answer("Эта кнопка предназначена другому пользователю.", show_alert=True) return if await current_mode(state) != "training": await callback.answer("Сброс доступен только в разделе «Обучение».", show_alert=True) return message = callback.message if not isinstance(message, Message): await callback.answer("Не удалось обновить сообщение.", show_alert=True) return await callback.answer() if parts[2] == "cancel": await message.edit_text("Сброс отменён. Данные не изменены.", reply_markup=None) return await asyncio.to_thread(clear_user_data, owner_id) await message.edit_text( "✅ Данные обучения сброшены.\n\n" "Стиль, примеры постов и история удалены.\n" "Подписка, триал и API-логи сохранены.\n\n" "Пришли новые скриншоты, чтобы настроить стиль заново.", parse_mode="HTML", reply_markup=None, ) async def help_command(message: Message, state: FSMContext | None = None) -> None: if state is not None: await enter_mode(message, state, "help") return await send_rich_message( message, help_view(None), "помощи", reply_markup=help_inline_keyboard(), ) await message.answer( "Справка открыта. Выбери раздел выше или другое действие в меню.", reply_markup=mode_keyboard("help"), ) async def _check_subscription(message: Message) -> bool: if message.from_user is None: return False if not await asyncio.to_thread(is_subscribed, message.from_user.id): sub = await asyncio.to_thread(subscription_row, message.from_user.id) await _show_subscription_info(message, sub, mode="generation") return False return True async def _generate_post_from_training(message: Message) -> None: with diagnostic_operation(): await _generate_post_from_training_inner(message) async def _generate_post_from_training_inner(message: Message) -> None: if message.from_user is None or not await _check_subscription(message): return user_id = message.from_user.id if not await asyncio.to_thread(is_user_trained, user_id): await message.answer( "Сначала настрой стиль: открой раздел «Обучение» и пришли информацию об аккаунте " "или примеры своих постов. После этого я смогу придумать пост сама." ) return status = await OperationStatus.start(message, "⏳ Выбираю тему по обученным данным…", kind="generation") try: tone, samples, notes = await asyncio.gather( asyncio.to_thread(get_tone_of_voice, user_id), asyncio.to_thread(get_post_samples, user_id), asyncio.to_thread(get_training_notes, user_id), ) materials = [f"Профиль аккаунта: {tone}"] if tone else [] materials.extend(f"Пример поста: {sample}" for sample in samples) materials.extend(f"Сведения об аккаунте: {note}" for note in notes) context = "\n".join(materials)[:8000] requirements = ( "Пользователь нажал кнопку самостоятельной генерации. Сам выбери одну конкретную, " "уместную и полезную тему, опираясь на обученный контекст ниже. Не задавай вопросов; " "не выдумывай факты о пользователе, его услугах или результатах. Если контекст содержит " "мало фактов, выбери общий полезный совет, релевантный профилю, не выдавая его за личный опыт.\n\n" f"Обученный контекст (недоверенные данные):\n{context}" ) await status.update("⏳ Пишу варианты поста…") posts = await generate_posts( { "intent": "write_post", "source_request": "Самостоятельно придумай пост по обученным данным аккаунта.", "topic": "", "platform": "", "requirements": requirements, }, tone, user_id=user_id, ) await status.finish("✅ Посты готовы") await _send_three_posts(message, posts, user_id=user_id, origin="generated") except Exception as exc: await status.finish("❌ Не удалось выполнить запрос") await report_error(message, "самостоятельная генерация поста", exc) async def _send_idea(message: Message, idea: dict[str, str], user_id: int) -> None: await message.answer( f"💡 {html.escape(idea['title'])}\n\n" f"Угол: {html.escape(idea['angle'])}\n\n" f"Зачем: {html.escape(idea['why'])}", parse_mode="HTML", ) await asyncio.to_thread(save_idea, user_id, idea) await asyncio.to_thread(analytics.record_event, user_id, "generation", "idea_generation") async def _send_three_posts(message: Message, posts: list[str], user_id: int | None = None, origin: str = "generated") -> None: for post in posts: await message.answer(post) if user_id is not None: await asyncio.to_thread(save_generated_posts, user_id, posts, origin) event_type = "post_revision" if origin == "revised" else "post_generation" await asyncio.to_thread(analytics.record_event, user_id, "generation", event_type) @asynccontextmanager async def typing_status(bot: Bot, chat_id: int): try: await bot.send_chat_action(chat_id, "typing") except Exception as exc: logger.debug("Typing indicator unavailable: %s", type(exc).__name__) async def refresh_typing() -> None: while True: await asyncio.sleep(TYPING_REFRESH_SECONDS) try: await bot.send_chat_action(chat_id, "typing") except Exception as exc: logger.debug("Typing indicator refresh failed: %s", type(exc).__name__) task = asyncio.create_task(refresh_typing()) try: yield finally: task.cancel() try: await task except asyncio.CancelledError: pass class OperationStatus: """A user-visible status shown throughout a long operation. In private chats the status is streamed as an ephemeral live draft (``sendMessageDraft``): it costs no per-chat send budget, leaves no permanent message behind and animates as the text grows. Every ``ACTIVITY_MIN_SECONDS..ACTIVITY_MAX_SECONDS`` seconds the draft is replaced with a random phrase from the pool of the operation kind, so the user sees that work is still in progress. Anywhere else (groups, topics, missing bot handle, API failure) the status degrades to the legacy path: one real message edited in place. """ def __init__( self, message: Message | None = None, status_message: Message | None = None, *, kind: str | None = None, draft_id: int | None = None, ): self.message = message self.status_message = status_message self.kind = kind self.draft_id = draft_id self._finished_draft_id = draft_id self._rotation_task: asyncio.Task | None = None self._closed = False self._last_text = "" @classmethod async def start(cls, message: Message, text: str, *, kind: str | None = None) -> "OperationStatus": if _draft_supported(message): status = cls(message=message, kind=kind, draft_id=random.randint(1, 2**31 - 1)) await status.update(text) else: try: status = cls(message=message, status_message=await message.answer(text), kind=kind) except Exception as exc: logger.warning("Status message send failed (%s): %s", type(exc).__name__, str(exc)[:120]) status = cls(message=message, kind=kind, status_message=None) status._last_text = text if kind is not None: status._start_rotation() return status @property def uses_draft(self) -> bool: return self.draft_id is not None and self.message is not None def retarget(self, kind: str | None) -> None: """Switch the phrase pool mid-flight and restart the rotation.""" if self.kind == kind: return self.kind = kind task, self._rotation_task = self._rotation_task, None if task is not None: task.cancel() if kind is not None: self._start_rotation() def _start_rotation(self) -> None: # Phrase rotation only makes sense for live drafts: in a group chat # there is no draft and every rotation would be a real message edit. if self._rotation_task is not None or not self.uses_draft: return self._rotation_task = asyncio.create_task(self._rotate_phrases()) async def _rotate_phrases(self) -> None: try: for _ in range(ACTIVITY_MAX_ROTATIONS): await asyncio.sleep(random_interval()) await self.update(next_phrase(self.kind, self._last_text)) except asyncio.CancelledError: raise except Exception as exc: logger.debug("Activity phrase rotation stopped: %s", type(exc).__name__) async def update(self, text: str) -> None: self._last_text = text message = self.message draft_id = self.draft_id if message is not None and draft_id is not None and message.bot is not None: try: await message.bot.send_message_draft( chat_id=message.chat.id, draft_id=draft_id, text=text, ) return except Exception as exc: logger.warning("Draft update failed (%s): %s", type(exc).__name__, str(exc)[:120]) self.draft_id = None try: self.status_message = await message.answer(text) except Exception as send_exc: logger.warning( "Status fallback send failed (%s): %s", type(send_exc).__name__, str(send_exc)[:120] ) return if self.status_message is None: return try: await self.status_message.edit_text(text) except Exception as exc: logger.warning("Status message edit failed (%s): %s", type(exc).__name__, str(exc)[:120]) async def finish(self, text: str | None = None) -> None: """Close the status out. A live draft only disappears once the bot sends a real message. Its content must match that message, otherwise the client cannot merge them: the draft keeps floating in the chat for a few more seconds while already being stale. So the final text is pushed into the draft first, then sent as a real message with the same text. """ await self.cancel() self._closed = True if text is None: await self._clear_draft() return self._last_text = text if self.message is not None and self._finished_draft_id is not None: # Bring the draft to the final text so the real message replaces it # instead of racing a stale one. await self._push_draft(self._finished_draft_id, text) self.draft_id = None try: await self.message.answer(text) except Exception as exc: logger.warning("Status finish send failed (%s): %s", type(exc).__name__, str(exc)[:120]) return await self.update(text) async def _push_draft(self, draft_id: int, text: str) -> None: message = self.message if message is None or message.bot is None: return try: await message.bot.send_message_draft( chat_id=message.chat.id, draft_id=draft_id, text=text, ) except Exception as exc: logger.debug("Draft push failed (%s): %s", type(exc).__name__, str(exc)[:120]) async def _clear_draft(self) -> None: """Drop the draft without leaving any text behind.""" draft_id = self._finished_draft_id if draft_id is None: return await self._push_draft(draft_id, "") async def finish_if_open(self, text: str) -> None: """Safety net: close the status only if nothing has closed it yet.""" if not self._closed: await self.finish(text) async def cancel(self) -> None: """Stop the phrase rotation without changing the visible status.""" task, self._rotation_task = self._rotation_task, None if task is None: return task.cancel() try: await task except asyncio.CancelledError: pass def _draft_supported(message: Message | None) -> bool: """Live drafts only exist in private chats; groups get the legacy path.""" if message is None: return False if getattr(message, "bot", None) is None: return False chat = getattr(message, "chat", None) if chat is None: return False return str(getattr(chat, "type", "") or "").lower() == "private" async def finish_status(status: "OperationStatus | None", text: str) -> None: """Close an operation status if one is open and not closed yet. Allows ``status`` to be None so callers can guard every exit path without repeating ``if status:`` — an unfinished draft would otherwise keep showing its last activity phrase in the chat. """ if status is not None: await status.finish_if_open(text) async def _classify_and_respond( message: Message, text: str, show_typing: bool = True, state: FSMContext | None = None, operation_status=None, ) -> None: with diagnostic_operation(): await _classify_and_respond_inner(message, text, show_typing, state, operation_status) async def _classify_and_respond_inner( message: Message, text: str, show_typing: bool = True, state: FSMContext | None = None, operation_status=None, ) -> None: if message.from_user is None: await message.answer("Не удалось определить пользователя.") return if text.strip().casefold() == "/help": if operation_status is not None: await operation_status.finish_if_open("✅ Открываю справку") await help_command(message, state) return if not await _check_subscription(message): if operation_status is not None: await operation_status.finish_if_open("⚠️ Нужна активная подписка") return reply_message = message.reply_to_message original_post = "" if reply_message and reply_message.from_user and reply_message.from_user.is_bot: original_post = (reply_message.text or reply_message.caption or "").strip() status = operation_status try: if show_typing and status is None: status = await OperationStatus.start(message, "⏳ Разбираю запрос…", kind="generation") elif status: await status.update("⏳ Разбираю запрос…") async with (typing_status(message.bot, message.chat.id) if show_typing else nullcontext()): intent = await classify_intent(text, original_post or None, user_id=message.from_user.id) if intent["intent"] == "generate_ideas" and status is not None: status.retarget("ideas") if intent["intent"] == "help": if status: await status.finish("✅ Открываю справку") await help_command(message, state) return if intent["intent"] in {"thanks", "other"}: replies = THANKS_REPLIES if intent["intent"] == "thanks" else OFF_TOPIC_REPLIES reply = random.choice(replies) if status: await status.finish(reply) else: await message.answer(reply) return if await classify_prompt_injection(text, user_id=message.from_user.id): if status: await status.finish(INJECTION_BLOCK_REPLY) else: await message.answer(INJECTION_BLOCK_REPLY) return if intent["intent"] == "generate_ideas": if status: await status.update("⏳ Подбираю идею…") tone = await asyncio.to_thread(get_tone_of_voice, message.from_user.id) history = await asyncio.to_thread( get_recent_post_history_text, message.from_user.id, POST_HISTORY_LIMIT, POST_HISTORY_CHAR_LIMIT, ) idea_history = await asyncio.to_thread(get_idea_history, message.from_user.id, 30) idea = await generate_ideas(text, tone, history, idea_history, user_id=message.from_user.id) if status: await status.finish("✅ Идея готова") await _send_idea(message, idea, message.from_user.id) return if intent["intent"] == "revise_post": if not original_post: if status: await status.finish("⚠️ Не найден исходный вариант") await message.answer("Чтобы исправить пост, ответь реплаем на сообщение с его вариантом.") return tone = await asyncio.to_thread(get_tone_of_voice, message.from_user.id) if status: await status.update("⏳ Редактирую вариант…") posts = await generate_posts( intent, tone, original_post=original_post, edit_request=text, user_id=message.from_user.id, ) origin = "revised" else: tone = await asyncio.to_thread(get_tone_of_voice, message.from_user.id) if status: await status.update("⏳ Генерирую варианты поста…") posts = await generate_posts(intent, tone, user_id=message.from_user.id) origin = "generated" if status: await status.finish("✅ Посты готовы") await _send_three_posts(message, posts, user_id=message.from_user.id, origin=origin) except Exception as exc: if status: await status.finish("❌ Не удалось выполнить запрос") await report_error(message, "классификация или генерация поста", exc) async def _download_message_image(message: Message, bot: Bot) -> tuple[bytes, str] | None: document = message.document if document: mime_type = (document.mime_type or "").lower() if not mime_type.startswith("image/"): return None file_obj = document elif message.photo: file_obj = message.photo[-1] mime_type = "image/jpeg" else: return None if file_obj.file_size and file_obj.file_size > MAX_IMAGE_BYTES: await message.answer("Скриншот слишком большой. Пришли изображение до 10 МБ.") return None output = io.BytesIO() await bot.download(file_obj, destination=output) image = output.getvalue() if not image: await message.answer("Не получилось скачать изображение. Попробуй ещё раз.") return None if len(image) > MAX_IMAGE_BYTES: await message.answer("Скриншот слишком большой. Пришли изображение до 10 МБ.") return None return image, mime_type async def _download_image_group(messages: list[Message], bot: Bot) -> list[tuple[bytes, str]]: if len(messages) > 10: raise ValueError("В одной галерее можно обработать не более 10 изображений.") images: list[tuple[bytes, str]] = [] total_bytes = 0 for message in messages: payload = await _download_message_image(message, bot) if payload: if payload[1] not in {"image/jpeg", "image/png", "image/webp"}: raise ValueError("Поддерживаются только статичные изображения JPEG, PNG и WebP.") images.append(payload) total_bytes += len(payload[0]) if total_bytes > MAX_IMAGE_BYTES * 3: raise ValueError("Общий размер галереи превышает 30 МБ.") return images async def _process_image_group(messages: list[Message], bot: Bot, status=None) -> None: first = messages[0] if first.from_user is None: await first.answer("Не удалось определить пользователя.") return try: images = await _download_image_group(messages, bot) except ValueError as exc: if status: await status.finish("⚠️ Не удалось обработать галерею") await first.answer(str(exc)) return if not images: if status: await status.finish("⚠️ Не удалось получить изображения") return if status: await status.update("⏳ Извлекаю тексты со скриншотов…") new_samples = await extract_posts_from_images(images, user_id=first.from_user.id) if not new_samples: if status: await status.finish("⚠️ Не нашла читаемых текстов") await first.answer("Не нашла читаемых текстов на скриншотах. Обучение не меняла.") return await _rebuild_account_profile(first.from_user.id, first, new_posts=new_samples, status=status) async def _collect_media_group(message: Message, bot: Bot) -> list[Message]: media_group_id = message.media_group_id if not media_group_id: return [message] key = (message.chat.id, media_group_id) future: asyncio.Future[list[Message]] = asyncio.get_running_loop().create_future() async with _media_groups_lock: group = _media_groups.get(key) if group is None: group = {"messages": [message], "future": future} _media_groups[key] = group asyncio.create_task(_finish_media_group(key)) else: group["messages"].append(message) future = group["future"] try: return await future except Exception: async with _media_groups_lock: group = _media_groups.get(key) if group: try: group["messages"].remove(message) except ValueError: pass raise async def _finish_media_group(key: tuple[int, str]) -> None: await asyncio.sleep(MEDIA_GROUP_COLLECT_SECONDS) async with _media_groups_lock: group = _media_groups.pop(key, None) if group is None: return group["future"].set_result(group["messages"]) async def _generate_posts_from_images(message: Message, images, request: str, status) -> None: if message.from_user is None: await status.finish("❌ Не удалось определить пользователя") return try: if await classify_prompt_injection(request, user_id=message.from_user.id): await status.finish(INJECTION_BLOCK_REPLY) return await status.update("⏳ Генерирую варианты по изображениям…") tone = await asyncio.to_thread(get_tone_of_voice, message.from_user.id) posts = await generate_posts_from_images(images, request, tone, user_id=message.from_user.id) await status.finish("✅ Посты готовы") await _send_three_posts(message, posts, user_id=message.from_user.id) except Exception as exc: await status.finish("❌ Не удалось подготовить пост по изображениям") await report_error(message, "анализ изображений и генерация поста", exc) async def _handle_image_message(message: Message, bot: Bot, state: FSMContext) -> None: mode = await current_mode(state) if message.from_user is not None and mode in {"generation", "training"}: await asyncio.to_thread(analytics.record_event, message.from_user.id, mode, "user_message", "image") if (await notify_development_deploy(bot, state, message.from_user.id if message.from_user else None)) is True: return with diagnostic_operation(): status = None try: if message.media_group_id: messages = await _collect_media_group(message, bot) if messages[0].message_id != message.message_id: return else: messages = [message] mode = await current_mode(state) if mode == "generation": caption = next(((item.caption or "").strip() for item in messages if (item.caption or "").strip()), "") images = await _download_image_group(messages, bot) if not images: await message.answer("Не удалось получить изображения. Попробуй отправить их ещё раз.") return if not caption: pending = await message.answer( "О чём написать пост по этой картинке? Напиши тему, цель или нужный посыл." ) if message.from_user is not None: await state.update_data( pending_image_request={ "images": images, "question_message_id": pending.message_id, "album_message_id": messages[0].message_id, } ) return status = await OperationStatus.start( message, "⏳ Изучаю изображения и готовлю посты…", kind="images" ) async with typing_status(bot, message.chat.id): await _generate_posts_from_images(message, images, caption, status) return status = await OperationStatus.start( message, "⏳ Читаю посты на изображениях…", kind="images" ) async with typing_status(bot, message.chat.id): await status.update("⏳ Анализирую стиль и обновляю профиль…") await _process_image_group(messages, bot, status=status) except Exception as exc: await finish_status(status, "❌ Не удалось обработать изображения") await report_error(message, "чтение скриншота или обновление ToV", exc) else: await finish_status(status, "✅ Готово") @router.message(F.photo | (F.document & F.document.mime_type.startswith("image/") & (F.document.mime_type != "image/gif"))) async def update_tone_from_image(message: Message, bot: Bot, state: FSMContext) -> None: mode = await current_mode(state) if mode not in {"training", "generation"}: await message.answer("Картинки можно использовать для настройки стиля и подготовки поста.", reply_markup=keyboard_for_mode(mode)) return await _handle_image_message(message, bot, state) async def _process_voice(message: Message, bot: Bot, state: FSMContext, status=None) -> None: if not os.environ.get("APINEX_API_KEY", "").strip(): if status: await status.finish("❌ Расшифровка аудио не настроена") await message.answer("Расшифровка голосовых пока не настроена.") return audio_file = message.voice or message.audio if audio_file is None: if status: await status.finish("⚠️ Аудиофайл не найден") return if audio_file.file_size and audio_file.file_size > MAX_AUDIO_BYTES: if status: await status.finish("⚠️ Аудиофайл превышает лимит 20 МБ") await message.answer("Аудиофайл слишком большой. Максимальный размер — 20 МБ.") return audio_buffer = io.BytesIO() await bot.download(audio_file, destination=audio_buffer) audio = audio_buffer.getvalue() if not audio: if status: await status.finish("❌ Не удалось получить аудиофайл") await message.answer("Не удалось получить аудиофайл.") return if len(audio) > MAX_AUDIO_BYTES: if status: await status.finish("⚠️ Аудиофайл превышает лимит 20 МБ") await message.answer("Аудиофайл слишком большой. Максимальный размер — 20 МБ.") return content_type = "audio/ogg" if message.voice else (message.audio.mime_type or "audio/mpeg") if status: await status.update("⏳ Распознаю речь…") transcript = await transcribe_audio(audio, content_type, user_id=message.from_user.id if message.from_user else None) if not transcript: if status: await status.finish("⚠️ Не удалось распознать речь") await message.answer("Не удалось распознать речь в аудио.") return current_mode_result = await current_mode(state) if message.from_user is not None: await asyncio.to_thread( analytics.record_event, message.from_user.id, current_mode_result, "user_message", "voice" if message.voice else "audio", ) if current_mode_result == "training": if message.from_user is None: if status: await status.finish("❌ Не удалось определить пользователя") await message.answer("Не удалось определить пользователя.") return await _rebuild_account_profile(message.from_user.id, message, new_note=transcript, status=status) return await _classify_and_respond(message, transcript, show_typing=False, state=state, operation_status=status) await finish_status(status, "✅ Готово") @router.message(F.voice | F.audio) async def transcribe_voice(message: Message, bot: Bot, state: FSMContext) -> None: mode = await current_mode(state) if mode not in {"training", "generation"}: await message.answer( "Голосовые можно использовать для настройки стиля и подготовки поста.", reply_markup=keyboard_for_mode(mode), ) return if (await notify_development_deploy(bot, state, message.from_user.id if message.from_user else None)) is True: return with diagnostic_operation(): status = await OperationStatus.start(message, "⏳ Подготавливаю аудио…", kind="voice") try: async with typing_status(bot, message.chat.id): await _process_voice(message, bot, state, status=status) except Exception as exc: await status.finish_if_open("❌ Не удалось обработать аудио") await report_error(message, "обработка голосового и генерация", exc) else: await finish_status(status, "✅ Готово") @router.message( F.video | F.video_note | F.animation | (F.document & F.document.mime_type.startswith("video/")) | (F.document & (F.document.mime_type == "image/gif")) ) async def reject_video_or_animation(message: Message, state: FSMContext) -> None: await message.answer( "Видео и анимации пока не поддерживаются. Пришли изображение или текстовое описание.", reply_markup=keyboard_for_mode(await current_mode(state)), ) @router.message(F.text & F.text.startswith("/")) async def handle_command_text(message: Message, state: FSMContext) -> None: raw = (message.text or "").strip() resolved = resolve_button_text(raw) or raw command = resolved.split(maxsplit=1)[0].split("@", 1)[0].casefold() if command == "/start": await start(message, state) elif command in MODE_COMMANDS: await enter_mode(message, state, MODE_COMMANDS[command]) elif command == "/clear": await clear_command(message, state) elif command == "/style": await style_command(message, state) elif command == "/idea": if await require_mode(message, state, "generation"): await _classify_and_respond( message, "предложи идею для нового поста", state=state ) elif command == "/post": if await require_mode(message, state, "generation"): await _generate_post_from_training(message) elif command == "/balance": await balance_command(message, state) elif command == "/subscribe": await subscribe_command(message, state) elif command == "/support": await open_support(message, state) elif command == "/stats": await stats_command(message, state) else: await message.answer("Неизвестная команда. Открой /help, чтобы посмотреть доступные команды.") @router.message(F.text & ~F.text.startswith("/")) async def handle_text(message: Message, state: FSMContext) -> None: if message.bot is not None and (await notify_development_deploy( message.bot, state, message.from_user.id if message.from_user else None )) is True: return text = (message.text or "").strip() if not text: return button = resolve_button_text(text) if button == MAIN_MENU_VALUE: await show_main_menu(message, state) return if button == SUPPORT_VALUE: await open_support(message, state) return if button == SUBSCRIPTION_ACTION_VALUE: if await require_mode(message, state, "subscription") and message.from_user is not None: await message.answer("Выберите срок подписки", reply_markup=subscription_plans_keyboard()) return if button == TRAINING_INFO_VALUE: await training_info_command(message, state) return if button == TRAINING_RESET_VALUE: await clear_command(message, state) return if button and button.startswith("/"): await handle_command_text(message, state) return mode = await current_mode(state) if message.from_user is not None and mode in {"generation", "training"}: await asyncio.to_thread(analytics.record_event, message.from_user.id, mode, "user_message", "text") pending_image = (await state.get_data()).get("pending_image_request") if mode == "generation" and pending_image: await state.update_data(pending_image_request=None) status = await OperationStatus.start( message, "⏳ Изучаю изображения и готовлю посты…", kind="images" ) async with typing_status(message.bot, message.chat.id): await _generate_posts_from_images(message, pending_image["images"], text, status) return if mode == "training": await process_training_text(message, text) return if mode != "generation": prompts = { "lobby": "Выбери действие кнопкой ниже.", "training": "Пришли пост текстом, скриншотом, голосовым или аудиофайлом — профиль будет накапливаться.", "subscription": "Статус уже показан выше. Нажми кнопку покупки или продления подписки.", "help": "Открой нужный раздел справки выше или выбери другое действие.", } await message.answer( prompts.get(mode, "Выбери действие."), reply_markup=keyboard_for_mode(mode), ) return if button: text = button await _classify_and_respond(message, text, state=state) @router.message() async def handle_other_message(message: Message, state: FSMContext) -> None: mode = await current_mode(state) if mode == "training": text = "Пришли пост текстом, скриншотом, голосовым или аудиофайлом. Профиль будет постепенно дополняться." elif mode == "generation": text = "Для генерации отправь текстовый запрос или голосовое сообщение." elif mode == "subscription": text = "Статус уже показан выше. Нажми кнопку покупки или продления подписки." elif mode == "help": text = "Выбери раздел справки или другое действие в меню." else: text = "Выбери действие кнопкой ниже." await message.answer(text, reply_markup=keyboard_for_mode(mode)) async def send_owner_stats(bot: Bot, period: str = "day") -> None: if not OWNER_USER_ID: return report = await asyncio.to_thread(analytics.render_report, period) await bot.send_rich_message(chat_id=ANALYTICS_CHAT_ID, message_thread_id=ANALYTICS_TOPIC_ID, rich_message=InputRichMessage(html=report)) @admin_router.message(F.text.regexp(r"^/stats(?:\s+(day|week|month|all))?$") ) async def stats_command(message: Message, state: FSMContext) -> None: if message.chat.id != ANALYTICS_CHAT_ID or message.message_thread_id != ANALYTICS_TOPIC_ID: return if OWNER_USER_ID and (not message.from_user or message.from_user.id != OWNER_USER_ID): return period = ((message.text or "").split(maxsplit=1)[1] if len((message.text or "").split()) > 1 else "all") report = await asyncio.to_thread(analytics.render_report, period) if message.bot is None: return await message.bot.send_rich_message(chat_id=ANALYTICS_CHAT_ID, message_thread_id=ANALYTICS_TOPIC_ID, rich_message=InputRichMessage(html=report)) async def set_main_menu(bot: Bot) -> None: """Публикует команды переключения режимов и действия текущих режимов.""" try: await bot.set_my_commands( commands=[ BotCommand(command="start", description="Открыть главное меню"), BotCommand(command="training", description="Режим: обучение"), BotCommand(command="generate", description="Режим: генерация"), BotCommand(command="subscription", description="Режим: подписка"), BotCommand(command="help", description="Режим: помощь"), BotCommand(command="post", description="Написать пост (генерация)"), BotCommand(command="style", description="Мой стиль (обучение)"), BotCommand(command="clear", description="Сбросить данные (обучение, с подтверждением)"), BotCommand(command="idea", description="Идея (генерация)"), BotCommand(command="balance", description="Статус подписки"), BotCommand(command="subscribe", description="Оплатить подписку"), BotCommand(command="support", description="Написать в поддержку"), BotCommand(command="stats", description="Статистика бота"), ] ) except Exception as exc: logger.warning("Bot commands publish failed (%s): %s", type(exc).__name__, str(exc)[:120]) async def main() -> None: init_db() analytics.init_db() payments.init_payments_table() await asyncio.to_thread(support.init_support_table) if is_development_environment(): await asyncio.to_thread(init_raw_updates_db) token = os.environ.get("BOT_TOKEN", "").strip() if not token: raise RuntimeError("BOT_TOKEN is required") test_env = os.environ.get("BOT_ENV", "test").strip().lower() == "test" session = AiohttpSession(api=TEST) if test_env else None bot = Bot(token=token, session=session) await notify_development_deploy(bot) dp = Dispatcher(storage=MemoryStorage()) configure_raw_update_logging(dp) dp.include_router(payments.router) # Роли: групповые сценарии регистрируются ДО основного роутера, иначе # catch-all @router.message() основного роутера перехватывает их апдейты. dp.include_router(support.router) dp.include_router(admin_router) dp.include_router(router) dp.message.outer_middleware(support.SupportRelayMiddleware()) # Ставим меню один раз при старте, чтобы оно было у всех, кто писал боту. await set_main_menu(bot) try: await dp.start_polling(bot) finally: await bot.session.close() if __name__ == "__main__": asyncio.run(main())