Refactor VNDB Telegram Bot

- Remove config.py and integrate environment variable handling directly in docker-compose.yml.
- Delete detailed_handlers.py and replace with simplified command handlers.
- Update requirements.txt to use aiogram and httpx instead of python-telegram-bot and requests.
- Remove test_bot.py and utils.py, consolidating functionality into new VNDBClient class.
- Implement VNDBClient for API interactions, simplifying query methods for visual novels, characters, and releases.
- Clean up docker-compose.yml for improved readability and maintainability.
This commit is contained in:
2026-05-01 18:04:13 +03:00
parent fd0a403f37
commit 88bba02983
19 changed files with 142 additions and 6647 deletions

348
bot.py
View File

@@ -1,297 +1,111 @@
import logging
from typing import Optional
import os
import asyncio
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from vndb import VNDBClient
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import (
Application,
CommandHandler,
MessageHandler,
CallbackQueryHandler,
ConversationHandler,
ContextTypes,
filters,
)
TOKEN = os.getenv("TELEGRAM_TOKEN")
from vndb_client import VndbClient
from config import Config
from utils import Formatter, ErrorHandler, QueryBuilder
from detailed_handlers import get_detail_handlers
bot = Bot(token=TOKEN)
dp = Dispatcher()
vndb = VNDBClient(os.getenv("VNDB_API_URL"))
# ======================
# LOGGING
# ======================
logging.basicConfig(
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
level=Config.LOG_LEVEL
)
logger = logging.getLogger(__name__)
def safe_text(x):
if not x:
return ""
return str(x)
# ======================
# STATES
# ======================
SEARCH_VN, SELECT_VN, VN_DETAILS = range(3)
SEARCH_CHARACTER, SELECT_CHARACTER = range(2)
SEARCH_RELEASE, SELECT_RELEASE = range(2)
SEARCH_STAFF, SELECT_STAFF = range(2)
@dp.message(Command("start"))
async def start(msg: types.Message):
await msg.answer(
"VNDB Bot ready.\n"
"Commands:\n"
"/search <vn>\n"
"/vn <id>\n"
"/char <id>\n"
"/release <id>"
)
# ======================
# CLIENT
# ======================
vndb_client = VndbClient(use_sandbox=Config.USE_SANDBOX)
@dp.message(Command("help"))
async def help(msg: types.Message):
await start(msg)
# ======================
# HELPERS
# ======================
HTML_HELP_HEADER = "<b>VNDB Telegram Bot</b>\n\n"
@dp.message(Command("search"))
async def search(msg: types.Message):
query = msg.text.replace("/search", "").strip()
if not query:
return await msg.answer("Empty query")
result = await vndb.search_vn(query)
# ======================
# HANDLERS
# ======================
class BotHandlers:
if not result:
return await msg.answer("Not found")
@staticmethod
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
text = """
<b>Добро пожаловать в VNDB Telegram Бот!</b>
text = "\n\n".join(
f"{i['id']}{i.get('title') or i.get('name')}"
for i in result
)
await msg.answer(text[:4000])
Этот бот позволяет искать визуальные новеллы, персонажей и релизы из VNDB.
<b>Команды:</b>
/search - поиск VN
/char - персонажи
/release - релизы
/staff - сотрудники
/producer - продюсеры
/tag - теги
/trait - черты
/quote - цитаты
/help - помощь
"""
@dp.message(Command("vn"))
async def vn(msg: types.Message):
vid = msg.text.replace("/vn", "").strip()
if not vid:
return await msg.answer("No ID")
await update.message.reply_text(text, parse_mode="HTML")
data = await vndb.get_vn(vid)
if not data:
return await msg.answer("Not found")
# ======================
# HELP
# ======================
@staticmethod
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
text = """
<b>Справка по командам</b>
await msg.answer(
f"{safe_text(data.get('title'))}\n"
f"Original: {safe_text(data.get('original'))}\n"
f"Released: {safe_text(data.get('released'))}\n"
f"Rating: {safe_text(data.get('rating'))}\n"
f"Votes: {safe_text(data.get('votecount'))}"
)
<b>Поиск:</b>
/search &lt;название&gt;
/char &lt;имя&gt;
/release &lt;название&gt;
/staff &lt;имя&gt;
/producer &lt;имя&gt;
<b>Информация:</b>
/tag - популярные теги
/trait - черты
/quote &lt;число&gt; - цитаты
/stats - статистика
/schema - API
@dp.message(Command("char"))
async def char(msg: types.Message):
cid = msg.text.replace("/char", "").strip()
if not cid:
return await msg.answer("No ID")
<b>Детальный просмотр:</b>
/vn_detail &lt;id&gt;
/char_detail &lt;id&gt;
/release_detail &lt;id&gt;
data = await vndb.get_char(cid)
if not data:
return await msg.answer("Not found")
<b>Пример:</b>
/search Steins Gate
/char Okabe
/vn_detail v17
await msg.answer(
f"{safe_text(data.get('name'))}\n"
f"Original: {safe_text(data.get('original'))}"
)
<b>Ссылка:</b>
<a href="https://git.kotac.ru/King-of-the-all-Cookies/ayako/src/branch/main/EXAMPLES.md">
Примеры команд
</a>
"""
await update.message.reply_text(text, parse_mode="HTML")
@dp.message(Command("release"))
async def release(msg: types.Message):
rid = msg.text.replace("/release", "").strip()
if not rid:
return await msg.answer("No ID")
# ======================
# STATS
# ======================
@staticmethod
async def stats(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
try:
stats = await vndb_client.get_stats()
data = await vndb.get_release(rid)
if not data:
return await msg.answer("Not found")
text = f"""
<b>Статистика VNDB</b>
await msg.answer(
f"{safe_text(data.get('title'))}\n"
f"Released: {safe_text(data.get('released'))}"
)
VN: {stats.get('vn', 0)}
Characters: {stats.get('chars', 0)}
Releases: {stats.get('releases', 0)}
Producers: {stats.get('producers', 0)}
Staff: {stats.get('staff', 0)}
"""
await update.message.reply_text(text, parse_mode="HTML")
except Exception as e:
logger.error(e)
await update.message.reply_text(f"{ErrorHandler.format_error(e)}")
# ======================
# SCHEMA
# ======================
@staticmethod
async def schema(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
try:
await update.message.reply_text("⏳ Загружаю схему...", parse_mode="HTML")
schema = await vndb_client.get_schema()
text = "<b>API Schema VNDB</b>\n\n"
if "db_types" in schema:
text += "<b>Типы:</b>\n"
for k in list(schema["db_types"].keys())[:5]:
text += f"{k}\n"
text += "\n<a href='https://api.vndb.org/kana'>Документация API</a>"
await update.message.reply_text(text, parse_mode="HTML")
except Exception as e:
logger.error(e)
await update.message.reply_text(f"{ErrorHandler.format_error(e)}")
# ======================
# SEARCH VN
# ======================
@staticmethod
async def search_vn(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
query = " ".join(context.args)
if not query:
await update.message.reply_text("Введите название")
return ConversationHandler.END
await update.message.reply_text(
f"<b>Поиск:</b> {query}",
parse_mode="HTML"
)
results = await vndb_client.query_vn(
filters=["search", "=", query.strip()],
fields=["id", "title"],
results=10
)
if not results.get("results"):
await update.message.reply_text("Ничего не найдено")
return ConversationHandler.END
text = "<b>Результаты:</b>\n\n"
for vn in results["results"]:
text += f"{vn.get('id')} - {vn.get('title')}\n"
await update.message.reply_text(text, parse_mode="HTML")
# images
for vn in results["results"]:
text += f"{vn.get('id')} - {vn.get('title')}\n"
# 👇 ДОБАВЬ ССЫЛКУ НА VNDB
text += f"https://vndb.org/{vn.get('id')}\n\n"
except Exception as e:
logger.error(e)
await update.message.reply_text(f"{ErrorHandler.format_error(e)}")
return ConversationHandler.END
# ======================
# CHARACTER
# ======================
@staticmethod
async def search_character(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
query = " ".join(context.args)
if not query:
await update.message.reply_text("Введите имя")
return ConversationHandler.END
results = await vndb_client.query_character(
filters=["search", "=", query.strip()],
fields=["id", "name", "original"],
results=10
)
text = "<b>Персонажи:</b>\n\n"
for c in results.get("results", []):
text += f"{c.get('id')} - {c.get('name')} ({c.get('original', '')})\n"
await update.message.reply_text(text, parse_mode="HTML")
except Exception as e:
logger.error(e)
await update.message.reply_text(f"{ErrorHandler.format_error(e)}")
return ConversationHandler.END
# ======================
# RELEASE
# ======================
@staticmethod
async def search_release(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
query = " ".join(context.args)
results = await vndb_client.query_release(
filters=["search", "=", query.strip()], # ✔️ важно
fields=["id", "title"],
results=10
)
text = "<b>Релизы:</b>\n\n"
for r in results.get("results", []):
text += f"{r.get('id')} - {r.get('title')}\n"
await update.message.reply_text(text, parse_mode="HTML")
except Exception as e:
logger.error(e)
await update.message.reply_text(f"{ErrorHandler.format_error(e)}")
return ConversationHandler.END
# ======================
# MAIN
# ======================
def main():
Config.validate()
app = Application.builder().token(Config.TELEGRAM_BOT_TOKEN).build()
app.add_handler(CommandHandler("start", BotHandlers.start))
app.add_handler(CommandHandler("help", BotHandlers.help_command))
app.add_handler(CommandHandler("stats", BotHandlers.stats))
app.add_handler(CommandHandler("schema", BotHandlers.schema))
app.add_handler(CommandHandler("search", BotHandlers.search_vn))
app.add_handler(CommandHandler("char", BotHandlers.search_character))
app.add_handler(CommandHandler("release", BotHandlers.search_release))
for h in get_detail_handlers():
app.add_handler(h)
logger.info("Bot started")
app.run_polling()
async def main():
await dp.start_polling(bot)
if __name__ == "__main__":
main()
asyncio.run(main())