TelegramBotAdmin/wbb/services/giveaways.py

from __future__ import annotations

import asyncio
import secrets
from collections import defaultdict
from contextlib import suppress
from datetime import UTC, datetime
from html import escape
from typing import Any
from zoneinfo import ZoneInfo

from pymongo.errors import DuplicateKeyError
from pyrogram.enums import ParseMode
from pyrogram.types import InlineKeyboardButton, InlineKeyboardMarkup

from wbb import BOT_PROFILE_ID, app, log
from wbb.services.giveaway_eligibility import EligibilityUnavailable, check_eligibility
from wbb.utils.dbadmin import record_audit
from wbb.utils.dbgiveaway import (
    STATUS_CANCELED,
    STATUS_CANCELING,
    STATUS_DRAWING,
    STATUS_FINISHED,
    STATUS_RUNNING,
    add_participant,
    as_utc,
    attach_giveaway_message,
    claim_giveaway_cancel,
    claim_giveaway_draw,
    count_participants,
    create_giveaway,
    finalize_giveaway_cancel,
    finish_giveaway,
    get_giveaway,
    get_participant,
    giveawaysdb,
    is_giveaway_banned,
    list_participants,
    mark_participant_refunded,
    participantsdb,
    record_reroll,
    remove_participant,
    save_pending_winners,
    ticket_ordersdb,
    update_running_giveaway,
    utc_now,
)
from wbb.utils.dbpoints import (
    SOURCE_GIVEAWAY_ENTRY,
    SOURCE_GIVEAWAY_PARTICIPATION,
    SOURCE_GIVEAWAY_REFUND,
    SOURCE_GIVEAWAY_WINNER,
    InsufficientPoints,
    adjust_points,
    get_point_account,
    get_point_transaction_by_key,
)
from wbb.utils.i18n import giveaway_status_label

_RANDOM = secrets.SystemRandom()
_giveaway_locks: defaultdict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
GIVEAWAY_TIMEZONE = ZoneInfo("Asia/Shanghai")


class GiveawayServiceError(RuntimeError):
    def __init__(self, code: str, message: str, *, status: int = 400):
        super().__init__(message)
        self.code = code
        self.status = status


def _mongo_datetime_millis(value: datetime) -> int:
    return int(as_utc(value).timestamp() * 1000)


def parse_giveaway_time(raw: str) -> datetime | None:
    try:
        local_time = datetime.strptime(raw.strip(), "%Y-%m-%d %H:%M")
    except ValueError:
        return None
    return local_time.replace(tzinfo=GIVEAWAY_TIMEZONE).astimezone(UTC)


def format_giveaway_time(value: datetime) -> str:
    return as_utc(value).astimezone(GIVEAWAY_TIMEZONE).strftime("%Y-%m-%d %H:%M")


def giveaway_starts_at(giveaway: dict[str, Any]) -> datetime:
    return as_utc(giveaway.get("starts_at") or giveaway["created_at"])


def _participant_mention(participant: dict[str, Any]) -> str:
    user_id = int(participant["user_id"])
    label = (
        participant.get("display_name")
        or ("@" + str(participant["username"]) if participant.get("username") else None)
        or participant.get("first_name")
        or str(user_id)
    )
    return f'<a href="tg://user?id={user_id}">{escape(str(label))}</a>'


def _format_prizes(prizes: list[dict[str, Any]]) -> str:
    lines: list[str] = []
    for prize in prizes:
        points = int(prize.get("points_reward", 0))
        suffix = f"（每人 +{points} 积分）" if points else ""
        lines.append(
            f"- {escape(str(prize['name']))}：{int(prize['count'])} 人{suffix}"
        )
    return "\n".join(lines)


async def render_giveaway(giveaway: dict[str, Any]) -> str:
    participants = await count_participants(giveaway["giveaway_id"])
    status_label = giveaway_status_label(giveaway["status"])
    if giveaway["status"] == STATUS_RUNNING and giveaway_starts_at(giveaway) > utc_now():
        status_label = "待报名"
    text = [
        f"<b>抽奖 #{escape(giveaway['giveaway_id'])}</b>",
        f"<b>状态：</b>{escape(status_label)}",
        f"<b>标题：</b>{escape(giveaway['title'])}",
    ]
    if giveaway.get("description"):
        text.append(f"<b>说明：</b>{escape(giveaway['description'])}")
    text.extend(
        [
            "<b>奖项：</b>",
            _format_prizes(giveaway["prizes"]),
            f"<b>报名开始：</b>{format_giveaway_time(giveaway_starts_at(giveaway))}",
            f"<b>开奖时间：</b>{format_giveaway_time(giveaway['ends_at'])}",
            "<b>时区：</b>北京时间",
            f"<b>参与人数：</b>{participants}",
        ]
    )
    minimum = int(giveaway.get("minimum_points", 0))
    cost = int(giveaway.get("entry_cost", 0))
    reward = int(giveaway.get("participation_reward", 0))
    if minimum:
        text.append(f"<b>最低积分：</b>{minimum}")
    if cost:
        text.append(f"<b>每张奖票：</b>{cost} 积分；每人最多 {int(giveaway.get('max_tickets_per_user', 10))} 张")
    else:
        text.append("<b>免费抽奖：</b>每人限 1 张奖票")
    targets = giveaway.get("eligibility_targets") or []
    if targets:
        condition = "全部满足" if giveaway.get("eligibility_mode", "all") == "all" else "满足任意一个"
        labels = [
            "@" + str(item["username"]) if item.get("username") else str(item.get("title") or item["chat_id"])
            for item in targets
        ]
        text.append(f"<b>报名资格（{condition}）：</b>{escape('、'.join(labels))}")
    if reward:
        text.append(f"<b>参与奖励：</b>+{reward}")
    return "\n".join(text)


