467 lines
20 KiB
Python
467 lines
20 KiB
Python
import asyncio
|
|
import html
|
|
import io
|
|
import logging
|
|
import os
|
|
import re
|
|
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 BufferedInputFile, Message
|
|
|
|
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, synthesize_idea_voice, transcribe_audio
|
|
from help_message import HELP_RICH_HTML, rich_to_plain
|
|
from user_store import (
|
|
add_post_samples,
|
|
clear_user_data,
|
|
get_deploy_notice_version,
|
|
get_idea_history,
|
|
get_post_history,
|
|
get_recent_post_history_text,
|
|
get_tone_of_voice,
|
|
init_db,
|
|
save_idea,
|
|
save_generated_posts,
|
|
save_tone_of_voice,
|
|
set_deploy_notice_version,
|
|
)
|
|
|
|
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Причина: <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__)
|
|
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:
|
|
await notify_development_deploy(message.bot)
|
|
await message.answer(
|
|
"Я помогаю писать посты для соцсетей. Пришли скриншот своих постов, чтобы настроить стиль, "
|
|
"или попроси написать пост, например: «Придумай пост про запуск нового продукта». "
|
|
"Подробная инструкция — /help."
|
|
)
|
|
|
|
|
|
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-логи для отладки сохранены. Пришли новые скриншоты, чтобы настроить стиль заново."
|
|
)
|
|
|
|
|
|
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 _send_idea_voice(message: Message, idea: dict[str, str], user_id: int) -> None:
|
|
text = f"{idea['title']}. {idea['angle']}. Зачем: {idea['why']}"
|
|
audio = await synthesize_idea_voice(text, user_id=user_id)
|
|
await message.answer_voice(BufferedInputFile(audio, filename="idea.mp3"))
|
|
await asyncio.to_thread(save_idea, user_id, idea)
|
|
|
|
|
|
async def _send_idea(message: Message, idea: dict[str, str], user_id: int) -> None:
|
|
await message.answer(f"{idea['title']}\n\n{idea['angle']}\n\nЗачем: {idea['why']}")
|
|
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
|
|
|
|
|
|
async def _classify_and_respond(message: Message, text: str, show_typing: bool = True) -> None:
|
|
if message.from_user is None:
|
|
await message.answer("Не удалось определить пользователя.")
|
|
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)
|
|
|
|
|
|
async def idea_voice_command(message: Message) -> None:
|
|
if message.from_user is None:
|
|
await message.answer("Не удалось определить пользователя.")
|
|
return
|
|
try:
|
|
async with typing_status(message.bot, message.chat.id):
|
|
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("предложи идею для поста", tone, history, idea_history, user_id=message.from_user.id)
|
|
await _send_idea_voice(message, idea, message.from_user.id)
|
|
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 == "/idea_voice":
|
|
await idea_voice_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())
|