102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
# bot/bot_factory.py
|
||
import logging
|
||
from typing import Optional, Sequence, Tuple
|
||
|
||
from telegram.constants import ParseMode
|
||
from telegram.ext import (
|
||
Application,
|
||
ApplicationBuilder,
|
||
CommandHandler,
|
||
MessageHandler,
|
||
ChatMemberHandler,
|
||
Defaults,
|
||
filters,
|
||
)
|
||
|
||
from .models import TelegramBot, BotConfig
|
||
from .handlers import start, ping, echo, my_chat_member_update, error_handler
|
||
from .services import build_services
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _resolve_parse_mode(pm: Optional[str]):
|
||
if not pm or pm == "None":
|
||
return None
|
||
return getattr(ParseMode, pm, ParseMode.HTML)
|
||
|
||
|
||
def build_application_for_bot(tb: TelegramBot, cfg: BotConfig) -> Tuple[Application, Optional[list]]:
|
||
"""
|
||
Фабрика PTB Application для КОНКРЕТНОГО бота.
|
||
ВАЖНО: никакого ORM внутри — cfg передаётся извне.
|
||
"""
|
||
parse_mode = _resolve_parse_mode(cfg.parse_mode)
|
||
defaults = Defaults(parse_mode=parse_mode) if parse_mode else None
|
||
allowed_updates = cfg.allowed_updates or None
|
||
|
||
builder: ApplicationBuilder = Application.builder().token(tb.token)
|
||
if defaults:
|
||
builder = builder.defaults(defaults)
|
||
|
||
app = builder.build()
|
||
|
||
# Хендлеры
|
||
app.add_handler(CommandHandler("start", start))
|
||
app.add_handler(CommandHandler("ping", ping))
|
||
app.add_handler(ChatMemberHandler(my_chat_member_update, ChatMemberHandler.MY_CHAT_MEMBER))
|
||
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))
|
||
|
||
# Глобальная обработка ошибок
|
||
app.add_error_handler(error_handler)
|
||
|
||
# DI: сервисы для этого бота
|
||
services = build_services(tb)
|
||
app.bot_data["services"] = services
|
||
app.bot_data["allowed_updates"] = allowed_updates
|
||
|
||
logger.info(
|
||
"Built app for bot=%s (@%s), parse_mode=%s, allowed_updates=%s",
|
||
tb.name, tb.username or "—", parse_mode, allowed_updates
|
||
)
|
||
return app, allowed_updates
|
||
|
||
|
||
def get_first_active_bot_with_config() -> Tuple[TelegramBot, BotConfig]:
|
||
"""
|
||
СИНХРОННО! Забирает первого активного бота и его конфиг.
|
||
"""
|
||
tb = TelegramBot.objects.filter(is_active=True).first()
|
||
if not tb:
|
||
raise RuntimeError("Нет активного TelegramBot (is_active=True). Создайте запись и включите её.")
|
||
cfg = BotConfig.objects.filter(bot=tb).first()
|
||
if not cfg:
|
||
raise RuntimeError(f"Для бота '{tb}' не найден BotConfig. Создайте связанную запись BotConfig.")
|
||
return tb, cfg
|
||
|
||
|
||
def get_all_active_bots_with_configs() -> Sequence[Tuple[TelegramBot, BotConfig]]:
|
||
"""
|
||
СИНХРОННО! Возвращает все (bot, cfg) для активных ботов.
|
||
Бросает, если хотя бы у одного нет BotConfig.
|
||
"""
|
||
bots = list(TelegramBot.objects.filter(is_active=True))
|
||
if not bots:
|
||
return []
|
||
|
||
cfg_map = {c.bot_id: c for c in BotConfig.objects.filter(bot__in=bots)}
|
||
result: list[Tuple[TelegramBot, BotConfig]] = []
|
||
missing = []
|
||
for b in bots:
|
||
cfg = cfg_map.get(b.id)
|
||
if not cfg:
|
||
missing.append(b)
|
||
else:
|
||
result.append((b, cfg))
|
||
|
||
if missing:
|
||
names = ", ".join(f"{b.name}(@{b.username or '—'})" for b in missing)
|
||
raise RuntimeError(f"Нет BotConfig для ботов: {names}")
|
||
|
||
return result
|