def join_markup(giveaway_id: str, entry_cost: int = 0) -> InlineKeyboardMarkup:
    rows = [
            [
                InlineKeyboardButton(
                    "立即报名" if not entry_cost else f"购买 1 张（{entry_cost} 积分）",
                    callback_data=f"giveaway_join:{giveaway_id}"
                )
            ],
        ]
    if entry_cost:
        rows.append([
            InlineKeyboardButton("购买 5 张", callback_data=f"giveaway_buy:{giveaway_id}:5"),
        ])
    rows.extend([
            [
                InlineKeyboardButton(
                    "查看参与者",
                    callback_data=f"giveaway_participants:{giveaway_id}",
                )
            ],
        ])
    return InlineKeyboardMarkup(rows)


async def refresh_giveaway_message(giveaway: dict[str, Any]) -> bool:
    message_id = giveaway.get("message_id")
    if not message_id:
        return True
    try:
        await app.edit_message_text(
            int(giveaway["chat_id"]),
            int(message_id),
            await render_giveaway(giveaway),
            parse_mode=ParseMode.HTML,
            reply_markup=join_markup(giveaway["giveaway_id"], int(giveaway.get("entry_cost", 0))),
        )
    except Exception:
        return False
    return True


async def update_and_refresh_giveaway(
    giveaway_id: str,
    *,
    title: str,
    description: str,
    starts_at: datetime,
    ends_at: datetime,
    expected_updated_at: datetime,
) -> dict[str, Any]:
    async with _giveaway_locks[giveaway_id]:
        giveaway = await get_giveaway(giveaway_id)
        if not giveaway:
            raise GiveawayServiceError(
                "giveaway_not_found", "未找到该抽奖。", status=404
            )
        if giveaway["status"] != STATUS_RUNNING:
            raise GiveawayServiceError(
                "giveaway_not_running", "只有进行中的抽奖可以编辑。", status=409
            )
        if _mongo_datetime_millis(giveaway["updated_at"]) != _mongo_datetime_millis(
            expected_updated_at
        ):
            raise GiveawayServiceError(
                "giveaway_stale",
                "抽奖已被其他操作更新，请刷新页面后重试。",
                status=409,
            )

        current_starts_at = giveaway_starts_at(giveaway)
        if current_starts_at <= utc_now() and _mongo_datetime_millis(
            starts_at
        ) != _mongo_datetime_millis(current_starts_at):
            raise GiveawayServiceError(
                "giveaway_registration_started",
                "报名开始后不能修改报名开始时间。",
                status=409,
            )
        try:
            updated = await update_running_giveaway(
                giveaway_id,
                title=title,
                description=description,
                starts_at=starts_at,
                ends_at=ends_at,
                expected_updated_at=giveaway["updated_at"],
            )
        except ValueError as exc:
            raise GiveawayServiceError("invalid_giveaway", str(exc)) from exc
        if not updated:
            raise GiveawayServiceError(
                "giveaway_stale",
                "抽奖状态或内容已变化，请刷新页面后重试。",
                status=409,
            )
        if not await refresh_giveaway_message(updated):
            raise GiveawayServiceError(
                "giveaway_message_refresh_failed",
                "抽奖内容已保存，但群消息刷新失败，请刷新页面后重试保存。",
                status=502,
            )
        return updated


def pick_winners(
    participants: list[dict[str, Any]], prizes: list[dict[str, Any]]
) -> list[dict[str, Any]]:
    winners: list[dict[str, Any]] = []
    picked_user_ids: set[int] = set()
    for tier_index, prize in enumerate(prizes):
        remaining = [
            participant
            for participant in participants
            if int(participant["user_id"]) not in picked_user_ids
        ]
        picked = []
        for _ in range(min(int(prize["count"]), len(remaining))):
            total = sum(max(1, int(item.get("paid_ticket_count", item.get("ticket_count", 1)))) for item in remaining)
            position = _RANDOM.randrange(total)
            for index, item in enumerate(remaining):
                position -= max(1, int(item.get("paid_ticket_count", item.get("ticket_count", 1))))
                if position < 0:
                    picked.append(remaining.pop(index))
                    break
        for participant in picked:
            user_id = int(participant["user_id"])
            picked_user_ids.add(user_id)
            winners.append(
                {
                    "user_id": user_id,
                    "username": participant.get("username"),
                    "first_name": participant.get("first_name"),
                    "display_name": participant.get("display_name"),
                    "tier": str(prize["name"]),
                    "tier_index": tier_index,
                    "points_reward": int(prize.get("points_reward", 0)),
                }
            )
    return winners


