TelegramBotAdmin/wbb/services/giveaway_eligibility.py

from __future__ import annotations

from typing import Any

from pyrogram.enums import ChatMemberStatus

from wbb import BOT_ID, app


class EligibilityUnavailable(RuntimeError):
    """Telegram cannot give a trustworthy membership decision."""


def is_member(member: Any) -> bool:
    if member.status in {ChatMemberStatus.LEFT, ChatMemberStatus.BANNED}:
        return False
    if member.status == ChatMemberStatus.RESTRICTED:
        return bool(getattr(member, "is_member", False))
    return member.status in {
        ChatMemberStatus.OWNER,
        ChatMemberStatus.ADMINISTRATOR,
        ChatMemberStatus.MEMBER,
    }


async def validate_targets(targets: list[dict[str, Any]]) -> list[dict[str, Any]]:
    if len(targets) > 10:
        raise ValueError("最多设置 10 个资格频道或群。")
    normalized: list[dict[str, Any]] = []
    seen: set[int] = set()
    for target in targets:
        try:
            chat_id = int(target["chat_id"])
        except (KeyError, TypeError, ValueError) as exc:
            raise ValueError("资格目标必须填写有效的 Telegram ID。") from exc
        if chat_id in seen:
            continue
        seen.add(chat_id)
        try:
            chat = await app.get_chat(chat_id)
            bot = await app.get_chat_member(chat_id, BOT_ID)
        except Exception as exc:
            raise ValueError(f"机器人无法访问资格目标 {chat_id}。") from exc
        kind = str(getattr(chat.type, "value", chat.type)).lower()
        if kind not in {"channel", "group", "supergroup"}:
            raise ValueError("资格目标只能是频道或群。")
        if bot.status not in {ChatMemberStatus.OWNER, ChatMemberStatus.ADMINISTRATOR}:
            raise ValueError(f"机器人必须是资格目标 {chat_id} 的管理员。")
        normalized.append(
            {
                "chat_id": chat_id, "type": kind,
                "title": str(chat.title or chat_id)[:200],
                "username": str(getattr(chat, "username", None) or "")[:100],
            }
        )
    return normalized


async def check_eligibility(
    giveaway: dict[str, Any], user_id: int
) -> tuple[bool, list[int]]:
    targets = list(giveaway.get("eligibility_targets") or [])
    chat_ids = [int(giveaway["chat_id"]), *[int(item["chat_id"]) for item in targets]]
    outcomes: list[bool] = []
    failed: list[int] = []
    unknown: list[int] = []
    for chat_id in dict.fromkeys(chat_ids):
        try:
            if chat_id != int(giveaway["chat_id"]):
                bot = await app.get_chat_member(chat_id, BOT_ID)
                if bot.status not in {ChatMemberStatus.OWNER, ChatMemberStatus.ADMINISTRATOR}:
                    raise EligibilityUnavailable(f"机器人不再是 {chat_id} 的管理员")
            member = await app.get_chat_member(chat_id, user_id)
        except EligibilityUnavailable:
            unknown.append(chat_id)
            continue
        except Exception:
            unknown.append(chat_id)
            continue
        present = is_member(member)
        if chat_id == int(giveaway["chat_id"]):
            if not present:
                failed.append(chat_id)
        else:
            outcomes.append(present)
            if not present:
                failed.append(chat_id)
    mode = giveaway.get("eligibility_mode", "all")
    if int(giveaway["chat_id"]) in failed:
        return False, failed
    if int(giveaway["chat_id"]) in unknown:
        raise EligibilityUnavailable("无法核验报名群的成员状态")
    if any(int(item["chat_id"]) == int(giveaway["chat_id"]) for item in targets):
        outcomes.append(True)
    if mode == "all" and failed:
        return False, failed
    if mode == "any" and any(outcomes):
        return True, []
    if unknown:
        raise EligibilityUnavailable(
            "无法核验以下频道或群的成员状态：" + ", ".join(map(str, unknown))
        )
    eligible = not targets or (all(outcomes) if mode == "all" else any(outcomes))
    return eligible, ([] if eligible else failed)