420 lines
16 KiB
Python
420 lines
16 KiB
Python
"""兑换码的 SQLite 存储与原子领取逻辑。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sqlite3
|
|
import threading
|
|
from contextlib import contextmanager
|
|
from datetime import UTC, datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
BEIJING_TZ = timezone(timedelta(hours=8), name="Asia/Shanghai")
|
|
|
|
|
|
def normalize_code(value: str) -> str:
|
|
return str(value or "").strip().casefold()
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")
|
|
|
|
|
|
def parse_beijing_datetime(value: str) -> str:
|
|
raw = str(value or "").strip()
|
|
if not raw:
|
|
raise ValueError("生效时间和失效时间不能为空")
|
|
try:
|
|
parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
|
|
except ValueError as exc:
|
|
raise ValueError("时间格式无效") from exc
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=BEIJING_TZ)
|
|
return parsed.astimezone(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")
|
|
|
|
|
|
def to_beijing_datetime(value: str | None) -> str:
|
|
if not value:
|
|
return ""
|
|
parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=UTC)
|
|
return parsed.astimezone(BEIJING_TZ).isoformat(timespec="minutes")
|
|
|
|
|
|
_SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS redemption_codes (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
code_normalized TEXT NOT NULL UNIQUE,
|
|
code_display TEXT NOT NULL,
|
|
points INTEGER NOT NULL CHECK(points > 0),
|
|
starts_at_utc TEXT NOT NULL,
|
|
ends_at_utc TEXT NOT NULL,
|
|
max_redemptions INTEGER CHECK(max_redemptions IS NULL OR max_redemptions > 0),
|
|
redeemed_count INTEGER NOT NULL DEFAULT 0 CHECK(redeemed_count >= 0),
|
|
enabled INTEGER NOT NULL DEFAULT 1 CHECK(enabled IN (0, 1)),
|
|
deleted_at_utc TEXT,
|
|
created_at_utc TEXT NOT NULL,
|
|
updated_at_utc TEXT NOT NULL,
|
|
CHECK(starts_at_utc < ends_at_utc)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS redemption_records (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
code_id INTEGER NOT NULL REFERENCES redemption_codes(id),
|
|
code_display TEXT NOT NULL,
|
|
platform TEXT NOT NULL,
|
|
platform_user_id TEXT NOT NULL,
|
|
display_name TEXT NOT NULL,
|
|
points INTEGER NOT NULL,
|
|
balance_before INTEGER NOT NULL,
|
|
balance_after INTEGER,
|
|
status TEXT NOT NULL DEFAULT 'pending',
|
|
redeemed_at_utc TEXT NOT NULL,
|
|
completed_at_utc TEXT,
|
|
UNIQUE(code_id, platform, platform_user_id)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_redemption_codes_active
|
|
ON redemption_codes(deleted_at_utc, enabled, starts_at_utc, ends_at_utc);
|
|
CREATE INDEX IF NOT EXISTS idx_redemption_records_code_time
|
|
ON redemption_records(code_id, redeemed_at_utc DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_redemption_records_user_time
|
|
ON redemption_records(platform, platform_user_id, redeemed_at_utc DESC);
|
|
"""
|
|
|
|
|
|
class RedemptionCodeStore:
|
|
"""用短事务处理兑换码，避免并发超领和重复领取。
|
|
|
|
当提供 stats_store 时，所有读写都复用 stats_store 的唯一连接与串行写队列，
|
|
消除多连接写同一文件导致的 "database is locked" 争用。
|
|
"""
|
|
|
|
def __init__(self, database_path: str | Path, stats_store: Any | None = None):
|
|
self.database_path = Path(database_path)
|
|
self._store = stats_store
|
|
self._lock = threading.RLock()
|
|
self.database_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with self._connect() as connection:
|
|
connection.executescript(_SCHEMA)
|
|
connection.commit()
|
|
|
|
@contextmanager
|
|
def _connect(self):
|
|
connection = sqlite3.connect(self.database_path, timeout=5.0)
|
|
connection.row_factory = sqlite3.Row
|
|
connection.execute("PRAGMA journal_mode=WAL")
|
|
connection.execute("PRAGMA synchronous=NORMAL")
|
|
connection.execute("PRAGMA foreign_keys=ON")
|
|
connection.execute("PRAGMA busy_timeout=5000")
|
|
try:
|
|
yield connection
|
|
finally:
|
|
connection.close()
|
|
|
|
def _run(self, func, *args):
|
|
"""在 stats_store 单连接（或自有连接）上执行 func(connection, *args)。"""
|
|
if self._store is not None:
|
|
return self._store.execute(func, *args)
|
|
return asyncio.to_thread(self._run_direct, func, args)
|
|
|
|
def _run_direct(self, func, args):
|
|
with self._lock, self._connect() as connection:
|
|
return func(connection, *args)
|
|
|
|
@staticmethod
|
|
def _serialize_code(row: sqlite3.Row) -> dict[str, Any]:
|
|
max_redemptions = row["max_redemptions"]
|
|
redeemed_count = int(row["redeemed_count"] or 0)
|
|
return {
|
|
"id": int(row["id"]),
|
|
"code": row["code_display"],
|
|
"points": int(row["points"]),
|
|
"starts_at": to_beijing_datetime(row["starts_at_utc"]),
|
|
"ends_at": to_beijing_datetime(row["ends_at_utc"]),
|
|
"max_redemptions": int(max_redemptions) if max_redemptions is not None else None,
|
|
"redeemed_count": redeemed_count,
|
|
"remaining_count": max(0, int(max_redemptions) - redeemed_count) if max_redemptions is not None else None,
|
|
"enabled": bool(row["enabled"]),
|
|
"created_at": to_beijing_datetime(row["created_at_utc"]),
|
|
"updated_at": to_beijing_datetime(row["updated_at_utc"]),
|
|
}
|
|
|
|
@staticmethod
|
|
def _serialize_record(row: sqlite3.Row) -> dict[str, Any]:
|
|
return {
|
|
"id": int(row["id"]),
|
|
"code_id": int(row["code_id"]),
|
|
"code": row["code_display"],
|
|
"platform": row["platform"],
|
|
"uid": row["platform_user_id"],
|
|
"uname": row["display_name"],
|
|
"points": int(row["points"]),
|
|
"balance_before": int(row["balance_before"]),
|
|
"balance_after": int(row["balance_after"]) if row["balance_after"] is not None else None,
|
|
"status": row["status"],
|
|
"redeemed_at": to_beijing_datetime(row["redeemed_at_utc"]),
|
|
"completed_at": to_beijing_datetime(row["completed_at_utc"]),
|
|
}
|
|
|
|
async def create_code(
|
|
self,
|
|
*,
|
|
code: str,
|
|
points: int,
|
|
starts_at: str,
|
|
ends_at: str,
|
|
max_redemptions: int | None,
|
|
enabled: bool = True,
|
|
) -> dict[str, Any]:
|
|
return await self._run(
|
|
self._create_code,
|
|
code,
|
|
points,
|
|
starts_at,
|
|
ends_at,
|
|
max_redemptions,
|
|
enabled,
|
|
)
|
|
|
|
def _create_code(
|
|
self,
|
|
connection,
|
|
code: str,
|
|
points: int,
|
|
starts_at: str,
|
|
ends_at: str,
|
|
max_redemptions: int | None,
|
|
enabled: bool,
|
|
) -> dict[str, Any]:
|
|
display = str(code or "").strip()
|
|
normalized = normalize_code(display)
|
|
if not normalized:
|
|
raise ValueError("兑换码不能为空")
|
|
if len(display) > 100:
|
|
raise ValueError("兑换码不能超过100个字符")
|
|
points = int(points)
|
|
if points <= 0:
|
|
raise ValueError("兑换积分必须是正整数")
|
|
if max_redemptions in ("", None):
|
|
max_value = None
|
|
else:
|
|
max_value = int(max_redemptions)
|
|
if max_value <= 0:
|
|
raise ValueError("总兑换次数必须是正整数，或留空表示不限")
|
|
starts_utc = parse_beijing_datetime(starts_at)
|
|
ends_utc = parse_beijing_datetime(ends_at)
|
|
if starts_utc >= ends_utc:
|
|
raise ValueError("失效时间必须晚于生效时间")
|
|
now = utc_now()
|
|
try:
|
|
cursor = connection.execute(
|
|
"""INSERT INTO redemption_codes (
|
|
code_normalized, code_display, points, starts_at_utc, ends_at_utc,
|
|
max_redemptions, enabled, created_at_utc, updated_at_utc
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(normalized, display, points, starts_utc, ends_utc, max_value, int(bool(enabled)), now, now),
|
|
)
|
|
connection.commit()
|
|
except sqlite3.IntegrityError as exc:
|
|
connection.rollback()
|
|
if "code_normalized" in str(exc) or "UNIQUE constraint" in str(exc):
|
|
raise ValueError("兑换码已存在") from exc
|
|
raise
|
|
row = connection.execute("SELECT * FROM redemption_codes WHERE id=?", (cursor.lastrowid,)).fetchone()
|
|
assert row is not None
|
|
return self._serialize_code(row)
|
|
|
|
async def list_codes(self) -> list[dict[str, Any]]:
|
|
return await self._run(self._list_codes)
|
|
|
|
def _list_codes(self, connection) -> list[dict[str, Any]]:
|
|
rows = connection.execute(
|
|
"SELECT * FROM redemption_codes WHERE deleted_at_utc IS NULL ORDER BY id DESC"
|
|
).fetchall()
|
|
return [self._serialize_code(row) for row in rows]
|
|
|
|
async def list_records(self, *, code_id: int | None = None, limit: int = 500) -> list[dict[str, Any]]:
|
|
return await self._run(self._list_records, code_id, limit)
|
|
|
|
def _list_records(self, connection, code_id: int | None, limit: int) -> list[dict[str, Any]]:
|
|
limit = max(1, min(2000, int(limit)))
|
|
if code_id is None:
|
|
rows = connection.execute(
|
|
"SELECT * FROM redemption_records ORDER BY id DESC LIMIT ?", (limit,)
|
|
).fetchall()
|
|
else:
|
|
rows = connection.execute(
|
|
"SELECT * FROM redemption_records WHERE code_id=? ORDER BY id DESC LIMIT ?",
|
|
(int(code_id), limit),
|
|
).fetchall()
|
|
return [self._serialize_record(row) for row in rows]
|
|
|
|
async def set_enabled(self, code_id: int, enabled: bool) -> dict[str, Any]:
|
|
return await self._run(self._set_enabled, code_id, enabled)
|
|
|
|
def _set_enabled(self, connection, code_id: int, enabled: bool) -> dict[str, Any]:
|
|
cursor = connection.execute(
|
|
"UPDATE redemption_codes SET enabled=?, updated_at_utc=? WHERE id=? AND deleted_at_utc IS NULL",
|
|
(int(bool(enabled)), utc_now(), int(code_id)),
|
|
)
|
|
if cursor.rowcount != 1:
|
|
connection.rollback()
|
|
raise ValueError("兑换码不存在")
|
|
connection.commit()
|
|
row = connection.execute("SELECT * FROM redemption_codes WHERE id=?", (int(code_id),)).fetchone()
|
|
assert row is not None
|
|
return self._serialize_code(row)
|
|
|
|
async def delete_code(self, code_id: int) -> None:
|
|
await self._run(self._delete_code, code_id)
|
|
|
|
def _delete_code(self, connection, code_id: int) -> None:
|
|
now = utc_now()
|
|
cursor = connection.execute(
|
|
"""UPDATE redemption_codes
|
|
SET enabled=0, deleted_at_utc=?, updated_at_utc=?
|
|
WHERE id=? AND deleted_at_utc IS NULL""",
|
|
(now, now, int(code_id)),
|
|
)
|
|
if cursor.rowcount != 1:
|
|
connection.rollback()
|
|
raise ValueError("兑换码不存在")
|
|
connection.commit()
|
|
|
|
async def reserve(
|
|
self,
|
|
message: str,
|
|
*,
|
|
platform: str,
|
|
platform_user_id: str,
|
|
display_name: str,
|
|
balance_before: int,
|
|
) -> dict[str, Any]:
|
|
return await self._run(
|
|
self._reserve,
|
|
message,
|
|
platform,
|
|
platform_user_id,
|
|
display_name,
|
|
balance_before,
|
|
)
|
|
|
|
def _reserve(
|
|
self,
|
|
connection,
|
|
message: str,
|
|
platform: str,
|
|
platform_user_id: str,
|
|
display_name: str,
|
|
balance_before: int,
|
|
) -> dict[str, Any]:
|
|
normalized = normalize_code(message)
|
|
if not normalized:
|
|
return {"status": "unknown"}
|
|
now = utc_now()
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
code = connection.execute(
|
|
"SELECT * FROM redemption_codes WHERE code_normalized=? AND deleted_at_utc IS NULL",
|
|
(normalized,),
|
|
).fetchone()
|
|
if code is None:
|
|
connection.rollback()
|
|
return {"status": "unknown"}
|
|
if not bool(code["enabled"]):
|
|
connection.rollback()
|
|
return {"status": "disabled"}
|
|
if now < code["starts_at_utc"]:
|
|
connection.rollback()
|
|
return {"status": "not_started"}
|
|
if now >= code["ends_at_utc"]:
|
|
connection.rollback()
|
|
return {"status": "expired"}
|
|
existing = connection.execute(
|
|
"""SELECT status FROM redemption_records
|
|
WHERE code_id=? AND platform=? AND platform_user_id=?""",
|
|
(code["id"], platform, platform_user_id),
|
|
).fetchone()
|
|
if existing is not None:
|
|
connection.rollback()
|
|
return {"status": "already_redeemed"}
|
|
max_redemptions = code["max_redemptions"]
|
|
if max_redemptions is not None and int(code["redeemed_count"]) >= int(max_redemptions):
|
|
connection.rollback()
|
|
return {"status": "exhausted"}
|
|
cursor = connection.execute(
|
|
"""INSERT INTO redemption_records (
|
|
code_id, code_display, platform, platform_user_id, display_name,
|
|
points, balance_before, status, redeemed_at_utc
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?)""",
|
|
(
|
|
code["id"], code["code_display"], platform, platform_user_id,
|
|
display_name, code["points"], int(balance_before), now,
|
|
),
|
|
)
|
|
updated = connection.execute(
|
|
"""UPDATE redemption_codes
|
|
SET redeemed_count=redeemed_count+1, updated_at_utc=?
|
|
WHERE id=? AND (max_redemptions IS NULL OR redeemed_count < max_redemptions)""",
|
|
(now, code["id"]),
|
|
)
|
|
if updated.rowcount != 1:
|
|
connection.rollback()
|
|
return {"status": "exhausted"}
|
|
connection.commit()
|
|
return {
|
|
"status": "reserved",
|
|
"record_id": int(cursor.lastrowid),
|
|
"code_id": int(code["id"]),
|
|
"code": code["code_display"],
|
|
"points": int(code["points"]),
|
|
}
|
|
except sqlite3.IntegrityError:
|
|
connection.rollback()
|
|
return {"status": "already_redeemed"}
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
|
|
async def finalize(self, record_id: int, balance_after: int) -> None:
|
|
await self._run(self._finalize, record_id, balance_after)
|
|
|
|
def _finalize(self, connection, record_id: int, balance_after: int) -> None:
|
|
cursor = connection.execute(
|
|
"""UPDATE redemption_records
|
|
SET status='completed', balance_after=?, completed_at_utc=?
|
|
WHERE id=? AND status='pending'""",
|
|
(int(balance_after), utc_now(), int(record_id)),
|
|
)
|
|
if cursor.rowcount != 1:
|
|
connection.rollback()
|
|
raise ValueError("兑换记录不存在或已经完成")
|
|
connection.commit()
|
|
|
|
async def cancel(self, record_id: int) -> None:
|
|
await self._run(self._cancel, record_id)
|
|
|
|
def _cancel(self, connection, record_id: int) -> None:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
row = connection.execute(
|
|
"SELECT code_id FROM redemption_records WHERE id=? AND status='pending'",
|
|
(int(record_id),),
|
|
).fetchone()
|
|
if row is None:
|
|
connection.rollback()
|
|
return
|
|
connection.execute("DELETE FROM redemption_records WHERE id=?", (int(record_id),))
|
|
connection.execute(
|
|
"""UPDATE redemption_codes
|
|
SET redeemed_count=MAX(0, redeemed_count-1), updated_at_utc=?
|
|
WHERE id=?""",
|
|
(utc_now(), int(row["code_id"])),
|
|
)
|
|
connection.commit()
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|