def render_winners(
    giveaway: dict[str, Any],
    winners: list[dict[str, Any]],
    *,
    reroll: bool = False,
    tier_name: str | None = None,
) -> str:
    title = "抽奖重抽结果" if reroll else "抽奖结果"
    lines = [
        f"<b>{title} #{escape(giveaway['giveaway_id'])}</b>",
        f"<b>标题：</b>{escape(giveaway['title'])}",
    ]
    if tier_name:
        lines.append(f"<b>奖项：</b>{escape(tier_name)}")
    if not winners:
        lines.append("没有符合条件的参与者。")
        return "\n".join(lines)
    lines.append("<b>中奖者：</b>")
    for winner in winners:
        points = int(winner.get("points_reward", 0))
        suffix = f"（+{points} 积分）" if points else ""
        lines.append(
            f"- {escape(str(winner['tier']))}：{_participant_mention(winner)}{suffix}"
        )
    return "\n".join(lines)


async def create_and_publish_giveaway(**values: Any) -> dict[str, Any]:
    giveaway = await create_giveaway(**values)
    try:
        sent = await app.send_message(
            giveaway["chat_id"],
            await render_giveaway(giveaway),
            parse_mode=ParseMode.HTML,
            reply_markup=join_markup(giveaway["giveaway_id"], int(giveaway.get("entry_cost", 0))),
            disable_web_page_preview=True,
        )
    except Exception:
        await claim_giveaway_cancel(giveaway["giveaway_id"], giveaway["chat_id"])
        await finalize_giveaway_cancel(giveaway["giveaway_id"])
        raise
    await attach_giveaway_message(giveaway["giveaway_id"], sent.chat.id, sent.id)
    giveaway["message_id"] = sent.id
    return giveaway


async def join_giveaway(
    giveaway_id: str,
    *,
    user_id: int,
    username: str | None,
    first_name: str | None,
    display_name: str | None = None,
    verify_membership: bool = True,
) -> tuple[str, dict[str, Any] | None]:
    async with _giveaway_locks[giveaway_id]:
        giveaway = await get_giveaway(giveaway_id)
        if not giveaway:
            return "missing", None
        if giveaway["status"] != STATUS_RUNNING:
            return "closed", giveaway
        now = utc_now()
        if giveaway_starts_at(giveaway) > now:
            return "not_started", giveaway
        if as_utc(giveaway["ends_at"]) <= now:
            return "ended", giveaway
        existing = await get_participant(giveaway_id, user_id)
        if existing:
            return "duplicate", giveaway
        if verify_membership:
            try:
                eligible, failed = await check_eligibility(giveaway, user_id)
            except EligibilityUnavailable:
                return "eligibility_unavailable", giveaway
            if not eligible:
                return (
                    "not_member" if int(giveaway["chat_id"]) in failed else "not_eligible",
                    giveaway,
                )

        minimum = int(giveaway.get("minimum_points", 0))
        entry_cost = int(giveaway.get("entry_cost", 0))
        account = await get_point_account(int(giveaway["chat_id"]), user_id)
        debit_key: str | None = None
        refund_key: str | None = None
        pending_debit = False
        if entry_cost:
            debit_key, refund_key, pending_debit = await _entry_attempt_keys(
                giveaway_id, user_id
            )
        available_balance = int(account.get("balance", 0)) + (
            entry_cost if pending_debit else 0
        )
        if available_balance < max(minimum, entry_cost):
            return "insufficient_points", giveaway

        if entry_cost and debit_key:
            try:
                await adjust_points(
                    chat_id=int(giveaway["chat_id"]),
                    user_id=user_id,
                    delta=-entry_cost,
                    source=SOURCE_GIVEAWAY_ENTRY,
                    idempotency_key=debit_key,
                    reference_id=giveaway_id,
                    reason=f"抽奖 #{giveaway_id} 报名消耗",
                    username=username,
                    first_name=first_name,
                    display_name=display_name,
                )
            except InsufficientPoints:
                return "insufficient_points", giveaway

        try:
            result = await add_participant(
                giveaway_id=giveaway_id,
                user_id=user_id,
                username=username,
                first_name=first_name,
                display_name=display_name,
            )
        except Exception:
            if entry_cost:
                await _refund_entry(
                    giveaway,
                    user_id=user_id,
                    username=username,
                    first_name=first_name,
                    display_name=display_name,
                    key_suffix="join-failed",
                    idempotency_key=refund_key,
                )
            raise
        if result != "ok" and entry_cost:
            await _refund_entry(
                giveaway,
                user_id=user_id,
                username=username,
                first_name=first_name,
                display_name=display_name,
                key_suffix=f"join-{result}",
                idempotency_key=refund_key,
            )
        if result == "ok":
            await refresh_giveaway_message(giveaway)
    return result, giveaway


