main.py - master - Kit_Michael/RusNaturalControl

main.py
626 строк25 KB
mrkit
v0.1.0
14 фев 2026, 05:13
14 фев 2026, 05:13df56670
import json
import logging
import os
import random
from dataclasses import dataclass
from datetime import datetime, time, timezone
from html import escape
from logging.handlers import RotatingFileHandler
from pathlib import Path
from typing import Any

from dotenv import load_dotenv
from telegram import Chat, Update
from telegram.ext import (
    Application,
    ChatMemberHandler,
    CommandHandler,
    ContextTypes,
    MessageHandler,
    filters,
)


def setup_logging() -> None:
    # Настраиваем логирование в консоль и в ротационный файл.
    logs_dir = Path("logs")
    logs_dir.mkdir(parents=True, exist_ok=True)

    formatter = logging.Formatter(
        fmt="%(asctime)s | %(levelname)s | %(name)s | %(message)s"
    )

    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)

    file_handler = RotatingFileHandler(
        filename=logs_dir / "bot.log",
        maxBytes=5 * 1024 * 1024,
        backupCount=5,
        encoding="utf-8",
    )
    file_handler.setFormatter(formatter)

    root_logger = logging.getLogger()
    root_logger.setLevel(logging.INFO)
    root_logger.handlers.clear()
    root_logger.addHandler(console_handler)
    root_logger.addHandler(file_handler)


setup_logging()
logger = logging.getLogger(__name__)


@dataclass
class Config:
    # Основной токен Telegram-бота.
    bot_token: str
    # Время ежедневного поста в UTC.
    daily_time_utc: time
    # Необязательный прокси для доступа к Telegram API.
    telegram_proxy_url: str | None
    # Пути к контентным JSON-файлам.
    answers_path: Path
    jokes_path: Path
    punchlines_path: Path
    gay_phrases_path: Path
    # Каталог с GIF/анимациями.
    gifs_dir: Path
    # Файл, где храним участников по чатам.
    participants_store_path: Path
    # Файл, где храним список чатов, где состоит бот.
    chats_store_path: Path


class ParticipantStore:
    """Хранилище участников по чатам, которых бот увидел в активности."""

    def __init__(self, storage_path: Path):
        self.storage_path = storage_path
        self._participants: dict[str, dict[str, dict[str, Any]]] = {}
        self._load()

    def _load(self) -> None:
        # Загружаем участников из файла, если он есть.
        if not self.storage_path.exists():
            self._participants = {}
            return

        try:
            raw = json.loads(self.storage_path.read_text(encoding="utf-8"))
            if isinstance(raw, dict):
                parsed: dict[str, dict[str, dict[str, Any]]] = {}
                for chat_id, users in raw.items():
                    if not isinstance(users, dict):
                        continue
                    parsed[str(chat_id)] = {}
                    for user_id, item in users.items():
                        if isinstance(item, dict) and "id" in item:
                            parsed[str(chat_id)][str(user_id)] = item
                self._participants = parsed
            elif isinstance(raw, list):
                # Обратная совместимость со старым форматом (один общий список).
                self._participants = {"global": {str(item["id"]): item for item in raw if "id" in item}}
            else:
                self._participants = {}
        except (json.JSONDecodeError, OSError, TypeError, KeyError):
            logger.warning("Не удалось загрузить список участников из %s", self.storage_path)
            self._participants = {}

    def _save(self) -> None:
        # Сохраняем участников в JSON для устойчивости между перезапусками.
        try:
            self.storage_path.parent.mkdir(parents=True, exist_ok=True)
            payload = self._participants
            self.storage_path.write_text(
                json.dumps(payload, ensure_ascii=False, indent=2),
                encoding="utf-8",
            )
        except OSError:
            logger.exception("Не удалось сохранить участников в %s", self.storage_path)

    def upsert(
        self,
        chat_id: int,
        user_id: int,
        username: str | None,
        first_name: str,
        last_name: str | None,
    ) -> None:
        # Обновляем запись о пользователе или создаем новую.
        chat_key = str(chat_id)
        key = str(user_id)
        if chat_key not in self._participants:
            self._participants[chat_key] = {}
        self._participants[chat_key][key] = {
            "id": user_id,
            "username": username,
            "first_name": first_name,
            "last_name": last_name,
            "last_seen_utc": datetime.now(timezone.utc).isoformat(),
        }
        self._save()

    def all(self, chat_id: int) -> list[dict[str, Any]]:
        # Возвращаем список участников только для указанного чата.
        return list(self._participants.get(str(chat_id), {}).values())


