re-xxm/reaction_board/notify/telegram.py

# -*- coding: utf-8 -*-
"""Telegram Bot 通知与回调。"""

from __future__ import annotations

import html
import json
import urllib.parse
import urllib.request
from typing import Any, Mapping, Protocol, Sequence

MAX_CANDIDATES_PER_MESSAGE = 3


class TelegramError(RuntimeError):
    pass


class TelegramTransport(Protocol):
    def post(self, url: str, data: Mapping[str, Any], timeout: float = 30) -> Any: ...


class UrllibTelegramTransport:
    def post(self, url: str, data: Mapping[str, Any], timeout: float = 30) -> Any:
        request = urllib.request.Request(
            url,
            data=urllib.parse.urlencode(data).encode("utf-8"),
            method="POST",
        )
        with urllib.request.urlopen(request, timeout=timeout) as response:
            return json.loads(response.read().decode("utf-8", "replace"))


def format_duration(seconds: int) -> str:
    seconds = max(0, int(seconds or 0))
    hours, remainder = divmod(seconds, 3600)
    minutes, secs = divmod(remainder, 60)
    if hours:
        return f"{hours}:{minutes:02d}:{secs:02d}"
    return f"{minutes}:{secs:02d}"


def format_candidate(candidate: Mapping[str, Any]) -> str:
    title = html.escape(str(candidate.get("title") or "(无标题)"))
    up_name = html.escape(str(candidate.get("up_name") or "未知UP"))
    duration = format_duration(int(candidate.get("duration") or 0))
    play = int(candidate.get("play") or 0)
    like = int(candidate.get("like") or 0)
    return (
        f"🎬 <b>{title}</b>\n"
        f"UP: {up_name}\n"
        f"时长 {duration} · 播放 {play} · 点赞 {like}（索引值）"
    )


class TelegramClient:
    def __init__(
        self,
        token: str,
        chat_id: int,
        *,
        transport: TelegramTransport | None = None,
        timeout: float = 30,
        poll_timeout: int = 25,
        logger=None,
    ) -> None:
        if not token:
            raise ValueError("telegram.bot_token 不能为空")
        if not chat_id:
            raise ValueError("telegram.chat_id 不能为空")
        self.token = token
        self.chat_id = int(chat_id)
        self.transport = transport or UrllibTelegramTransport()
        self.timeout = timeout
        self.poll_timeout = poll_timeout
        self.logger = logger

    def _call(self, method: str, params: Mapping[str, Any] | None = None) -> Any:
        url = f"https://api.telegram.org/bot{self.token}/{method}"
        payload = self.transport.post(url, params or {}, timeout=self.timeout)
        if not isinstance(payload, Mapping):
            raise TelegramError("Telegram 响应不是对象")
        if not payload.get("ok"):
            raise TelegramError(str(payload.get("description") or "Telegram API 错误"))
        return payload.get("result")

    def send_text(
        self, text: str, *, reply_markup: Mapping[str, Any] | None = None
    ) -> Any:
        params: dict[str, Any] = {
            "chat_id": self.chat_id,
            "text": text,
            "parse_mode": "HTML",
            "disable_web_page_preview": True,
        }
        if reply_markup is not None:
            params["reply_markup"] = json.dumps(reply_markup, ensure_ascii=False)
        return self._call("sendMessage", params)

    def send_candidates(self, candidates: Sequence[Mapping[str, Any]]) -> bool:
        # 通知器边界再做一次 aid 去重，避免调用方重复入队导致消息主体和按钮重复。
        items: list[Mapping[str, Any]] = []
        seen: set[int] = set()
        for candidate in candidates:
            aid = candidate.get("aid")
            if aid in seen:
                continue
            seen.add(aid)
            items.append(candidate)
        if not items:
            return True
        for index in range(0, len(items), MAX_CANDIDATES_PER_MESSAGE):
            chunk = items[index:index + MAX_CANDIDATES_PER_MESSAGE]
            text = "\n\n".join(format_candidate(candidate) for candidate in chunk)
            keyboard = {
                "inline_keyboard": [
                    [
                        {"text": "⭐ 收藏", "callback_data": f"fav:{candidate['aid']}"},
                        {"text": "🙈 忽略", "callback_data": f"ign:{candidate['aid']}"},
                        {"text": "🚫 拉黑", "callback_data": f"blk:{candidate['aid']}"},
                    ]
                    for candidate in chunk
                ]
            }
            self.send_text(text, reply_markup=keyboard)
        return True

    def get_updates(
        self, offset: int | None = None, *, timeout: int | None = None
    ) -> list[dict[str, Any]]:
        params: dict[str, Any] = {
            "timeout": self.poll_timeout if timeout is None else int(timeout),
            "allowed_updates": json.dumps(["callback_query"]),
        }
        if offset is not None:
            params["offset"] = int(offset)
        result = self._call("getUpdates", params)
        return result if isinstance(result, list) else []

    def answer_callback(self, callback_id: str, text: str = "") -> Any:
        return self._call(
            "answerCallbackQuery",
            {"callback_query_id": callback_id, "text": text[:200]},
        )