async def buy_additional_tickets(
    giveaway_id: str,
    *,
    user_id: int,
    quantity: int,
    request_id: str,
    username: str | None = None,
    first_name: str | None = None,
    display_name: str | None = None,
) -> tuple[str, dict[str, Any] | None]:
    """Reserve capacity before charging; replaying a request never charges twice."""
    if quantity < 1 or quantity > 100 or not request_id or len(request_id) > 100:
        return "invalid_quantity", None
    order_id = f"{giveaway_id}:{user_id}:{request_id}"
    async with _giveaway_locks[giveaway_id]:
        giveaway = await get_giveaway(giveaway_id)
        if not giveaway:
            return "missing", None
        if int(giveaway.get("entry_cost", 0)) <= 0:
            return "free_single_ticket", giveaway
        order = await ticket_ordersdb.find_one({"order_id": order_id})
        if order and order.get("status") == "complete":
            return "duplicate", giveaway
        if order and order.get("status") == "rejected":
            return str(order.get("failure_code") or "ticket_limit"), giveaway
        if not order:
            now = utc_now()
            if giveaway["status"] != STATUS_RUNNING or as_utc(giveaway["ends_at"]) <= now:
                return "ended", giveaway
            if giveaway_starts_at(giveaway) > now:
                return "not_started", giveaway
            participant = await get_participant(giveaway_id, user_id)
            if participant and participant.get("active") is False:
                return "not_joined", giveaway
            if await is_giveaway_banned(int(giveaway["chat_id"]), user_id):
                return "banned", giveaway
            try:
                eligible, _ = await check_eligibility(giveaway, user_id)
            except EligibilityUnavailable:
                return "eligibility_unavailable", giveaway
            if not eligible:
                return "not_eligible", giveaway
            if not participant:
                account = await get_point_account(int(giveaway["chat_id"]), user_id)
                if int(account.get("balance", 0)) < int(giveaway.get("minimum_points", 0)):
                    return "insufficient_points", giveaway
                try:
                    await participantsdb.insert_one({
                        "giveaway_id": giveaway_id,
                        "chat_id": int(giveaway["chat_id"]),
                        "user_id": int(user_id),
                        "username": username,
                        "first_name": first_name,
                        "display_name": display_name or first_name,
                        "entry_cost": int(giveaway["entry_cost"]),
                        "ticket_count": 0,
                        "paid_ticket_count": 0,
                        "pending": True,
                        "active": True,
                        "joined_at": now,
                    })
                except DuplicateKeyError:
                    pass
            order = {
                "order_id": order_id,
                "bot_id": BOT_PROFILE_ID,
                "request_id": request_id,
                "giveaway_id": giveaway_id,
                "chat_id": int(giveaway["chat_id"]),
                "user_id": int(user_id),
                "quantity": quantity,
                "amount": int(giveaway["entry_cost"]) * quantity,
                "username": username,
                "first_name": first_name,
                "display_name": display_name,
                "status": "pending",
                "created_at": now,
                "updated_at": now,
            }
            try:
                await ticket_ordersdb.insert_one(order)
            except Exception:
                order = await ticket_ordersdb.find_one({"order_id": order_id})
                if not order:
                    raise
        quantity = int(order["quantity"])
        participant = await get_participant(giveaway_id, user_id)
        if participant and "ticket_count" not in participant:
            await participantsdb.update_one(
                {"giveaway_id": giveaway_id, "user_id": user_id, "ticket_count": {"$exists": False}},
                {"$set": {"ticket_count": 1, "paid_ticket_count": 1}},
            )
        if not participant or participant.get("active") is False or giveaway["status"] in {STATUS_CANCELED, STATUS_CANCELING, STATUS_FINISHED}:
            debit_key = f"giveaway-ticket:{order_id}"
            if await get_point_transaction_by_key(debit_key):
                await adjust_points(
                    chat_id=int(giveaway["chat_id"]), user_id=user_id,
                    delta=int(order["amount"]), source=SOURCE_GIVEAWAY_REFUND,
                    idempotency_key=f"giveaway-ticket-refund:{order_id}",
                    reference_id=giveaway_id, reason=f"抽奖 #{giveaway_id} 奖票购买取消退款",
                    username=username, first_name=first_name, display_name=display_name,
                )
            if participant and order_id in participant.get("ticket_order_ids", []):
                await participantsdb.update_one(
                    {"giveaway_id": giveaway_id, "user_id": user_id, "ticket_order_ids": order_id},
                    {"$inc": {"ticket_count": -quantity}, "$pull": {"ticket_order_ids": order_id}},
                )
            await ticket_ordersdb.update_one(
                {"order_id": order_id},
                {"$set": {"status": "rejected", "failure_code": "not_joined", "updated_at": utc_now()}},
            )
            return "not_joined", giveaway
        participant_filter = {
            "giveaway_id": giveaway_id,
            "user_id": int(user_id),
            "active": {"$ne": False},
        }
        reserved = await participantsdb.update_one(
            {
                **participant_filter,
                "ticket_order_ids": {"$ne": order_id},
                "ticket_count": {
                    "$lte": int(giveaway.get("max_tickets_per_user", 10)) - quantity
                },
            },
            {"$inc": {"ticket_count": quantity}, "$addToSet": {"ticket_order_ids": order_id}},
        )
        if not reserved.modified_count:
            participant = await get_participant(giveaway_id, user_id)
            if not participant or order_id not in participant.get("ticket_order_ids", []):
                await ticket_ordersdb.update_one(
                    {"order_id": order_id}, {"$set": {"status": "rejected", "failure_code": "ticket_limit", "updated_at": utc_now()}}
                )
                return "ticket_limit", giveaway
        try:
            await adjust_points(
                chat_id=int(giveaway["chat_id"]), user_id=user_id,
                delta=-int(order["amount"]), source=SOURCE_GIVEAWAY_ENTRY,
                idempotency_key=f"giveaway-ticket:{order_id}", reference_id=giveaway_id,
                reason=f"抽奖 #{giveaway_id} 购买 {quantity} 张奖票",
                username=username, first_name=first_name, display_name=display_name,
            )
        except InsufficientPoints:
            await participantsdb.update_one(
                {**participant_filter, "ticket_order_ids": order_id},
                {"$inc": {"ticket_count": -quantity}, "$pull": {"ticket_order_ids": order_id}},
            )
            await ticket_ordersdb.update_one(
                {"order_id": order_id}, {"$set": {"status": "rejected", "failure_code": "insufficient_points", "updated_at": utc_now()}}
            )
            return "insufficient_points", giveaway
        await participantsdb.update_one(
            {**participant_filter, "completed_ticket_order_ids": {"$ne": order_id}},
            {
                "$inc": {"paid_ticket_count": quantity},
                "$addToSet": {"completed_ticket_order_ids": order_id},
                "$set": {"pending": False},
            },
        )
        await ticket_ordersdb.update_one(
            {"order_id": order_id}, {"$set": {"status": "complete", "updated_at": utc_now()}}
        )
        await refresh_giveaway_message(giveaway)
        return "ok", giveaway