class ChatStore:
    """Хранилище чатов, где бот должен публиковать ежедневный пост."""

    def __init__(self, storage_path: Path):
        self.storage_path = storage_path
        self._chats: dict[str, dict[str, Any]] = {}
        self._load()

    def _load(self) -> None:
        if not self.storage_path.exists():
            self._chats = {}
            return
        try:
            raw = json.loads(self.storage_path.read_text(encoding="utf-8"))
            if isinstance(raw, list):
                self._chats = {str(item["chat_id"]): item for item in raw if isinstance(item, dict) and "chat_id" in item}
            elif isinstance(raw, dict):
                self._chats = {str(k): v for k, v in raw.items() if isinstance(v, dict)}
            else:
                self._chats = {}
        except (json.JSONDecodeError, OSError, TypeError, KeyError):
            logger.warning("Не удалось загрузить список чатов из %s", self.storage_path)
            self._chats = {}

    def _save(self) -> None:
        try:
            self.storage_path.parent.mkdir(parents=True, exist_ok=True)
            self.storage_path.write_text(
                json.dumps(list(self._chats.values()), ensure_ascii=False, indent=2),
                encoding="utf-8",
            )
        except OSError:
            logger.exception("Не удалось сохранить список чатов в %s", self.storage_path)

    def upsert(self, chat_id: int, title: str | None, chat_type: str) -> None:
        key = str(chat_id)
        self._chats[key] = {
            "chat_id": chat_id,
            "title": title or "",
            "chat_type": chat_type,
            "updated_at_utc": datetime.now(timezone.utc).isoformat(),
        }
        self._save()

    def remove(self, chat_id: int) -> None:
        key = str(chat_id)
        if key in self._chats:
            del self._chats[key]
            self._save()

    def all_chat_ids(self) -> list[int]:
        return [int(item["chat_id"]) for item in self._chats.values()]


def parse_utc_time(value: str) -> time:
    # Парсим формат HH:MM (или HH:MM:SS) и принудительно ставим UTC.
    chunks = value.strip().split(":")
    if len(chunks) not in (2, 3):
        raise ValueError("DAILY_POST_TIME_UTC должен быть в формате HH:MM или HH:MM:SS")

    hour = int(chunks[0])
    minute = int(chunks[1])
    second = int(chunks[2]) if len(chunks) == 3 else 0
    return time(hour=hour, minute=minute, second=second, tzinfo=timezone.utc)


def load_config() -> Config:
    # Подгружаем .env из корня проекта.
    # Используем utf-8-sig, чтобы корректно читать файл даже с BOM.
    env_path = Path(__file__).resolve().parent / ".env"
    load_dotenv(dotenv_path=env_path, encoding="utf-8-sig")

    bot_token = os.getenv("BOT_TOKEN", "").strip()
    if not bot_token:
        raise ValueError("Не задан BOT_TOKEN в .env")

    daily_time_raw = os.getenv("DAILY_POST_TIME_UTC", "09:00")
    daily_time_utc = parse_utc_time(daily_time_raw)
    telegram_proxy_url = os.getenv("TELEGRAM_PROXY_URL", "").strip() or None

    return Config(
        bot_token=bot_token,
        daily_time_utc=daily_time_utc,
        telegram_proxy_url=telegram_proxy_url,
        answers_path=Path(os.getenv("ANSWERS_PATH", "content/answers.json")),
        jokes_path=Path(os.getenv("JOKES_PATH", "content/jokes.json")),
        punchlines_path=Path(os.getenv("PUNCHLINES_PATH", "content/punchlines.json")),
        gay_phrases_path=Path(os.getenv("GAY_PHRASES_PATH", "content/gay.json")),
        gifs_dir=Path(os.getenv("GIFS_DIR", "content/gifs")),
        participants_store_path=Path(
            os.getenv("PARTICIPANTS_STORE_PATH", "data/participants.json")
        ),
        chats_store_path=Path(
            os.getenv("CHATS_STORE_PATH", "data/chats.json")
        ),
    )


