import asyncio
import html
import io
import logging
from logging.handlers import RotatingFileHandler
import os
import random
import re
import sqlite3
import sys
from contextlib import asynccontextmanager, nullcontext
from pathlib import Path
from aiogram import Bot, Dispatcher, F, Router
from aiogram.client.session.aiohttp import AiohttpSession
from aiogram.client.telegram import TEST
from aiogram.filters import CommandStart
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.types import (
BotCommand,
CallbackQuery,
InlineKeyboardButton,
InlineKeyboardMarkup,
InputRichMessage,
Message,
ReplyKeyboardMarkup,
)
import payments
import user_store
from apinex_client import APINEX_STT_URL, POST_HISTORY_CHAR_LIMIT, POST_HISTORY_LIMIT, classify_intent, classify_prompt_injection, create_tone_profile, current_operation_id, diagnostic_operation, extract_posts_from_images, extract_transcript, generate_ideas, generate_posts, generate_posts_from_images, transcribe_audio
from help_message import WELCOME_RICH_HTML, help_plain_view, help_view, rich_to_plain
from keyboards import MAIN_MENU_VALUE, TRAINING_RESET_VALUE, main_menu_keyboard, mode_keyboard, resolve_button_text
from payments import send_subscription_invoice
from user_store import (
add_post_samples,
clear_user_data,
get_deploy_notice_version,
get_idea_history,
get_post_samples,
get_training_notes,
add_training_note,
get_recent_post_history_text,
get_tone_of_voice,
init_db,
is_subscribed,
is_user_trained,
save_idea,
save_generated_posts,
save_tone_of_voice,
set_deploy_notice_version,
SUBSCRIPTION_PRICE_STARS,
ensure_subscription,
subscription_row,
)
def _configure_logging() -> None:
handlers: list[logging.Handler] = [logging.StreamHandler()]
try:
log_path = Path(os.getenv("BOT_LOG_PATH", "/app/data/bot.log"))
log_path.parent.mkdir(parents=True, exist_ok=True)
handlers.append(RotatingFileHandler(
log_path,
maxBytes=5 * 1024 * 1024,
backupCount=3,
encoding="utf-8",
))
except OSError as exc:
print(f"File logging unavailable ({type(exc).__name__}); using container logs", file=sys.stderr)
logging.basicConfig(level=logging.INFO, handlers=handlers)
_configure_logging()
logger = logging.getLogger(__name__)
VERSION = Path(__file__).with_name("VERSION").read_text(encoding="utf-8").strip()
BOT_ENV = os.getenv("BOT_ENV", "test").strip().lower()
DEPLOY_ENV = os.getenv("DEPLOY_ENV", "production")
_deploy_notice_lock = asyncio.Lock()
router = Router()
MAX_AUDIO_BYTES = 20 * 1024 * 1024
MAX_IMAGE_BYTES = 10 * 1024 * 1024
TYPING_REFRESH_SECONDS = 4
THANKS_REPLIES = (
"Пожалуйста! Если понадобится пост или идея — напиши.",
"Рада помочь! Обращайся, когда понадобится текст для публикации.",
"Всегда пожалуйста! Готова помочь с постами и идеями.",
)
OFF_TOPIC_REPLIES = (
"Я помогаю с постами и идеями для соцсетей. Напиши, что нужно подготовить.",
"Похоже, это не задача для режима «Генерация». Я могу помочь написать или изменить пост либо предложить идею.",
"В этом режиме я работаю с публикациями для соцсетей. Сформулируй запрос на пост или идею.",
)
INJECTION_BLOCK_REPLY = "Не могу выполнить инструкции, которые пытаются изменить правила работы бота. Могу помочь с постом или идеей для соцсетей."
MEDIA_GROUP_COLLECT_SECONDS = 0.8
_media_groups: dict[tuple[int, str], dict[str, object]] = {}
_media_groups_lock = asyncio.Lock()
class BotMode(StatesGroup):
lobby = State()
training = State()
generation = State()
subscription = State()
help = State()
MODE_STATES: dict[str, State] = {
"lobby": BotMode.lobby,
"training": BotMode.training,
"generation": BotMode.generation,
"subscription": BotMode.subscription,
"help": BotMode.help,
}
MODE_BY_STATE = {value.state: key for key, value in MODE_STATES.items()}
MODE_COMMANDS = {
"/training": "training",
"/generate": "generation",
"/subscription": "subscription",
"/help": "help",
}
MODE_COPY = {
"lobby": ("Главное меню", "Выбери режим работы в меню ниже."),
"training": (
"Обучение",
"Пришли скриншоты или тексты своих постов, опиши аккаунт или отправь голосовое.\n"
"Можно дополнять знания постепенно; «Сбросить данные» начнёт обучение заново.",
),
"generation": (
"Генерация",
"Напиши запрос или отправь голосовое.\nДля правки ответь на мой вариант.",
),
"subscription": (
"Подписка",
"Проверь статус ниже.\nКнопка откроет оплату или продление.",
),
"help": ("Помощь", "Выбери раздел в сообщении выше."),
}
async def current_mode(state: FSMContext) -> str:
return MODE_BY_STATE.get(await state.get_state(), "lobby")
def keyboard_for_mode(mode: str) -> ReplyKeyboardMarkup:
return main_menu_keyboard() if mode == "lobby" else mode_keyboard(mode)
async def enter_mode(message: Message, state: FSMContext, mode: str) -> None:
await state.clear()
await state.set_state(MODE_STATES[mode])
title, description = MODE_COPY[mode]
if mode == "help":
await send_rich_message(message, help_view(None), "помощи")
await message.answer(
f"Режим «{title}».\n{description}",
reply_markup=mode_keyboard(mode),
)
if mode == "subscription" and message.from_user is not None:
sub = await asyncio.to_thread(subscription_row, message.from_user.id)
await _show_subscription_info(message, sub)
async def show_main_menu(message: Message, state: FSMContext) -> None:
await state.clear()
await state.set_state(BotMode.lobby)
await message.answer("Выбери режим работы.", reply_markup=main_menu_keyboard())
async def require_mode(message: Message, state: FSMContext, mode: str) -> bool:
if await current_mode(state) == mode:
return True
title = MODE_COPY[mode][0]
command = next(command for command, target in MODE_COMMANDS.items() if target == mode)
await message.answer(
f"Эта команда работает только в режиме «{title}». Переключись: {command}.",
reply_markup=keyboard_for_mode(await current_mode(state)),
)
return False
def is_development_environment() -> bool:
"""Enable developer-only behavior only for the explicit Coolify dev value."""
return DEPLOY_ENV == "development"
def _read_version() -> str:
version_path = Path(__file__).with_name("VERSION")
return version_path.read_text(encoding="utf-8").strip()
def split_text(text: str, limit: int = 4000) -> list[str]:
chunks: list[str] = []
remaining = text.strip()
while len(remaining) > limit:
split_at = remaining.rfind(" ", 0, limit + 1)
if split_at < limit // 2:
split_at = limit
chunks.append(remaining[:split_at].strip())
remaining = remaining[split_at:].strip()
if remaining:
chunks.append(remaining)
return chunks
async def report_error(message: Message, stage: str, exc: Exception) -> None:
detail = str(exc).strip()
detail = re.sub(r"(?i)(?:sk-[A-Za-z0-9_-]{12,}|[0-9]{8,}:[A-Za-z0-9_-]{20,})", "[секрет скрыт]", detail)
detail = re.sub(r"(?i)bearer\s+[^\s\"']+", "Bearer [секрет скрыт]", detail)
detail = re.sub(r"(?i)(api[_-]?key|access[_-]?token|secret|password)(\s*[=:]\s*)[^\s,;\"']+", r"\1\2[секрет скрыт]", detail)
detail = re.sub(r"https?://\S+", "[внешний сервис]", detail)
detail = detail or "нет подробностей"
operation_id = current_operation_id() or "-"
user_id = message.from_user.id if message.from_user else "-"
try:
await message.answer(
"❌ Не удалось выполнить действие\n\n"
f"{html.escape(stage.capitalize())}. Попробуй ещё раз позже.",
parse_mode="HTML",
)
if is_development_environment():
await message.answer(
f"🧪 Dev-диагностика: {stage}. ID: {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) -> 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),
)
return True
except Exception as exc:
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
if await current_mode(state) != "help":
await callback.answer("Сначала открой режим /help.", show_alert=True)
return
section = (callback.data or "help:menu").split(":", 1)[1]
if not await edit_rich_message(message, help_view(section)):
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:
user_id = int((callback.data or "").split(":", 1)[1])
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
await asyncio.to_thread(ensure_subscription, user_id)
days = await asyncio.to_thread(user_store.activate_subscription, user_id, months=1)
await callback.answer()
if isinstance(callback.message, Message):
await callback.message.edit_reply_markup(reply_markup=None)
await callback.message.answer(
f"🧪 Dev-платёж подтверждён. Подписка включена на {days} дн. Реальная оплата не проводилась."
)
@router.message(CommandStart())
async def start(message: Message, state: FSMContext) -> None:
if message.from_user is None:
await message.answer("Не удалось определить пользователя.")
return
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 = (
f"Подписка не активирована. Оформи её за {SUBSCRIPTION_PRICE_STARS} звёзд "
"за 30 дней, чтобы пользоваться генерацией."
)
else:
days_left = user_store._days_until(sub)
if sub["last_paid_at"]:
if days_left >= 0:
text = f"Активирована платная подписка. Осталось {days_left} дн."
else:
text = (
f"Срок платной подписки истёк. Продление — "
f"{SUBSCRIPTION_PRICE_STARS} звёзд за 30 дней."
)
elif days_left >= 0:
text = f"Активирован бесплатный пробный период. Осталось {days_left} дн."
else:
text = (
f"Бесплатный пробный период истёк. Подписка — "
f"{SUBSCRIPTION_PRICE_STARS} звёзд за 30 дней."
)
if mode == "subscription":
text += f"\n\nПродление на 30 дней — {SUBSCRIPTION_PRICE_STARS} ⭐."
if message.from_user is not None:
text += f"\nID пользователя: {message.from_user.id}"
markup = subscription_renew_keyboard()
else:
markup = mode_keyboard(mode)
await message.answer(text, reply_markup=markup)
def subscription_renew_keyboard() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(
inline_keyboard=[[
InlineKeyboardButton(
text="Продлить подписку",
callback_data="subscription:renew",
)
]]
)
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_subscription_invoice(message: Message, user_id: int) -> None:
await asyncio.to_thread(ensure_subscription, user_id)
try:
await send_subscription_invoice(
message.bot, message.chat.id, message.message_thread_id
)
except Exception as exc:
logger.error("Invoice send failed (%s): %s", type(exc).__name__, str(exc)[:180])
if is_development_environment():
await report_error(message, "отправка счёта на оплату", exc)
else:
await message.answer(
"Не получилось открыть счёт на оплату. Попробуй ещё раз через минуту "
"или напиши поддержке."
)
return
await message.answer(
f"🧾 Счёт на подписку\n"
f"Сумма: {SUBSCRIPTION_PRICE_STARS} ⭐ · Срок: 30 дней\n"
f"ID пользователя: {user_id}",
parse_mode="HTML",
)
if is_development_environment():
await message.answer(
"🧪 Dev-режим: для продолжения разработки нажми кнопку ниже. "
"Она включит доступ без реального платежа.",
reply_markup=InlineKeyboardMarkup(
inline_keyboard=[[InlineKeyboardButton(
text="🧪 Оплатил (dev)",
callback_data=f"dev_paid:{user_id}",
)]]
),
)
else:
await message.answer(
"Подписка активируется сразу после оплаты — отдельного подтверждения не нужно."
)
@router.callback_query(F.data == "subscription:renew")
async def subscription_renew(callback: CallbackQuery, state: FSMContext) -> None:
if await current_mode(state) != "subscription":
await callback.answer("Открой режим «Подписка» из главного меню.", show_alert=True)
return
message = callback.message
if not isinstance(message, Message) or callback.from_user is None:
await callback.answer("Не удалось открыть оплату. Попробуй ещё раз.", show_alert=True)
return
await callback.answer()
await _send_subscription_invoice(message, callback.from_user.id)
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"),
)
if samples or notes:
await message.answer(f"Накоплено постов: {len(samples)}. Дополнений профиля: {len(notes)}.")
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)
if status is None:
rich_html = (
"
Накоплено постов: {len(samples)}. Дополнений профиля: {len(notes)}.
" f"{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, "⏳ Обрабатываю сведения об аккаунте…") 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), "помощи") 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, "⏳ Выбираю тему по обученным данным…") 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) 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) @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 message edited throughout a long operation.""" def __init__(self, status_message: Message | None): self.status_message = status_message @classmethod async def start(cls, message: Message, text: str) -> "OperationStatus": try: return cls(await message.answer(text)) except Exception as exc: logger.warning("Status message send failed (%s): %s", type(exc).__name__, str(exc)[:120]) return cls(None) async def update(self, text: str) -> None: 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: await self.update(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": await help_command(message, state) return if not await _check_subscription(message): 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, "⏳ Разбираю запрос…") 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"] == "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: 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, "⏳ Изучаю изображения и готовлю посты…") async with typing_status(bot, message.chat.id): await _generate_posts_from_images(message, images, caption, status) return status = await OperationStatus.start(message, "⏳ Читаю посты на изображениях…") async with typing_status(bot, message.chat.id): await status.update("⏳ Анализирую стиль и обновляю профиль…") await _process_image_group(messages, bot, status=status) except Exception as exc: if status: await status.finish("❌ Не удалось обработать изображения") await report_error(message, "чтение скриншота или обновление ToV", exc) @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 if await current_mode(state) == "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) @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, "⏳ Подготавливаю аудио…") try: async with typing_status(bot, message.chat.id): await _process_voice(message, bot, state, status=status) except Exception as exc: await status.finish("❌ Не удалось обработать аудио") await report_error(message, "обработка голосового и генерация", exc) @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) 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 == 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) 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, "⏳ Изучаю изображения и готовлю посты…") 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 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="Оплатить подписку"), ] ) except Exception as exc: logger.warning("Bot commands publish failed (%s): %s", type(exc).__name__, str(exc)[:120]) async def main() -> None: init_db() payments.init_payments_table() 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()) dp.include_router(payments.router) dp.include_router(router) # Ставим меню один раз при старте, чтобы оно было у всех, кто писал боту. await set_main_menu(bot) try: await dp.start_polling(bot) finally: await bot.session.close() if __name__ == "__main__": asyncio.run(main())