async def reconcile_pending_ticket_orders(limit: int = 50) -> None:
    cursor = ticket_ordersdb.find({"bot_id": BOT_PROFILE_ID, "status": "pending"}).sort("created_at", 1).limit(limit)
    for order in [item async for item in cursor]:
        try:
            await buy_additional_tickets(
                str(order["giveaway_id"]), user_id=int(order["user_id"]),
                quantity=int(order["quantity"]),
                request_id=str(order.get("request_id") or order["order_id"].rsplit(":", 1)[-1]),
                username=order.get("username"), first_name=order.get("first_name"),
                display_name=order.get("display_name"),
            )
        except Exception as exc:
            log.error(f"奖票订单 {order['order_id']} 恢复失败：{exc}")


async def _entry_attempt_keys(
    giveaway_id: str, user_id: int
) -> tuple[str, str, bool]:
    base = f"giveaway-entry:{giveaway_id}:{user_id}"
    for attempt in range(100):
        suffix = "" if attempt == 0 else f":{attempt}"
        debit_key = f"{base}{suffix}"
        refund_key = f"giveaway-refund:{giveaway_id}:{user_id}:entry-attempt:{attempt}"
        debit = await get_point_transaction_by_key(debit_key)
        if not debit:
            return debit_key, refund_key, False
        if not await get_point_transaction_by_key(refund_key):
            return debit_key, refund_key, True
    raise GiveawayServiceError(
        "entry_retry_limit", "报名重试次数过多，请稍后再试。", status=409
    )


async def _refund_entry(
    giveaway: dict[str, Any],
    *,
    user_id: int,
    username: str | None,
    first_name: str | None,
    display_name: str | None,
    key_suffix: str,
    idempotency_key: str | None = None,
    amount: int | None = None,
) -> bool:
    entry_cost = int(giveaway.get("entry_cost", 0)) if amount is None else int(amount)
    if entry_cost <= 0:
        return False
    _, created = await adjust_points(
        chat_id=int(giveaway["chat_id"]),
        user_id=user_id,
        delta=entry_cost,
        source=SOURCE_GIVEAWAY_REFUND,
        idempotency_key=idempotency_key
        or f"giveaway-refund:{giveaway['giveaway_id']}:{user_id}:{key_suffix}",
        reference_id=giveaway["giveaway_id"],
        reason=f"抽奖 #{giveaway['giveaway_id']} 报名退款",
        username=username,
        first_name=first_name,
        display_name=display_name,
    )
    if created:
        await mark_participant_refunded(
            giveaway["giveaway_id"], user_id, key_suffix
        )
    return created