def load_text_list(path: Path, field: str) -> list[str]:
    # Читаем массив объектов вида [{field: "..."}] и вытаскиваем строки.
    if not path.exists():
        logger.warning("Файл %s не найден", path)
        return []

    raw: Any = None
    last_error: Exception | None = None
    # Пробуем несколько кодировок, чтобы пережить BOM и ручные правки файлов.
    for enc in ("utf-8-sig", "utf-8", "cp1251"):
        try:
            raw = json.loads(path.read_text(encoding=enc))
            break
        except (json.JSONDecodeError, OSError) as exc:
            last_error = exc
            raw = None

    if raw is None:
        logger.warning("Не удалось прочитать JSON из %s: %s", path, last_error)
        return []

    if not isinstance(raw, list):
        logger.warning("Ожидался список в %s", path)
        return []

    values: list[str] = []
    for item in raw:
        if isinstance(item, dict):
            value = item.get(field)
            if isinstance(value, str) and value.strip():
                values.append(value.strip())
    return values


def render_user_mention(member: dict[str, Any]) -> str:
    # Если есть username, упоминаем через @username; иначе делаем HTML-ссылку по ID.
    username = member.get("username")
    if isinstance(username, str) and username:
        return f"@{escape(username)}"

    first_name = str(member.get("first_name") or "Участник")
    user_id = int(member["id"])
    safe_name = escape(first_name)
    return f'<a href="tg://user?id={user_id}">{safe_name}</a>'


def pick_random_media(config: Config) -> tuple[str, Path] | None:
    # Собираем только GIF/анимации.
    gifs_ext = {".gif", ".mp4"}

    media_candidates: list[tuple[str, Path]] = []

    if config.gifs_dir.exists():
        media_candidates.extend(
            ("animation", p) for p in config.gifs_dir.iterdir() if p.is_file() and p.suffix.lower() in gifs_ext
        )

    if not media_candidates:
        return None

    # Если гифки есть, всегда выбираем случайную.
    return random.choice(media_candidates)


def build_daily_message(
    participants: list[dict[str, Any]],
    jokes: list[str],
    punchlines: list[str],
    gay_phrases: list[str],
) -> tuple[str, list[tuple[dict[str, Any], int]], dict[str, Any] | None]:
    # Формируем текст: шутка + список участников с процентом + финальный панч.
    if not participants:
        return (
            "Сегодня никого не удалось поймать в статистику. Завтра попробуем снова.",
            [],
            None,
        )

    joke = random.choice(jokes) if jokes else "Ежедневный розыгрыш начинается!"

    sample_size = random.randint(1, min(5, len(participants)))
    selected = random.sample(participants, sample_size)

    scored: list[tuple[dict[str, Any], int]] = []
    for member in selected:
        percent = random.randint(1, 100)
        scored.append((member, percent))

    winner, _ = max(scored, key=lambda row: row[1])
    punchline = random.choice(punchlines) if punchlines else "сегодня ты забрал топ!"

    lines = [escape(joke), ""]
    for member, percent in scored:
        mention = render_user_mention(member)
        # Формулировка перед процентом берется случайно из content/gay.json.
        gay_prefix = random.choice(gay_phrases) if gay_phrases else "на приколе"
        lines.append(f"{mention} {escape(gay_prefix)} {percent}%")

    lines.append("")
    winner_mention = render_user_mention(winner)
    lines.append(f"{winner_mention}, {escape(punchline)}")

    return "\n".join(lines), scored, winner


async def send_daily_post(
    app: Application,
    chat_id: int,
    config: Config,
    store: ParticipantStore,
    jokes: list[str],
    punchlines: list[str],
    gay_phrases: list[str],
) -> None:
    # Готовим и отправляем ежедневный пост в заданное сообщество.
    try:
        participants = store.all(chat_id)
        message_text, _, _ = build_daily_message(participants, jokes, punchlines, gay_phrases)
        media = pick_random_media(config)

        if media is None:
            await app.bot.send_message(chat_id=chat_id, text=message_text, parse_mode="HTML")
            return

        media_type, path = media
        with path.open("rb") as media_file:
            if media_type == "animation":
                await app.bot.send_animation(
                    chat_id=chat_id,
                    animation=media_file,
                    caption=message_text,
                    parse_mode="HTML",
                )
    except Exception:
        logger.exception("Ошибка при отправке поста в чат %s", chat_id)
        raise


