yml2213
bb20533c90
- 扩展号池支持 peace (和平精英/小跑道) 活动 scope，映射 xpd_balance 与 exchange_xpd_goods 任务 - 新增 POST /douyu/pool/refresh-goods 接口与前端一键刷新，支持复用库内有效 CK 自动更新商品快照 - 前端 DouyuPoolPage 升级视觉设计与栅格布局，商品选择器支持按活动类型分类联动 - 修复 DouyuTask.result 类型守卫、account_events actor_id 类型防御与 runner 过滤规则边界值处理 - 验证方式：全量 pytest (797 通过)、pyright (0 错误)、tsc -b 及 ruff format/check 全部通过
742 lines
25 KiB
Python
742 lines
25 KiB
Python
"""账号池：候选筛选、两梯队排序、供给预估、租约领取与释放。
|
||
|
||
配套文档：`docs/设计-账号自动分配与回收.md`（§3.3 候选与排序、§3.4 补给链、
|
||
§3.5 原子领取、§3.9 自动收回）。
|
||
|
||
## 为什么要拆「纯函数」与「SQL」
|
||
|
||
排序与充值量计算是**业务规则的集中地**（先花快过期的、先用小余额、
|
||
缺多少补多少），也是最容易被改错、必须能被反向验证的部分 ——
|
||
所以它们是不碰数据库的纯函数，用例可以脱离 MySQL 直接喂数据。
|
||
|
||
SQL 只负责**筛选**（谁的 CK 有效、谁已开通、谁 365 天内没兑过、谁没被租走）。
|
||
|
||
## 两个必须记住的约束
|
||
|
||
1. **`elite_opened = 1` 是硬门槛**：开通精英宝典要人工扫码
|
||
（`create_elite_qr` 生成支付码），系统不能自动开通 —— 未开通的账号进池也白进。
|
||
2. **并发抢号靠数据库**：`account_leases.active_account_id` 上的唯一索引裁决胜负，
|
||
撞键就换下一个候选（SAVEPOINT 包住，不影响已经领到的那些）。
|
||
不要退回「先查再插」——那在并发下必然双发。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Iterable, Sequence
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any
|
||
|
||
from sqlalchemy import and_, func, not_, or_, select
|
||
from sqlalchemy.exc import IntegrityError
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..models import Account, AccountEvent, AccountLease, DouyuTask, User
|
||
from .account_binding import SCOPE_ELITE, SCOPE_ESPORTS, SCOPE_PEACE
|
||
|
||
#: 租约默认有效期（分钟）。到点由 `account_lease_reaper` 兜底回收。
|
||
DEFAULT_LEASE_TTL_MINUTES = 60
|
||
|
||
#: 同一件皮肤多久内不允许在同一账号再兑（滚动天数）。
|
||
EXCHANGE_COOLDOWN_DAYS = 365
|
||
|
||
#: 1 个精英令 = 10 精英积分 = 1 鱼翅 = 1 元（2026-10-07 生产实测）。
|
||
POINTS_PER_GIFT = 10
|
||
|
||
#: 单次鱼翅充值上限（元）。见设计 §3.4.2。
|
||
MAX_RECHARGE_AMOUNT = 800
|
||
|
||
#: 池成员 = 未分配 ∪ 管理员/运营名下；客服名下账号不进池（不跨归属）。
|
||
_POOL_OWNER_ROLES = ("super_admin", "operation")
|
||
|
||
#: 作用域 → （积分列名，开通标记列名）。和平小店没有「开通」概念。
|
||
_SCOPE_COLUMNS: dict[str, tuple[str, str | None]] = {
|
||
SCOPE_ELITE: ("points", "elite_opened"),
|
||
SCOPE_ESPORTS: ("esports_points", "esports_opened"),
|
||
SCOPE_PEACE: ("xpd_balance", None),
|
||
}
|
||
|
||
SUPPORTED_POOL_SCOPES = tuple(_SCOPE_COLUMNS)
|
||
|
||
|
||
def _now(now: datetime | None = None) -> datetime:
|
||
return now or datetime.now(UTC)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Candidate:
|
||
"""一个可分配账号在**本次兑换**下的画像。"""
|
||
|
||
account_id: int
|
||
username: str
|
||
points: int
|
||
score: int
|
||
tier: str # "A" = 积分已够（免充值）；"B" = 需要充值
|
||
topup_gifts: int
|
||
points_expire_at: datetime | None = None
|
||
role_change_available_at: datetime | None = None
|
||
|
||
@property
|
||
def topup_yuan(self) -> int:
|
||
"""需要充多少元（1 元 = 1 鱼翅 = 1 精英令）。"""
|
||
return self.topup_gifts
|
||
|
||
@property
|
||
def needs_topup(self) -> bool:
|
||
return self.topup_gifts > 0
|
||
|
||
|
||
def topup_gifts_for(points: int | None, score: int) -> int:
|
||
"""凑够 `score` 还差几个精英令（向下取整的缺口向上取整到整数个令）。
|
||
|
||
>>> topup_gifts_for(300, 1280)
|
||
98
|
||
>>> topup_gifts_for(1280, 1280)
|
||
0
|
||
>>> topup_gifts_for(None, 20)
|
||
2
|
||
"""
|
||
current = int(points or 0)
|
||
missing = max(0, int(score) - current)
|
||
return (missing + POINTS_PER_GIFT - 1) // POINTS_PER_GIFT
|
||
|
||
|
||
def _sort_key(candidate: Candidate):
|
||
"""两梯队排序键（文档 §3.3）。
|
||
|
||
A 梯队（免充值）：先花**快过期**的 → 再用**小余额**（把大额账号留给贵商品）。
|
||
B 梯队（需充值）：先选**充值最少**的 → 再用**大余额**（留存量的先用掉）。
|
||
|
||
`points_expire_at` 为空排最后；最后统一用 `account_id` 兜稳定顺序，
|
||
保证同样的输入永远得到同样的顺序（否则用例无法断言）。
|
||
"""
|
||
if candidate.tier == "A":
|
||
expire_missing = candidate.points_expire_at is None
|
||
expire_key = candidate.points_expire_at or datetime.max.replace(tzinfo=UTC)
|
||
return (0, expire_missing, expire_key, candidate.points, candidate.account_id)
|
||
return (
|
||
1,
|
||
candidate.topup_gifts,
|
||
-candidate.points,
|
||
candidate.account_id,
|
||
)
|
||
|
||
|
||
def make_candidate(
|
||
*,
|
||
account_id: int,
|
||
username: str,
|
||
points: int | None,
|
||
score: int,
|
||
points_expire_at: datetime | None = None,
|
||
role_change_available_at: datetime | None = None,
|
||
) -> Candidate:
|
||
"""由账号行构造候选（纯函数，便于用例直接喂数据）。"""
|
||
current = int(points or 0)
|
||
gifts = topup_gifts_for(current, score)
|
||
return Candidate(
|
||
account_id=account_id,
|
||
username=username,
|
||
points=current,
|
||
score=int(score),
|
||
tier="A" if gifts == 0 else "B",
|
||
topup_gifts=gifts,
|
||
points_expire_at=points_expire_at,
|
||
role_change_available_at=role_change_available_at,
|
||
)
|
||
|
||
|
||
def rank_candidates(candidates: Iterable[Candidate]) -> list[Candidate]:
|
||
"""按两梯队规则排序（纯函数）。"""
|
||
return sorted(candidates, key=_sort_key)
|
||
|
||
|
||
def latest_exchange_exists_clause(commodity_id: str, since: datetime):
|
||
"""「该账号在 `since` 之后兑过这件商品」的子查询条件。"""
|
||
return (
|
||
select(AccountEvent.id)
|
||
.where(
|
||
AccountEvent.account_id == Account.id,
|
||
AccountEvent.event_type == "exchange",
|
||
# 只按 subject_key 判：D1 回填后我方任务与官方补录都是商品 ID。
|
||
# 不额外约束 subject_type —— 部分回填的历史行 subject_type 仍是 goods，
|
||
# 按它过滤会漏判（这正是 D1 要修的东西）。
|
||
AccountEvent.subject_key == commodity_id,
|
||
AccountEvent.occurred_at >= since,
|
||
)
|
||
.exists()
|
||
)
|
||
|
||
|
||
def active_lease_exists_clause():
|
||
"""「该账号存在有效租约」的子查询条件。"""
|
||
return (
|
||
select(AccountLease.id)
|
||
.where(AccountLease.active_account_id == Account.id)
|
||
.exists()
|
||
)
|
||
|
||
|
||
def pool_membership_clause():
|
||
"""池成员：未分配 ∪ 管理员/运营名下（客服名下账号不进池）。"""
|
||
owner_ids = select(User.id).where(User.role.in_(_POOL_OWNER_ROLES))
|
||
return or_(Account.assigned_to.is_(None), Account.assigned_to.in_(owner_ids))
|
||
|
||
|
||
def candidate_query(
|
||
*,
|
||
scope: str,
|
||
score: int,
|
||
commodity_id: str,
|
||
now: datetime | None = None,
|
||
):
|
||
"""候选账号的基础查询（**未排序、未限量**，供计数与取行共用）。"""
|
||
if scope not in _SCOPE_COLUMNS:
|
||
raise ValueError(f"暂不支持的作用域: {scope}")
|
||
points_column_name, opened_column_name = _SCOPE_COLUMNS[scope]
|
||
points_column = getattr(Account, points_column_name)
|
||
since = _now(now) - timedelta(days=EXCHANGE_COOLDOWN_DAYS)
|
||
|
||
conditions: list[Any] = [
|
||
Account.status == "active",
|
||
Account.ck_check_status == "valid",
|
||
Account.cookie.isnot(None),
|
||
Account.cookie != "",
|
||
Account.auto_managed.is_(True),
|
||
pool_membership_clause(),
|
||
not_(active_lease_exists_clause()),
|
||
not_(latest_exchange_exists_clause(commodity_id, since)),
|
||
]
|
||
if opened_column_name:
|
||
# 未开通的账号系统开不了（要人工扫码），直接排除。
|
||
# `NULL` = 没查过，同样排除 —— 不能把「不知道」当「已开通」放行。
|
||
conditions.append(getattr(Account, opened_column_name).is_(True))
|
||
# 积分未知（没查过）时按 0 处理会把充值量算错，先排除。
|
||
conditions.append(points_column.isnot(None))
|
||
return select(Account).where(and_(*conditions)), points_column
|
||
|
||
|
||
def eligible_accounts(
|
||
db: Session,
|
||
*,
|
||
scope: str = SCOPE_ELITE,
|
||
score: int,
|
||
commodity_id: str,
|
||
now: datetime | None = None,
|
||
limit: int | None = None,
|
||
) -> list[Candidate]:
|
||
"""筛出可分配账号并按两梯队排序。
|
||
|
||
`limit` 是为了「要 5 个」时不用把整个池拉出来 —— 但**排序键里有 Python 侧
|
||
的表达式**（到期时间、充值量），所以 SQL 只做粗筛 + 有界取行，再在 Python 里排。
|
||
取值上限给足余量（默认 `limit*5`，至少 200），否则「积分够的账号排在后面」
|
||
会被截断掉。
|
||
"""
|
||
query, points_column = candidate_query(
|
||
scope=scope, score=score, commodity_id=commodity_id, now=now
|
||
)
|
||
query = query.order_by(points_column.desc(), Account.id.asc())
|
||
if limit is not None:
|
||
query = query.limit(max(200, int(limit) * 5))
|
||
rows = db.execute(query).scalars().all()
|
||
candidates = [
|
||
make_candidate(
|
||
account_id=row.id,
|
||
username=row.username,
|
||
points=getattr(row, _SCOPE_COLUMNS[scope][0]),
|
||
score=score,
|
||
points_expire_at=row.points_expire_at,
|
||
role_change_available_at=row.role_change_available_at,
|
||
)
|
||
for row in rows
|
||
]
|
||
return rank_candidates(candidates)
|
||
|
||
|
||
def estimate_supply(
|
||
db: Session,
|
||
*,
|
||
scope: str = SCOPE_ELITE,
|
||
score: int,
|
||
commodity_id: str,
|
||
now: datetime | None = None,
|
||
) -> dict[str, Any]:
|
||
"""不做任何锁定，预估「现在能分出几个、要花多少钱」（文档 §3.8）。
|
||
|
||
返回的四个数在写死的口径下**互斥且可加**：
|
||
`zero_topup + (after_topup - zero_topup) + blocked 三项 = 池外无关账号` 不计。
|
||
"""
|
||
if scope not in _SCOPE_COLUMNS:
|
||
raise ValueError(f"暂不支持的作用域: {scope}")
|
||
points_column_name, opened_column_name = _SCOPE_COLUMNS[scope]
|
||
points_column = getattr(Account, points_column_name)
|
||
opened_column = getattr(Account, opened_column_name) if opened_column_name else None
|
||
since = _now(now) - timedelta(days=EXCHANGE_COOLDOWN_DAYS)
|
||
|
||
valid_base = [
|
||
Account.status == "active",
|
||
Account.ck_check_status == "valid",
|
||
Account.cookie.isnot(None),
|
||
Account.cookie != "",
|
||
Account.auto_managed.is_(True),
|
||
pool_membership_clause(),
|
||
]
|
||
not_redeemed = not_(latest_exchange_exists_clause(commodity_id, since))
|
||
|
||
def count(*extra) -> int:
|
||
return int(
|
||
db.execute(
|
||
select(func.count(Account.id)).where(and_(*valid_base, *extra))
|
||
).scalar()
|
||
or 0
|
||
)
|
||
|
||
opened_ok = (
|
||
[opened_column.is_(True), points_column.isnot(None)]
|
||
if opened_column is not None
|
||
else [points_column.isnot(None)]
|
||
)
|
||
eligible = count(*opened_ok, not_redeemed, not_(active_lease_exists_clause()))
|
||
zero_topup = count(
|
||
*opened_ok,
|
||
not_redeemed,
|
||
not_(active_lease_exists_clause()),
|
||
points_column >= score,
|
||
)
|
||
# 已兑过（含仍在冷却期）的账号里，只看「手册已开通」的那部分 ——
|
||
# 未开通的账号无论如何都用不了，混进这个数会让人误以为是"冷却挡住了"。
|
||
redeemed = count(*opened_ok, latest_exchange_exists_clause(commodity_id, since))
|
||
not_opened = (
|
||
count(points_column.isnot(None), not_(opened_column.is_(True)))
|
||
if opened_column is not None
|
||
else 0
|
||
)
|
||
leased = count(*opened_ok, not_redeemed, active_lease_exists_clause())
|
||
|
||
gap_expr = func.greatest(0, score - func.coalesce(points_column, 0))
|
||
total_gap = int(
|
||
db.execute(
|
||
select(func.coalesce(func.sum(gap_expr), 0)).where(
|
||
and_(
|
||
*valid_base,
|
||
*opened_ok,
|
||
not_redeemed,
|
||
not_(active_lease_exists_clause()),
|
||
)
|
||
)
|
||
).scalar()
|
||
or 0
|
||
)
|
||
est_gifts = (total_gap + POINTS_PER_GIFT - 1) // POINTS_PER_GIFT
|
||
|
||
return {
|
||
"scope": scope,
|
||
"commodity_id": commodity_id,
|
||
"score": int(score),
|
||
"zero_topup": zero_topup,
|
||
"after_topup": eligible,
|
||
"leased": leased,
|
||
"blocked": {
|
||
f"{scope}_not_opened": not_opened,
|
||
"redeemed_within_cooldown": redeemed,
|
||
},
|
||
"cost": {"est_gifts": est_gifts, "est_yuan": est_gifts},
|
||
"cooldown_days": EXCHANGE_COOLDOWN_DAYS,
|
||
}
|
||
|
||
|
||
def active_lease_for_account(db: Session, account_id: int) -> AccountLease | None:
|
||
return (
|
||
db.query(AccountLease)
|
||
.filter(
|
||
AccountLease.account_id == account_id,
|
||
AccountLease.status == "active",
|
||
)
|
||
.first()
|
||
)
|
||
|
||
|
||
def leased_account_ids(
|
||
db: Session, *, holder_user_id: int, include_expired: bool = False
|
||
) -> list[int]:
|
||
"""当前用户持有的有效租约对应的账号 ID。"""
|
||
query = db.query(AccountLease.account_id).filter(
|
||
AccountLease.holder_user_id == holder_user_id,
|
||
AccountLease.status == "active",
|
||
)
|
||
if not include_expired:
|
||
query = query.filter(AccountLease.expires_at > _now())
|
||
return [int(row[0]) for row in query.all()]
|
||
|
||
|
||
def lease_accounts(
|
||
db: Session,
|
||
*,
|
||
holder_user_id: int,
|
||
candidates: Sequence[Candidate],
|
||
scope: str = SCOPE_ELITE,
|
||
commodity_id: str = "",
|
||
batch_id: str = "",
|
||
quantity: int | None = None,
|
||
ttl_minutes: int = DEFAULT_LEASE_TTL_MINUTES,
|
||
now: datetime | None = None,
|
||
) -> list[AccountLease]:
|
||
"""按顺序领取租约，返回**真正领到**的那些。
|
||
|
||
并发抢号由 `uq_lease_active_account` 裁决：每个候选用一个 SAVEPOINT 包住，
|
||
撞唯一键就跳过它继续下一个 —— 已领到的不受影响（`rollback()` 会连坐，
|
||
所以必须用 `begin_nested()`）。
|
||
|
||
**本函数会 commit**：租约必须先落库，后续建任务才有依据。
|
||
"""
|
||
moment = _now(now)
|
||
expires_at = moment + timedelta(minutes=max(1, int(ttl_minutes)))
|
||
leased: list[AccountLease] = []
|
||
for candidate in candidates:
|
||
if quantity is not None and len(leased) >= int(quantity):
|
||
break
|
||
lease = AccountLease(
|
||
account_id=candidate.account_id,
|
||
holder_user_id=holder_user_id,
|
||
scope=scope,
|
||
purpose="exchange",
|
||
commodity_id=commodity_id,
|
||
batch_id=batch_id,
|
||
status="active",
|
||
leased_at=moment,
|
||
expires_at=expires_at,
|
||
release_reason="",
|
||
created_at=moment,
|
||
)
|
||
try:
|
||
with db.begin_nested():
|
||
db.add(lease)
|
||
db.flush()
|
||
except IntegrityError:
|
||
# 撞唯一键 = 别人刚抢走这个账号。这是**预期内**的并发结果，不是故障。
|
||
# SAVEPOINT 已经把这一条回滚掉了，直接换下一个候选。
|
||
if lease in db:
|
||
db.expunge(lease)
|
||
continue
|
||
leased.append(lease)
|
||
db.commit()
|
||
return leased
|
||
|
||
|
||
def bind_leases_to_batch(
|
||
db: Session, leases: Sequence[AccountLease], batch_id: str
|
||
) -> None:
|
||
"""把租约与任务批次关联起来（任务创建后才拿得到 batch_id）。"""
|
||
if not leases or not batch_id:
|
||
return
|
||
for lease in leases:
|
||
lease.batch_id = batch_id
|
||
db.commit()
|
||
|
||
|
||
def release_lease(
|
||
db: Session,
|
||
lease: AccountLease,
|
||
*,
|
||
reason: str,
|
||
now: datetime | None = None,
|
||
) -> bool:
|
||
"""释放单个租约；已经是终态则返回 False（幂等）。"""
|
||
if lease.status != "active":
|
||
return False
|
||
lease.status = "released"
|
||
lease.released_at = _now(now)
|
||
lease.release_reason = str(reason or "")[:64]
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def release_for_task(
|
||
db: Session,
|
||
task: DouyuTask,
|
||
*,
|
||
reason: str,
|
||
now: datetime | None = None,
|
||
) -> bool:
|
||
"""任务终态时释放它对应账号的租约（文档 §3.9 的即时释放）。
|
||
|
||
按 `(batch_id, account_id)` 定位：一个批次里每个账号各有一条租约，
|
||
不能因为某一条任务结束就把整批的租约都释放掉。
|
||
"""
|
||
if not task.batch_id:
|
||
return False
|
||
lease = (
|
||
db.query(AccountLease)
|
||
.filter(
|
||
AccountLease.batch_id == task.batch_id,
|
||
AccountLease.account_id == task.account_id,
|
||
AccountLease.status == "active",
|
||
)
|
||
.first()
|
||
)
|
||
if lease is None:
|
||
return False
|
||
return release_lease(db, lease, reason=reason, now=now)
|
||
|
||
|
||
def release_many(
|
||
db: Session,
|
||
leases: Sequence[AccountLease],
|
||
*,
|
||
reason: str,
|
||
now: datetime | None = None,
|
||
status: str = "released",
|
||
) -> int:
|
||
"""批量释放（补位失败、客服放弃、超管强制释放）。
|
||
|
||
`status` 只允许 `released`（人/流程主动放）与 `expired`（TTL 或孤儿批次兜底）——
|
||
两者都会让生成列变 NULL、从而释放唯一索引，区别只在于事后怎么看这件事。
|
||
"""
|
||
moment = _now(now)
|
||
count = 0
|
||
for lease in leases:
|
||
if lease.status != "active":
|
||
continue
|
||
lease.status = status
|
||
lease.released_at = moment
|
||
lease.release_reason = str(reason or "")[:64]
|
||
count += 1
|
||
if count:
|
||
db.commit()
|
||
return count
|
||
|
||
|
||
def release_batch(
|
||
db: Session, batch_id: str, *, reason: str, now: datetime | None = None
|
||
) -> int:
|
||
"""释放一个任务批次下的全部有效租约。"""
|
||
leases = (
|
||
db.query(AccountLease)
|
||
.filter(AccountLease.batch_id == batch_id, AccountLease.status == "active")
|
||
.all()
|
||
)
|
||
return release_many(db, leases, reason=reason, now=now)
|
||
|
||
|
||
def reap_expired(
|
||
db: Session, *, now: datetime | None = None, batch_size: int = 500
|
||
) -> int:
|
||
"""把过期租约置为 expired（兜底作业调用）。
|
||
|
||
只动**已过期**的：正在跑的任务即使超了 TTL 也先留着，
|
||
避免把还在执行中的账号放给第二个人 —— TTL 到期的第一批由客服/运营处理，
|
||
真正兜底的是「批次已不在内存」那条路径（见 reaper 作业）。
|
||
"""
|
||
moment = _now(now)
|
||
leases = (
|
||
db.query(AccountLease)
|
||
.filter(AccountLease.status == "active", AccountLease.expires_at <= moment)
|
||
.order_by(AccountLease.id.asc())
|
||
.limit(max(1, int(batch_size)))
|
||
.all()
|
||
)
|
||
return release_many(db, leases, reason="ttl_expired", now=moment, status="expired")
|
||
|
||
|
||
def orphan_leases(db: Session, *, active_batch_ids: set[str], limit: int = 500):
|
||
"""批次已不在内存注册表（进程重启）的有效租约 —— 交给 reaper 释放。"""
|
||
query = db.query(AccountLease).filter(AccountLease.status == "active")
|
||
if active_batch_ids:
|
||
query = query.filter(AccountLease.batch_id.notin_(list(active_batch_ids)))
|
||
return query.order_by(AccountLease.id.asc()).limit(max(1, int(limit))).all()
|
||
|
||
|
||
#: 任务还没跑完的状态（与 `douyu_service.DOUYU_ACTIVE_TASK_STATUSES` 同义）。
|
||
#: 这里写一份而不是 import，是为了不让纯函数模块反向依赖整条 runner 依赖链。
|
||
_ACTIVE_TASK_STATUSES = ("planned", "pending", "running")
|
||
|
||
|
||
def reap_missed_releases(db: Session, *, limit: int = 500) -> int:
|
||
"""回收「任务已经终态、租约却还挂着」的租约。
|
||
|
||
`_mark_task` 的即时释放是主路径，但进程被杀、任务行被保留策略删掉、
|
||
释放时抛异常，都会留下悬空租约。判据是**该批次已经没有未完成任务**：
|
||
|
||
- 批次里还有 `planned/pending/running` → 说明还在跑，不动它；
|
||
- `batch_id` 为空 → 租约刚建、任务还没建出来，不动（由 TTL 兜底）。
|
||
|
||
只动「没有在跑的任务」的租约，所以绝不会把正在执行的账号交给第二个人。
|
||
"""
|
||
active_batches = select(DouyuTask.batch_id).where(
|
||
DouyuTask.status.in_(_ACTIVE_TASK_STATUSES)
|
||
)
|
||
leases = (
|
||
db.query(AccountLease)
|
||
.filter(
|
||
AccountLease.status == "active",
|
||
AccountLease.batch_id != "",
|
||
AccountLease.batch_id.notin_(active_batches),
|
||
)
|
||
.order_by(AccountLease.id.asc())
|
||
.limit(max(1, int(limit)))
|
||
.all()
|
||
)
|
||
return release_many(db, leases, reason="task_finished_missed")
|
||
|
||
|
||
def recharge_amounts_for(gold_needed: int) -> list[int]:
|
||
"""把一笔充值需求拆成每笔不超过 800 元的订单（文档 §3.4.2）。
|
||
|
||
返回**空列表**表示不需要充值。单账号一件 1280 分的套装只需要 128 元，
|
||
正常不会拆单；一次补很多分才会走到这里。
|
||
|
||
>>> recharge_amounts_for(0)
|
||
[]
|
||
>>> recharge_amounts_for(98)
|
||
[98]
|
||
>>> recharge_amounts_for(1700)
|
||
[800, 800, 100]
|
||
"""
|
||
remaining = max(0, int(gold_needed))
|
||
orders: list[int] = []
|
||
while remaining > 0:
|
||
chunk = min(remaining, MAX_RECHARGE_AMOUNT)
|
||
orders.append(chunk)
|
||
remaining -= chunk
|
||
return orders
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SupplyStep:
|
||
"""补给链里的一步：任务类型 + 该任务自己的载荷。"""
|
||
|
||
task_type: str
|
||
payload: dict
|
||
|
||
|
||
def build_supply_chain(
|
||
candidate: Candidate,
|
||
*,
|
||
commodity_id: str,
|
||
exchange_task_type: str = "exchange_goods",
|
||
) -> list[SupplyStep]:
|
||
"""给一个 B 梯队账号排出「充值 → 赠令 → 兑换」的执行链（纯函数）。
|
||
|
||
设计要点（都是被现实逼出来的）：
|
||
|
||
1. **先查余额再决定要不要花钱**。链首固定放 `query_gold_balance` /`query_points`：
|
||
充值步骤带 `_skip_when` 闸门，依据的是**执行那一刻**账号上的鱼翅/积分；
|
||
不先刷新一次，闸门读的是旧值 —— 旧值偏低就会**多充真金白银**。
|
||
2. **缺多少充多少**，超过 800 元自动拆单（服务端另有硬校验）。
|
||
3. **赠令带 `points >= score` 闸门**：积分已经够了就别再送礼（送出去是钱）。
|
||
4. A 梯队（积分已够）**不生成补给步骤**，链上只有兑换一步。
|
||
|
||
执行顺序由「同一批次 + 并发 1」保证（`DouyuBatchRunner` 按任务 id 顺序取行）。
|
||
"""
|
||
if not candidate.needs_topup:
|
||
return [
|
||
SupplyStep(
|
||
exchange_task_type,
|
||
{"commodity_id": commodity_id, "num": 1},
|
||
)
|
||
]
|
||
|
||
# 检查手册开通状态与最新积分
|
||
status_step = (
|
||
SupplyStep("query_elite_status", {"require_opened": True})
|
||
if exchange_task_type == "exchange_goods"
|
||
else SupplyStep("query_points", {})
|
||
)
|
||
steps: list[SupplyStep] = [
|
||
status_step,
|
||
SupplyStep("query_gold_balance", {}),
|
||
]
|
||
# 充值：余额够就跳过（闸门在执行时求值，读的是刚刷新过的账号字段）
|
||
for amount in recharge_amounts_for(candidate.topup_gifts):
|
||
steps.append(
|
||
SupplyStep(
|
||
"create_gold_qr",
|
||
{
|
||
"amount": amount,
|
||
"_skip_when": {
|
||
"column": "gold_balance",
|
||
"op": ">=",
|
||
"value": candidate.topup_gifts,
|
||
},
|
||
},
|
||
)
|
||
)
|
||
steps.append(
|
||
SupplyStep(
|
||
"donate_elite_gift",
|
||
{
|
||
"gift_count": candidate.topup_gifts,
|
||
"_skip_when": {
|
||
"column": "points",
|
||
"op": ">=",
|
||
"value": candidate.score,
|
||
},
|
||
},
|
||
)
|
||
)
|
||
steps.append(
|
||
SupplyStep(exchange_task_type, {"commodity_id": commodity_id, "num": 1})
|
||
)
|
||
return steps
|
||
|
||
|
||
def validate_recharge_amount(amount: int | None) -> int:
|
||
"""校验单次充值金额（元）。
|
||
|
||
只做**上限**校验，不做下限 —— 下限由调用方按业务给默认值。
|
||
"""
|
||
value = int(amount or 0)
|
||
if value <= 0:
|
||
raise ValueError("充值金额必须大于 0")
|
||
if value > MAX_RECHARGE_AMOUNT:
|
||
raise ValueError(
|
||
f"单次充值不得超过 {MAX_RECHARGE_AMOUNT} 元，本次请求 {value} 元"
|
||
)
|
||
return value
|
||
|
||
|
||
def lease_status_summary(db: Session, *, holder_user_id: int) -> dict[str, int]:
|
||
"""给池工作台用的租约概览。"""
|
||
moment = _now()
|
||
base = db.query(AccountLease).filter(
|
||
AccountLease.holder_user_id == holder_user_id,
|
||
AccountLease.status == "active",
|
||
)
|
||
return {
|
||
"active": base.count(),
|
||
"expiring_soon": base.filter(
|
||
AccountLease.expires_at <= moment + timedelta(minutes=10)
|
||
).count(),
|
||
}
|
||
|
||
|
||
__all__ = [
|
||
"DEFAULT_LEASE_TTL_MINUTES",
|
||
"EXCHANGE_COOLDOWN_DAYS",
|
||
"MAX_RECHARGE_AMOUNT",
|
||
"POINTS_PER_GIFT",
|
||
"SUPPORTED_POOL_SCOPES",
|
||
"Candidate",
|
||
"SupplyStep",
|
||
"active_lease_for_account",
|
||
"bind_leases_to_batch",
|
||
"build_supply_chain",
|
||
"eligible_accounts",
|
||
"estimate_supply",
|
||
"lease_accounts",
|
||
"lease_status_summary",
|
||
"leased_account_ids",
|
||
"make_candidate",
|
||
"orphan_leases",
|
||
"rank_candidates",
|
||
"reap_expired",
|
||
"reap_missed_releases",
|
||
"recharge_amounts_for",
|
||
"release_batch",
|
||
"release_for_task",
|
||
"release_lease",
|
||
"release_many",
|
||
"topup_gifts_for",
|
||
"validate_recharge_amount",
|
||
]
|