async def _award_draw_points(
    giveaway: dict[str, Any],
    participants: list[dict[str, Any]],
    winners: list[dict[str, Any]],
) -> None:
    participation_reward = int(giveaway.get("participation_reward", 0))
    if participation_reward:
        for participant in participants:
            await adjust_points(
                chat_id=int(giveaway["chat_id"]),
                user_id=int(participant["user_id"]),
                delta=participation_reward,
                source=SOURCE_GIVEAWAY_PARTICIPATION,
                idempotency_key=(
                    f"giveaway-participation:{giveaway['giveaway_id']}:"
                    f"{participant['user_id']}"
                ),
                reference_id=giveaway["giveaway_id"],
                reason=f"抽奖 #{giveaway['giveaway_id']} 参与奖励",
                username=participant.get("username"),
                first_name=participant.get("first_name"),
                display_name=participant.get("display_name"),
            )
    for winner in winners:
        reward = int(winner.get("points_reward", 0))
        if not reward:
            continue
        await adjust_points(
            chat_id=int(giveaway["chat_id"]),
            user_id=int(winner["user_id"]),
            delta=reward,
            source=SOURCE_GIVEAWAY_WINNER,
            idempotency_key=(
                f"giveaway-winner:{giveaway['giveaway_id']}:"
                f"{winner.get('tier_index', 0)}:{winner['user_id']}"
            ),
            reference_id=giveaway["giveaway_id"],
            reason=(
                f"抽奖 #{giveaway['giveaway_id']} 中奖奖励"
                f"（{winner['tier']}）"
            ),
            username=winner.get("username"),
            first_name=winner.get("first_name"),
            display_name=winner.get("display_name"),
        )


async def finish_and_publish_giveaway(
    giveaway_id: str, *, publish: bool = True
) -> tuple[bool, str, dict[str, Any] | None]:
    async with _giveaway_locks[giveaway_id]:
        giveaway = await claim_giveaway_draw(giveaway_id)
        if not giveaway:
            existing = await get_giveaway(giveaway_id)
            if not existing:
                return False, "未找到该抽奖。", None
            if existing["status"] == STATUS_FINISHED:
                return False, "该抽奖已经完成开奖。", existing
            return False, "该抽奖当前不在进行中。", existing

        pending_order = await ticket_ordersdb.find_one(
            {"giveaway_id": giveaway_id, "status": "pending"}
        )
        if pending_order:
            raise GiveawayServiceError(
                "ticket_order_pending", "奖票订单尚未结算，开奖将稍后重试。", status=503
            )
        participants = await list_participants(giveaway_id)
        eligible_participants = [
            participant for participant in participants
            if participant.get("eligibility_status") == "eligible"
        ]
        if giveaway.get("pending_winners") is not None and not any(
            "eligibility_status" in item for item in participants
        ) and not giveaway.get("eligibility_targets"):
            eligible_participants = participants
        if giveaway.get("pending_winners") is None:
            eligible_participants = []
            try:
                for participant in participants:
                    eligible, failed = await check_eligibility(
                        giveaway, int(participant["user_id"])
                    )
                    await participantsdb.update_one(
                        {"giveaway_id": giveaway_id, "user_id": participant["user_id"]},
                        {"$set": {
                            "eligibility_status": "eligible" if eligible else "ineligible",
                            "eligibility_failed_chat_ids": failed,
                            "eligibility_checked_at": utc_now(),
                        }},
                    )
                    if eligible:
                        eligible_participants.append(participant)
            except EligibilityUnavailable as exc:
                changed = await giveawaysdb.update_one(
                    {
                        "giveaway_id": giveaway_id,
                        "status": STATUS_DRAWING,
                        "eligibility_error": {"$ne": str(exc)},
                    },
                    {"$set": {"eligibility_error": str(exc), "updated_at": utc_now()}},
                )
                if changed.modified_count:
                    with suppress(Exception):
                        await record_audit(
                            source="system", actor_id=None, actor_name="抽奖定时任务",
                            action="giveaway.eligibility_paused",
                            chat_id=int(giveaway["chat_id"]), target_id=giveaway_id,
                            summary="开奖资格核验失败，已暂停并等待重试",
                            success=False, error=str(exc),
                        )
                raise GiveawayServiceError(
                    "eligibility_unavailable", "资格核验暂不可用，开奖已暂停并将自动重试。", status=503
                ) from exc
            await giveawaysdb.update_one(
                {"giveaway_id": giveaway_id}, {"$unset": {"eligibility_error": ""}}
            )
            pending = pick_winners(eligible_participants, giveaway["prizes"])
            giveaway = await save_pending_winners(giveaway_id, pending) or giveaway
        winners = list(giveaway.get("pending_winners", []))
        await _award_draw_points(giveaway, eligible_participants, winners)
        marked = await finish_giveaway(giveaway_id, winners)
        if not marked:
            current = await get_giveaway(giveaway_id)
            if not current or current["status"] != STATUS_FINISHED:
                return False, "抽奖无法完成，请稍后重试。", current
            giveaway = current
            winners = list(current.get("winners", []))
        else:
            giveaway = {**giveaway, "status": STATUS_FINISHED, "winners": winners}

    if publish:
        await publish_finished(giveaway, winners)
    return True, f"抽奖 #{giveaway_id} 已完成开奖。", giveaway


