Files
social-post-writer-bot/bot.py
T

1188 lines
53 KiB
Python

import asyncio
import html
import io
import logging
import os
import random
import re
import sqlite3
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,
save_idea,
save_generated_posts,
save_tone_of_voice,
set_deploy_notice_version,
SUBSCRIPTION_PRICE_STARS,
ensure_subscription,
subscription_row,
)
logging.basicConfig(level=logging.INFO)
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(
"❌ <b>Не удалось выполнить действие</b>\n\n"
f"{html.escape(stage.capitalize())}. Попробуй ещё раз позже.",
parse_mode="HTML",
)
if is_development_environment():
await message.answer(
f"🧪 Dev-диагностика: {stage}. ID: <code>{operation_id}</code>. "
f"Причина: <code>{html.escape(detail[:240])}</code>",
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) -> None:
if not is_development_environment():
return
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
async with _deploy_notice_lock:
previous = await asyncio.to_thread(get_deploy_notice_version)
if previous == VERSION:
return
try:
await asyncio.wait_for(
bot.send_message(chat_id=int(chat_id), text=f"🚀 Бот обновлён\nВерсия: {VERSION}"),
timeout=8,
)
except Exception as exc:
logger.error("Development deploy notice failed (%s): %s", type(exc).__name__, str(exc)[:160])
return
await asyncio.to_thread(set_deploy_notice_version, VERSION)
logger.info("Development deploy notice sent: version=%s", VERSION)
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"🧾 <b>Счёт на подписку</b>\n"
f"Сумма: {SUBSCRIPTION_PRICE_STARS} ⭐ · Срок: 30 дней\n"
f"ID пользователя: <code>{user_id}</code>",
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(
"🎓 <b>Профиль аккаунта ещё не настроен</b>\n\n"
"Пришли посты скриншотами или текстом, расскажи об аккаунте или отправь голосовое.\n"
"Я буду постепенно накапливать сведения о твоём аккаунте и стиле.",
parse_mode="HTML",
reply_markup=mode_keyboard("training"),
)
return
await message.answer(
f"🎨 <b>Профиль аккаунта</b>\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 = (
"<h2>✅ Профиль аккаунта обновлён</h2>"
f"<p>Накоплено постов: {len(samples)}. Дополнений профиля: {len(notes)}.</p>"
f"<p>{html.escape(profile)}</p>"
"<p>Присылай новые посты или сведения в режиме «Обучение» — профиль будет дополняться.</p>"
)
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(
"⚠️ <b>Сбросить данные обучения?</b>\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(
"✅ <b>Данные обучения сброшены.</b>\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 _send_idea(message: Message, idea: dict[str, str], user_id: int) -> None:
await message.answer(
f"💡 <b>{html.escape(idea['title'])}</b>\n\n"
f"<b>Угол:</b> {html.escape(idea['angle'])}\n\n"
f"<b>Зачем:</b> {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:
await notify_development_deploy(bot)
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
await notify_development_deploy(bot)
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 _classify_and_respond(message, "Напиши пост", state=state)
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:
await notify_development_deploy(message.bot)
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())