import asyncio import html import io import logging import os import re import sqlite3 from contextlib import asynccontextmanager, nullcontext from pathlib import Path import aiohttp 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.types import Message import user_store from apinex_client import APINEX_STT_URL, POST_HISTORY_CHAR_LIMIT, POST_HISTORY_LIMIT, classify_intent, create_tone_profile, extract_posts_from_images, extract_transcript, generate_ideas, generate_posts, transcribe_audio from help_message import HELP_RICH_HTML, rich_to_plain from user_store import ( add_post_samples, clear_user_data, deactivate_subscription, get_deploy_notice_version, get_idea_history, get_post_history, 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, activate_subscription, ) 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").strip().lower() DEV_CHAT_ID = os.getenv("DEV_CHAT_ID", "").strip() HELP_COMMANDS = {"help", "помощь"} _deploy_notice_lock = asyncio.Lock() router = Router() MAX_AUDIO_BYTES = 20 * 1024 * 1024 MAX_IMAGE_BYTES = 10 * 1024 * 1024 TYPING_REFRESH_SECONDS = 4 MEDIA_GROUP_COLLECT_SECONDS = 0.8 _media_groups: dict[tuple[int, str], dict[str, object]] = {} _media_groups_lock = asyncio.Lock() _deploy_notice_sent = False _deploy_notice_lock = asyncio.Lock() 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"https?://\S+", "[внешний сервис]", detail) detail = detail or "нет подробностей" try: await message.answer(f"❌ Не получилось: {stage}.\nПричина: {html.escape(detail[:240])}", parse_mode="HTML") except Exception as send_exc: logger.error("Unable to report processing error: %s", type(send_exc).__name__) logger.error("User operation failed at %s (%s): %s", stage, type(exc).__name__, detail[:240]) async def notify_development_deploy(bot: Bot) -> None: if DEPLOY_ENV != "development": 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_help(message: Message) -> bool: bot = message.bot payload = {"chat_id": message.chat.id, "rich_message": {"html": HELP_RICH_HTML}} if message.message_thread_id is not None: payload["message_thread_id"] = message.message_thread_id url = bot.session.api.api_url(token=bot.token, method="sendRichMessage") timeout = aiohttp.ClientTimeout(total=15) try: async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post( url, json=payload, headers={"User-Agent": "Mozilla/5.0 (compatible; SocialPostWriterBot/1.0)"}, ) as response: data = await response.json(content_type=None) if response.status == 200 and data.get("ok"): return True logger.warning("Rich help rejected with HTTP %s", response.status) return False except Exception as exc: logger.warning("Rich help delivery error (%s): %s", type(exc).__name__, str(exc)[:160]) return False @router.message(CommandStart()) async def start(message: Message) -> None: if message.from_user is None: await message.answer("Не удалось определить пользователя.") return await notify_development_deploy(message.bot) await asyncio.to_thread(ensure_subscription, message.from_user.id) await message.answer( "Привет! Я помогаю писать посты для соцсетей. Расскажи, о чём нужен пост — " "а я верну три варианта: короткий, развёрнутый и с призывом к действию." ) await asyncio.sleep(0.8) await message.answer( "Примеры запросов:\n" "• «Напиши пост про запуск нового продукта»\n" "• «Сделай пост-напоминание о вебинаре в официальном тоне»\n" "• «Обнови этот вариант, добавь эмодзи» (реплаем на мой текст)" ) await asyncio.sleep(0.8) await message.answer( "Также я умею запоминать твой стиль: пришли скриншоты своих постов. " "Хочешь идею — напиши «предложи тему». Подробно — /help" ) async def balance_command(message: Message) -> None: 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) -> None: if message.from_user is None: await message.answer("Не удалось определить пользователя.") return await asyncio.to_thread(ensure_subscription, message.from_user.id) await message.answer( f"Активирую подписку на 30 дней. " f"Для оплаты пришли {SUBSCRIPTION_PRICE_STARS} звёзд в этом чате.\n\n" f"Жду оплаты..." ) days = await asyncio.to_thread(activate_subscription, message.from_user.id, months=1) await asyncio.sleep(0.6) await message.answer( f"Подписка активирована на {days} дн. Спасибо за оплату {SUBSCRIPTION_PRICE_STARS} звёзд!\n\n" f"Теперь доступны генерация постов, идей и голосовые." ) async def clear_command(message: Message) -> None: if message.from_user is None: await message.answer("Не удалось определить пользователя.") return await asyncio.to_thread(clear_user_data, message.from_user.id) await message.answer( "Команда /clear полностью обнулила твои данные: Tone of Voice, тексты со скриншотов, историю постов, историю идей и настройки. " "API-логи для отладки сохранены. Пришли новые скриншоты, чтобы настроить стиль заново." ) deactivate_subscription(message.from_user.id) await message.answer( "Подписка обнулена. У тебя нет активной подписки и триала — сгенерировать пост нельзя. " f"Чтобы продолжить, купи подписку за {SUBSCRIPTION_PRICE_STARS} звёзд командой /subscribe." ) async def _show_subscription_info(message: Message, sub: sqlite3.Row | None) -> None: if not sub: await message.answer("У тебя нет активной подписки или триала. Начни с команды /subscribe.") return days_left = user_store._days_until(sub) if days_left >= 0: if sub["last_paid_at"]: await message.answer( f"Подписка активна — осталось {days_left} дн.\n" f"Следующее списание через {days_left} дн." ) else: await message.answer( f"Триал активен — осталось {days_left} дн.\n" f"Когда триал закончится, нужно оплатить подписку — команда /subscribe." ) else: await message.answer( "Твой период (триал или подписка) истёк. " "Оплати /subscribe, чтобы продолжить. Стоимость — 100 звёзд в месяц." ) if sub["last_paid_at"] is None: await message.answer( f"Чтобы активировать — пришли {SUBSCRIPTION_PRICE_STARS} звёзд командой /subscribe." ) async def balance_command(message: Message) -> None: 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) -> None: if message.from_user is None: await message.answer("Не удалось определить пользователя.") return await asyncio.to_thread(ensure_subscription, message.from_user.id) await message.answer( f"Активирую подписку на 30 дней. " f"Для оплаты пришли {SUBSCRIPTION_PRICE_STARS} звёзд в этом чате.\n\n" f"Жду оплаты..." ) days = await asyncio.to_thread(activate_subscription, message.from_user.id, months=1) await asyncio.sleep(0.6) await message.answer( f"Подписка активирована на {days} дн. Спасибо за оплату {SUBSCRIPTION_PRICE_STARS} звёзд!\n\n" f"Теперь доступны генерация постов, идей и голосовые." ) async def balance_command(message: Message) -> None: 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) -> None: if message.from_user is None: await message.answer("Не удалось определить пользователя.") return await asyncio.to_thread(ensure_subscription, message.from_user.id) await message.answer( f"Активирую подписку на 30 дней. " f"Для оплаты пришли {SUBSCRIPTION_PRICE_STARS} звёзд в этом чате.\n\n" f"Жду оплаты..." ) days = await asyncio.to_thread(activate_subscription, message.from_user.id, months=1) await asyncio.sleep(0.6) await message.answer( f"Подписка активирована на {days} дн. Спасибо за оплату {SUBSCRIPTION_PRICE_STARS} звёзд!\n\n" f"Теперь доступны генерация постов, идей и голосовые." ) async def clear_command(message: Message) -> None: if message.from_user is None: await message.answer("Не удалось определить пользователя.") return await asyncio.to_thread(clear_user_data, message.from_user.id) deactivate_subscription(message.from_user.id) await message.answer( "Команда /clear полностью обнулила твои данные: Tone of Voice, тексты со скриншотов, историю постов, историю идей и настройки. " "API-логи для отладки сохранены. Пришли новые скриншоты, чтобы настроить стиль заново." ) await asyncio.sleep(0.5) await message.answer( "Подписка обнулена. У тебя нет активной подписки и триала — сгенерировать пост нельзя. " f"Чтобы продолжить, купи подписку за {user_store.SUBSCRIPTION_PRICE_STARS} звёзд командой /subscribe." ) async def help_command(message: Message) -> None: try: rich_sent = await send_rich_help(message) except Exception as exc: rich_sent = False logger.warning("Rich help send failed (%s): %s", type(exc).__name__, str(exc)[:180]) if DEPLOY_ENV == "development": await report_error(message, "отправка Rich Message помощи", exc) if not rich_sent: if DEPLOY_ENV == "development": await report_error(message, "отправка Rich Message помощи", RuntimeError("sendRichMessage returned false")) await message.answer(rich_to_plain(HELP_RICH_HTML)) 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) return False return True 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) 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) return False return True @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 async def _classify_and_respond(message: Message, text: str, show_typing: bool = True) -> None: if message.from_user is None: await message.answer("Не удалось определить пользователя.") return # Если подписка не активна — блокируем генерацию, но пропускаем /help if text.strip() != "/help": 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() try: 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": await help_command(message) return if intent["intent"] == "generate_ideas": 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) await _send_idea(message, idea, message.from_user.id) return if intent["intent"] == "revise_post": if not original_post: await message.answer("Чтобы исправить пост, ответь реплаем на сообщение с его вариантом.") return tone = await asyncio.to_thread(get_tone_of_voice, message.from_user.id) 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) posts = await generate_posts(intent, tone, user_id=message.from_user.id) origin = "generated" await _send_three_posts(message, posts, user_id=message.from_user.id, origin=origin) except Exception as exc: 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 _process_image_group(messages: list[Message], bot: Bot) -> None: first = messages[0] if first.from_user is None: await first.answer("Не удалось определить пользователя.") return if len(messages) > 10: await first.answer("В одной галерее можно обработать не более 10 изображений.") return images: list[tuple[bytes, str]] = [] for message in messages: image_payload = await _download_message_image(message, bot) if image_payload: images.append(image_payload) if not images: return previous_tone = await asyncio.to_thread(get_tone_of_voice, first.from_user.id) new_samples = await extract_posts_from_images(images, user_id=first.from_user.id) if not new_samples: await first.answer("Не нашла читаемых текстов постов на скриншотах. Стиль не меняла.") return all_samples = await asyncio.to_thread(add_post_samples, first.from_user.id, new_samples) tone = await create_tone_profile(all_samples, previous_tone, user_id=first.from_user.id) await asyncio.to_thread(save_tone_of_voice, first.from_user.id, tone) await first.answer( f"Обновила только правила стиля по накопленным примерам ({len(all_samples)} постов).\n\n{tone}\n\n" "Распознанные тексты сохранены для следующих обновлений и не переписывались." ) 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 _handle_image_message(message: Message, bot: Bot) -> None: await notify_development_deploy(bot) 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] async with typing_status(bot, message.chat.id): await _process_image_group(messages, bot) except Exception as exc: await report_error(message, "чтение скриншота или обновление ToV", exc) @router.message(F.photo | (F.document & F.document.mime_type.startswith("image/"))) async def update_tone_from_image(message: Message, bot: Bot) -> None: await _handle_image_message(message, bot) async def _process_voice(message: Message, bot: Bot) -> None: if not os.environ.get("APINEX_API_KEY", "").strip(): await message.answer("Расшифровка голосовых пока не настроена.") return audio_file = message.voice or message.audio if audio_file is None: return if audio_file.file_size and audio_file.file_size > MAX_AUDIO_BYTES: await message.answer("Аудиофайл слишком большой. Максимальный размер — 20 МБ.") return audio_buffer = io.BytesIO() await bot.download(audio_file, destination=audio_buffer) audio = audio_buffer.getvalue() if not audio: await message.answer("Не удалось получить аудиофайл.") return if len(audio) > MAX_AUDIO_BYTES: await message.answer("Аудиофайл слишком большой. Максимальный размер — 20 МБ.") return content_type = "audio/ogg" if message.voice else (message.audio.mime_type or "audio/mpeg") transcript = await transcribe_audio(audio, content_type, user_id=message.from_user.id if message.from_user else None) if not transcript: await message.answer("Не удалось распознать речь в аудио.") return await _classify_and_respond(message, transcript, show_typing=False) @router.message(F.voice | F.audio) async def transcribe_voice(message: Message, bot: Bot) -> None: await notify_development_deploy(bot) try: async with typing_status(bot, message.chat.id): await _process_voice(message, bot) except Exception as exc: await report_error(message, "обработка голосового и генерация", exc) @router.message(F.text & F.text.startswith("/")) async def handle_command_text(message: Message) -> None: command = (message.text or "").strip().split(maxsplit=1)[0].split("@", 1)[0].casefold() if command == "/help": await help_command(message) elif command == "/start": await start(message) elif command == "/clear": await clear_command(message) elif command == "/idea": await _classify_and_respond(message, "предложи идею для нового поста") elif command == "/balance": await balance_command(message) elif command == "/subscribe": await subscribe_command(message) else: await message.answer("Неизвестная команда. Используй /help.") @router.message(F.text & ~F.text.startswith("/")) async def handle_text(message: Message) -> None: await notify_development_deploy(message.bot) if message.text: await _classify_and_respond(message, message.text) async def main() -> None: init_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() dp.include_router(router) try: await dp.start_polling(bot) finally: await bot.session.close() if __name__ == "__main__": asyncio.run(main())