async def publish_finished(
    giveaway: dict[str, Any], winners: list[dict[str, Any]]
) -> None:
    giveaway = await get_giveaway(giveaway["giveaway_id"]) or giveaway
    chat_id = int(giveaway["chat_id"])
    message_id = giveaway.get("message_id")
    text = render_winners(giveaway, winners)
    result_message = None
    if giveaway.get("result_message_id"):
        from types import SimpleNamespace

        result_message = SimpleNamespace(id=int(giveaway["result_message_id"]))
    else:
        try:
            async for recent in app.get_chat_history(chat_id, limit=100):
                if f"抽奖结果 #{giveaway['giveaway_id']}" in str(getattr(recent, "text", "")):
                    result_message = recent
                    break
        except Exception:
            pass
    if message_id:
        with suppress(Exception):
            await app.edit_message_text(
                chat_id,
                int(message_id),
                await render_giveaway({**giveaway, "status": STATUS_FINISHED}),
                parse_mode=ParseMode.HTML,
                reply_markup=None,
            )
        if result_message is None:
            with suppress(Exception):
                result_message = await app.send_message(
                    chat_id, text, parse_mode=ParseMode.HTML,
                    reply_to_message_id=int(message_id),
                )
    if result_message is None:
        with suppress(Exception):
            result_message = await app.send_message(
                chat_id, text, parse_mode=ParseMode.HTML
            )
    if result_message is None:
        log.error(f"抽奖 {giveaway['giveaway_id']} 结果发送失败，等待重试")
        return
    await giveawaysdb.update_one(
        {"giveaway_id": giveaway["giveaway_id"]},
        {"$set": {"result_message_id": int(result_message.id), "updated_at": utc_now()}},
    )
    result_pinned = bool(giveaway.get("result_pinned"))
    if not result_pinned:
        try:
            await app.pin_chat_message(
                chat_id,
                int(result_message.id),
                disable_notification=True,
            )
            result_pinned = True
        except Exception as exc:
            log.error(
                f"抽奖 {giveaway['giveaway_id']} 结果置顶失败：{exc}"
            )
    if result_pinned and message_id and int(message_id) != int(result_message.id):
        try:
            await app.delete_messages(chat_id, int(message_id))
        except Exception:
            # If Telegram refuses deletion, at least remove the stale pinned entry.
            with suppress(Exception):
                await app.unpin_chat_message(chat_id, int(message_id))
    await giveawaysdb.update_one(
        {"giveaway_id": giveaway["giveaway_id"]},
        {"$set": {
            "result_pinned": result_pinned,
            "publication_pending": not result_pinned,
            "updated_at": utc_now(),
        }},
    )
    if not result_pinned:
        return
    for winner in winners:
        with suppress(Exception):
            await app.send_message(
                int(winner["user_id"]),
                (
                    f"恭喜你在 <b>{escape(giveaway['title'])}</b> 中获得"
                    f"<b>{escape(str(winner['tier']))}</b>。"
                ),
                parse_mode=ParseMode.HTML,
            )


async def cancel_and_refund_giveaway(
    giveaway_id: str, *, chat_id: int, publish: bool = True
) -> tuple[bool, str, dict[str, Any] | None]:
    await reconcile_pending_ticket_orders()
    async with _giveaway_locks[giveaway_id]:
        giveaway = await claim_giveaway_cancel(giveaway_id, chat_id)
        if not giveaway:
            existing = await get_giveaway(giveaway_id)
            if not existing:
                return False, "未找到该抽奖。", None
            if existing["status"] == STATUS_CANCELED:
                return False, "该抽奖已经取消。", existing
            return False, "该抽奖当前不在进行中。", existing
        participants = await list_participants(giveaway_id)
        for participant in participants:
            await _refund_entry(
                giveaway,
                user_id=int(participant["user_id"]),
                username=participant.get("username"),
                first_name=participant.get("first_name"),
                display_name=participant.get("display_name"),
                key_suffix="cancel",
                amount=int(giveaway.get("entry_cost", 0)) * int(
                    participant.get("paid_ticket_count", participant.get("ticket_count", 1))
                ),
            )
        await finalize_giveaway_cancel(giveaway_id)
        giveaway = {**giveaway, "status": STATUS_CANCELED}
    if publish and giveaway.get("message_id"):
        with suppress(Exception):
            await app.edit_message_text(
                int(giveaway["chat_id"]),
                int(giveaway["message_id"]),
                await render_giveaway(giveaway),
                parse_mode=ParseMode.HTML,
                reply_markup=None,
            )
    return True, f"抽奖 #{giveaway_id} 已取消，报名积分已退还。", giveaway