async def scheduled_daily_post(context: ContextTypes.DEFAULT_TYPE) -> None:
    # Колбэк ежедневного задания из JobQueue.
    app = context.application
    cfg: Config = app.bot_data["config"]
    chat_store: ChatStore = app.bot_data["chat_store"]
    store: ParticipantStore = app.bot_data["participant_store"]
    jokes: list[str] = app.bot_data["jokes"]
    punchlines: list[str] = app.bot_data["punchlines"]
    gay_phrases: list[str] = app.bot_data["gay_phrases"]

    chat_ids = chat_store.all_chat_ids()
    if not chat_ids:
        logger.info("Ежедневный пост пропущен: бот пока не состоит ни в одном чате")
        return

    for chat_id in chat_ids:
        try:
            await send_daily_post(
                app=app,
                chat_id=chat_id,
                config=cfg,
                store=store,
                jokes=jokes,
                punchlines=punchlines,
                gay_phrases=gay_phrases,
            )
        except Exception as exc:
            logger.warning("Не удалось отправить ежедневный пост в чат %s: %s", chat_id, exc)


async def handle_private(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    # В личке бот не работает по функционалу, только шутливо отвечает.
    try:
        if not update.effective_chat or update.effective_chat.type != Chat.PRIVATE:
            return

        answers: list[str] = context.application.bot_data["answers"]
        response = random.choice(answers) if answers else "Пиши в сообщество, тут я бездельничаю."

        if update.effective_message:
            await update.effective_message.reply_text(response)
    except Exception:
        logger.exception("Ошибка в обработчике личного сообщения")


async def handle_group_activity(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    # Любая активность в группе помогает пополнять список участников для ежедневного выбора.
    try:
        chat = update.effective_chat
        user = update.effective_user
        if not chat or chat.type not in (Chat.GROUP, Chat.SUPERGROUP) or not user:
            return

        chat_store: ChatStore = context.application.bot_data["chat_store"]
        chat_store.upsert(chat_id=chat.id, title=chat.title, chat_type=chat.type)

        store: ParticipantStore = context.application.bot_data["participant_store"]
        store.upsert(
            chat_id=chat.id,
            user_id=user.id,
            username=user.username,
            first_name=user.first_name,
            last_name=user.last_name,
        )
    except Exception:
        logger.exception("Ошибка в обработчике активности группы")


async def handle_chat_member(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    # Отслеживаем изменения участников (вступил/вернулся), чтобы расширять базу пользователей.
    try:
        if not update.chat_member:
            return

        chat = update.chat_member.chat
        user = update.chat_member.new_chat_member.user

        if chat.type not in (Chat.GROUP, Chat.SUPERGROUP):
            return

        chat_store: ChatStore = context.application.bot_data["chat_store"]
        chat_store.upsert(chat_id=chat.id, title=chat.title, chat_type=chat.type)

        store: ParticipantStore = context.application.bot_data["participant_store"]
        store.upsert(
            chat_id=chat.id,
            user_id=user.id,
            username=user.username,
            first_name=user.first_name,
            last_name=user.last_name,
        )
    except Exception:
        logger.exception("Ошибка в обработчике chat_member")


async def handle_my_chat_member(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    # Срабатывает, когда меняется статус самого бота в чате.
    # Удобно для получения chat_id без команд и без конфликтов с другими ботами.
    try:
        if not update.my_chat_member:
            return

        chat = update.my_chat_member.chat
        new_member = update.my_chat_member.new_chat_member
        status = new_member.status

        if chat.type not in (Chat.GROUP, Chat.SUPERGROUP):
            return

        chat_store: ChatStore = context.application.bot_data["chat_store"]

        if status in ("member", "administrator"):
            chat_store.upsert(chat_id=chat.id, title=chat.title, chat_type=chat.type)
            logger.info("Бот добавлен в чат: id=%s type=%s title=%s", chat.id, chat.type, chat.title)
            try:
                await context.bot.send_message(
                    chat_id=chat.id,
                    text=f"chat_id: {chat.id}\nchat_type: {chat.type}",
                )
            except Exception as exc:
                logger.warning("Не удалось отправить chat_id в чат %s: %s", chat.id, exc)
        elif status in ("left", "kicked"):
            chat_store.remove(chat.id)
            logger.info("Бот удален из чата: id=%s type=%s title=%s", chat.id, chat.type, chat.title)
    except Exception:
        logger.exception("Ошибка в обработчике my_chat_member")


async def force_daily_post(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    # Команда для ручного теста ежедневного поста в группе.
    try:
        chat = update.effective_chat
        if not chat or chat.type not in (Chat.GROUP, Chat.SUPERGROUP):
            return

        chat_store: ChatStore = context.application.bot_data["chat_store"]
        chat_store.upsert(chat_id=chat.id, title=chat.title, chat_type=chat.type)

        cfg: Config = context.application.bot_data["config"]
        store: ParticipantStore = context.application.bot_data["participant_store"]
        jokes: list[str] = context.application.bot_data["jokes"]
        punchlines: list[str] = context.application.bot_data["punchlines"]
        gay_phrases: list[str] = context.application.bot_data["gay_phrases"]

        await send_daily_post(
            app=context.application,
            chat_id=chat.id,
            config=cfg,
            store=store,
            jokes=jokes,
            punchlines=punchlines,
            gay_phrases=gay_phrases,
        )
    except Exception:
        logger.exception("Ошибка выполнения /force_daily")


async def handle_error(update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
    # Глобальный обработчик непойманных исключений Telegram-обработчиков.
    logger.exception("Непойманная ошибка в update=%s", update, exc_info=context.error)


def main() -> None:
    # Инициализация конфигурации, контента и запуск polling-бота.
    config = load_config()

    answers = load_text_list(config.answers_path, "answer")
    jokes = load_text_list(config.jokes_path, "joke")
    punchlines = load_text_list(config.punchlines_path, "punchline")
    gay_phrases = load_text_list(config.gay_phrases_path, "gay")

    participant_store = ParticipantStore(config.participants_store_path)
    chat_store = ChatStore(config.chats_store_path)

    builder = Application.builder().token(config.bot_token)
    # Явно используем HTTP/1.1 для совместимости с сетями, где HTTP/2 ломается.
    builder = builder.http_version("1.1").get_updates_http_version("1.1")
    # Увеличиваем сетевые таймауты для нестабильных/медленных прокси.
    builder = (
        builder
        .connect_timeout(20.0)
        .read_timeout(30.0)
        .write_timeout(30.0)
        .pool_timeout(30.0)
        .get_updates_connect_timeout(20.0)
        .get_updates_read_timeout(60.0)
        .get_updates_write_timeout(30.0)
        .get_updates_pool_timeout(30.0)
    )
    # Если задан прокси, используем его для всех запросов к Telegram API.
    if config.telegram_proxy_url:
        builder = builder.proxy(config.telegram_proxy_url).get_updates_proxy(config.telegram_proxy_url)

    app = builder.build()
    app.bot_data["config"] = config
    app.bot_data["answers"] = answers
    app.bot_data["jokes"] = jokes
    app.bot_data["punchlines"] = punchlines
    app.bot_data["gay_phrases"] = gay_phrases
    app.bot_data["participant_store"] = participant_store
    app.bot_data["chat_store"] = chat_store

    # Личные сообщения: бот отшучивается.
    app.add_handler(MessageHandler(filters.ChatType.PRIVATE & filters.ALL, handle_private))

    # Групповые сообщения: только мониторинг участников.
    app.add_handler(MessageHandler(filters.ChatType.GROUPS & filters.ALL, handle_group_activity))

    # События по участникам (вступления/изменения статуса) тоже используем для мониторинга.
    app.add_handler(ChatMemberHandler(handle_chat_member, ChatMemberHandler.CHAT_MEMBER))
    # События статуса самого бота в чате (добавили/вернули) используем для авто-выдачи chat_id.
    app.add_handler(ChatMemberHandler(handle_my_chat_member, ChatMemberHandler.MY_CHAT_MEMBER))

    # Команда для ручной проверки ежедневного поста.
    app.add_handler(CommandHandler("force_daily", force_daily_post))
    # Глобальный перехват исключений из обработчиков.
    app.add_error_handler(handle_error)

    # Ежедневное задание выполняется строго в UTC-времени из env.
    if app.job_queue is None:
        raise RuntimeError(
            "JobQueue не инициализирован. Установите зависимости: "
            "pip install \"python-telegram-bot[job-queue]\""
        )

    app.job_queue.run_daily(
        callback=scheduled_daily_post,
        time=config.daily_time_utc,
        name="daily_fun_post",
    )

    logger.info("Бот запущен. Ежедневный пост в UTC: %s", config.daily_time_utc)
    # При нестабильной сети/прокси не выходим после первой ошибки подключения.
    app.run_polling(
        allowed_updates=Update.ALL_TYPES,
        bootstrap_retries=-1,
    )


if __name__ == "__main__":
    main()