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'{escape(str(label))}'
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"抽奖 #{escape(giveaway['giveaway_id'])}",
f"状态:{escape(status_label)}",
f"标题:{escape(giveaway['title'])}",
]
if giveaway.get("description"):
text.append(f"说明:{escape(giveaway['description'])}")
text.extend(
[
"奖项:",
_format_prizes(giveaway["prizes"]),
f"报名开始:{format_giveaway_time(giveaway_starts_at(giveaway))}",
f"开奖时间:{format_giveaway_time(giveaway['ends_at'])}",
"时区:北京时间",
f"参与人数:{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"最低积分:{minimum}")
if cost:
text.append(f"每张奖票:{cost} 积分;每人最多 {int(giveaway.get('max_tickets_per_user', 10))} 张")
else:
text.append("免费抽奖:每人限 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"报名资格({condition}):{escape('、'.join(labels))}")
if reward:
text.append(f"参与奖励:+{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"{title} #{escape(giveaway['giveaway_id'])}",
f"标题:{escape(giveaway['title'])}",
]
if tier_name:
lines.append(f"奖项:{escape(tier_name)}")
if not winners:
lines.append("没有符合条件的参与者。")
return "\n".join(lines)
lines.append("中奖者:")
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]:
current = await get_giveaway(giveaway_id)
if (
current
and current["status"] in {STATUS_RUNNING, STATUS_DRAWING}
and current.get("template_id")
and not current.get("message_id")
):
return False, "报名公告未发布,本期不能开奖。", current
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"]))
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"恭喜你在 {escape(giveaway['title'])} 中获得"
f"{escape(str(winner['tier']))}。"
),
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"] == STATUS_RUNNING
and giveaway.get("template_id")
and not giveaway.get("message_id")
and as_utc(giveaway["ends_at"]) <= utc_now()
):
canceled, _, _ = await cancel_and_refund_giveaway(
giveaway["giveaway_id"], chat_id=int(giveaway["chat_id"]),
publish=False,
)
if canceled:
await giveawaysdb.update_one(
{"giveaway_id": giveaway["giveaway_id"], "status": STATUS_CANCELED},
{"$set": {
"cancellation_reason": "announcement_not_published",
"updated_at": utc_now(),
}},
)
with suppress(Exception):
await record_audit(
source="system", actor_id=None, actor_name="抽奖定时任务",
action="giveaway.announcement_missed",
chat_id=int(giveaway["chat_id"]),
target_id=giveaway["giveaway_id"],
summary="报名公告未发布,开奖时间已到,本期自动取消",
success=False,
)
return
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}"
)