async def remove_and_optionally_refund_participant(
    *,
    giveaway_id: str,
    user_id: int,
    moderator_id: int,
    reason: str,
    refund: bool = True,
) -> bool:
    giveaway = await get_giveaway(giveaway_id)
    if not giveaway or giveaway["status"] != STATUS_RUNNING:
        raise GiveawayServiceError(
            "giveaway_not_running", "该抽奖当前不在进行中。"
        )
    participant = await remove_participant(
        giveaway_id=giveaway_id,
        user_id=user_id,
        moderator_id=moderator_id,
        reason=reason,
        refund=refund,
    )
    if not participant:
        return False
    if refund:
        await _refund_entry(
            giveaway,
            user_id=user_id,
            username=participant.get("username"),
            first_name=participant.get("first_name"),
            display_name=participant.get("display_name"),
            key_suffix="participant-removed",
            amount=int(giveaway.get("entry_cost", 0)) * int(
                participant.get("paid_ticket_count", participant.get("ticket_count", 1))
            ),
        )
    await refresh_giveaway_message(giveaway)
    return True


async def reroll_giveaway(
    *,
    giveaway_id: str,
    moderator_id: int,
    tier_name: str | None = None,
    reroll_id: str | None = None,
    publish: bool = True,
) -> tuple[list[dict[str, Any]], str]:
    reroll_id = reroll_id or secrets.token_hex(8)
    async with _giveaway_locks[giveaway_id]:
        giveaway = await get_giveaway(giveaway_id)
        if not giveaway or giveaway["status"] != STATUS_FINISHED:
            raise GiveawayServiceError(
                "giveaway_not_finished", "只有已开奖的抽奖可以重抽。"
            )
        for previous in giveaway.get("rerolls", []):
            if previous.get("reroll_id") == reroll_id:
                return list(previous.get("winners", [])), reroll_id
        prizes = list(giveaway["prizes"])
        if tier_name:
            prizes = [
                prize
                for prize in prizes
                if str(prize["name"]).strip().lower() == tier_name.strip().lower()
            ]
        if not prizes:
            raise GiveawayServiceError("tier_not_found", "未找到该奖项。")
        excluded = {int(winner["user_id"]) for winner in giveaway.get("winners", [])}
        for previous in giveaway.get("rerolls", []):
            excluded.update(
                int(winner["user_id"]) for winner in previous.get("winners", [])
            )
        participants = [
            participant
            for participant in await list_participants(giveaway_id)
            if int(participant["user_id"]) not in excluded
        ]
        eligible_participants = []
        try:
            for participant in participants:
                eligible, _ = await check_eligibility(giveaway, int(participant["user_id"]))
                if eligible:
                    eligible_participants.append(participant)
        except EligibilityUnavailable as exc:
            raise GiveawayServiceError(
                "eligibility_unavailable", "资格核验暂不可用，重抽已暂停。", status=503
            ) from exc
        winners = pick_winners(eligible_participants, prizes)
        await record_reroll(
            giveaway_id,
            winners,
            tier_name,
            moderator_id,
            reroll_id=reroll_id,
        )
        for winner in winners:
            reward = int(winner.get("points_reward", 0))
            if reward:
                await adjust_points(
                    chat_id=int(giveaway["chat_id"]),
                    user_id=int(winner["user_id"]),
                    delta=reward,
                    source=SOURCE_GIVEAWAY_WINNER,
                    idempotency_key=(
                        f"giveaway-reroll-winner:{giveaway_id}:{reroll_id}:"
                        f"{winner.get('tier_index', 0)}:{winner['user_id']}"
                    ),
                    reference_id=giveaway_id,
                    reason=f"抽奖 #{giveaway_id} 重抽中奖奖励",
                    username=winner.get("username"),
                    first_name=winner.get("first_name"),
                    display_name=winner.get("display_name"),
                )
    if publish:
        with suppress(Exception):
            await app.send_message(
                int(giveaway["chat_id"]),
                render_winners(giveaway, winners, reroll=True, tier_name=tier_name),
                parse_mode=ParseMode.HTML,
                disable_web_page_preview=True,
            )
    return winners, reroll_id


async def resume_pending_giveaway(giveaway: dict[str, Any]) -> None:
    try:
        if giveaway["status"] in {STATUS_RUNNING, STATUS_DRAWING}:
            await finish_and_publish_giveaway(giveaway["giveaway_id"])
        elif giveaway["status"] == STATUS_CANCELING:
            await cancel_and_refund_giveaway(
                giveaway["giveaway_id"], chat_id=int(giveaway["chat_id"])
            )
    except Exception as exc:
        log.error(
            f"抽奖 {giveaway.get('giveaway_id')} 恢复任务失败：{exc}"
        )