from __future__ import annotations from calendar import monthrange from datetime import datetime, timedelta from secrets import token_hex from typing import Any from zoneinfo import ZoneInfo from wbb import BOT_PROFILE_ID, app from wbb.services.giveaway_eligibility import validate_targets from wbb.utils.dbgiveaway import ( as_utc, attach_giveaway_message, create_giveaway, ensure_giveaway_indexes, giveawaysdb, normalize_prizes, templatesdb, utc_now, ) TIMEZONE = ZoneInfo("Asia/Shanghai") def _time(value: str) -> tuple[int, int]: try: parsed = datetime.strptime(value, "%H:%M") except (TypeError, ValueError) as exc: raise ValueError("时间格式必须为 HH:mm。") from exc return parsed.hour, parsed.minute def period_schedule(template: dict[str, Any], year: int, month: int) -> tuple[datetime, datetime]: day = min(int(template["day_of_month"]), monthrange(year, month)[1]) start_hour, start_minute = _time(template["start_time"]) draw_hour, draw_minute = _time(template["draw_time"]) local_start = datetime(year, month, day, start_hour, start_minute, tzinfo=TIMEZONE) local_draw = datetime(year, month, day, draw_hour, draw_minute, tzinfo=TIMEZONE) local_draw += timedelta(days=int(template["draw_day_offset"])) if local_draw <= local_start: raise ValueError("开奖时间必须晚于报名开始时间。") return as_utc(local_start), as_utc(local_draw) def normalize_template(body: dict[str, Any]) -> dict[str, Any]: day = int(body.get("day_of_month", 1)) offset = int(body.get("draw_day_offset", 0)) if not 1 <= day <= 31 or offset not in {0, 1}: raise ValueError("每月日期必须为 1 至 31，开奖只能在当天或次日。") mode = str(body.get("eligibility_mode", "all")) if mode not in {"all", "any"}: raise ValueError("资格条件组合方式必须是全部或任一。") cost = int(body.get("entry_cost", 0)) limit = int(body.get("max_tickets_per_user", 10)) if cost < 0 or not 1 <= limit <= 100: raise ValueError("积分和奖票上限无效。") values = { "chat_id": int(body["chat_id"]), "title": str(body.get("title") or "").strip()[:200], "description": str(body.get("description") or "").strip()[:4000], "prizes": normalize_prizes(body["prizes"]), "minimum_points": int(body.get("minimum_points", 0)), "entry_cost": cost, "participation_reward": int(body.get("participation_reward", 0)), "max_tickets_per_user": limit if cost else 1, "eligibility_targets": list(body.get("eligibility_targets") or []), "eligibility_mode": mode, "day_of_month": day, "start_time": str(body.get("start_time", "00:00")), "draw_day_offset": offset, "draw_time": str(body.get("draw_time", "20:00")), } if not values["title"] or any(values[key] < 0 for key in ("minimum_points", "participation_reward")): raise ValueError("标题和积分配置无效。") period_schedule(values, 2024, 2) return values async def create_template(body: dict[str, Any], creator_name: str) -> dict[str, Any]: await ensure_giveaway_indexes() values = normalize_template(body) now = utc_now() template = { **values, "template_id": token_hex(8), "bot_id": BOT_PROFILE_ID, "creator_name": creator_name, "active": True, "created_at": now, "updated_at": now, } await templatesdb.insert_one(template) return template async def list_templates(chat_id: int | None = None) -> list[dict[str, Any]]: await ensure_giveaway_indexes() query: dict[str, Any] = {"bot_id": BOT_PROFILE_ID} if chat_id is not None: query["chat_id"] = int(chat_id) return [item async for item in templatesdb.find(query).sort("created_at", -1)] async def update_template(template_id: str, body: dict[str, Any]) -> dict[str, Any] | None: await ensure_giveaway_indexes() changes = normalize_template(body) changes["updated_at"] = utc_now() if "active" in body: changes["active"] = bool(body["active"]) from pymongo import ReturnDocument return await templatesdb.find_one_and_update( {"template_id": template_id, "bot_id": BOT_PROFILE_ID}, {"$set": changes}, return_document=ReturnDocument.AFTER, ) async def materialize_template(template: dict[str, Any], now: datetime | None = None) -> None: await ensure_giveaway_indexes() if not template.get("active"): return now = now or utc_now() local = now.astimezone(TIMEZONE) periods = {(local.year, local.month)} previous = local.replace(day=1) - timedelta(days=1) periods.add((previous.year, previous.month)) from pyrogram.enums import ParseMode from wbb.services.giveaways import join_markup, render_giveaway for year, month in sorted(periods): starts_at, ends_at = period_schedule(template, year, month) if starts_at > now or ends_at <= now or as_utc(template["created_at"]) > ends_at: continue period_key = f"{year:04d}-{month:02d}" existing = await giveawaysdb.find_one( {"template_id": template["template_id"], "period_key": period_key} ) if existing and (existing.get("message_id") or existing["status"] != "running"): continue if not existing: try: await validate_targets(list(template.get("eligibility_targets") or [])) except ValueError as exc: await templatesdb.update_one( {"template_id": template["template_id"]}, {"$set": {"last_error": str(exc), "last_error_at": utc_now()}}, ) raise await templatesdb.update_one( {"template_id": template["template_id"]}, {"$unset": {"last_error": "", "last_error_at": ""}}, ) giveaway = existing or await create_giveaway( chat_id=int(template["chat_id"]), creator_id=0, creator_name=template["creator_name"], title=template["title"], description=template["description"], prizes=template["prizes"], starts_at=starts_at, ends_at=ends_at, minimum_points=template["minimum_points"], entry_cost=template["entry_cost"], participation_reward=template["participation_reward"], max_tickets_per_user=template["max_tickets_per_user"], eligibility_targets=template["eligibility_targets"], eligibility_mode=template["eligibility_mode"], template_id=template["template_id"], period_key=period_key, ) if giveaway.get("message_id"): continue if as_utc(giveaway["ends_at"]) <= utc_now(): continue # Bot accounts cannot call messages.GetHistory. The persisted message ID # is the only reliable indicator that this period was announced. sent = await app.send_message( int(template["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, ) await attach_giveaway_message( giveaway["giveaway_id"], int(template["chat_id"]), sent.id )