xsl
76a0502ea3
- claim_task_bonus 在幂等快捷路径前做完成凭据校验：report_feedback 查当日 rp_feedback、first_report 查 succeeded rp_report，无凭据 40311 且不打流水；sign_in/share_report 不校验（产品决策：签到行为即领取，分享内测期接受端侧自报，hard_cap 锁死刷量收益） - task_bonus_rules 默认值与 .env.example 同步产品定稿（产品方案 §3.2，2026-09-11）：sign_in 对话+10/day、report_feedback 发型+2/day、share_report 换装+3/day、first_report 重生成+1/lifetime；下架 invite_friend/profile_complete - billing/repo 新增只读凭据探测 has_feedback_between/has_completed_report（import-linter 禁 billing→report，直查 shared.models 的 rp_* 表，永远只读） - 测试：test_task_bonus_eligibility 5 用例（40311 两分支/凭据后 200/40910/40410/零流水） 偏差说明：test_task_bonus 与 test_task_bonus_rules 断言数值随定稿更新（发放/幂等/封顶语义未动）；quota.py docstring「占位值」说法顺手更正。
119 lines
4.0 KiB
Python
Executable File
119 lines
4.0 KiB
Python
Executable File
"""Data access for the billing domain. **Only this module may touch bl_* tables.**
|
||
|
||
架构文档 §3.2 硬性规则 1: 每个域只能通过对方暴露的 service 接口调用，禁止直接
|
||
查对方的表。import-linter enforces it (contract "no cross-domain repository access").
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from packages.shared.models import Ledger, Quota, QuotaRule, RpFeedback, RpReport
|
||
|
||
|
||
async def get_rule(session: AsyncSession, quota_key: str) -> QuotaRule | None:
|
||
return await session.get(QuotaRule, quota_key)
|
||
|
||
|
||
async def lock_or_create_quota(
|
||
session: AsyncSession,
|
||
*,
|
||
user_id: int,
|
||
quota_key: str,
|
||
period_key: str,
|
||
limit_value: int,
|
||
) -> Quota:
|
||
"""Get the period row with a row lock, creating it on first use.
|
||
|
||
``ON CONFLICT DO NOTHING`` + ``SELECT ... FOR UPDATE`` makes concurrent first
|
||
requests for the same period safe without an advisory lock.
|
||
"""
|
||
await session.execute(
|
||
pg_insert(Quota)
|
||
.values(
|
||
user_id=user_id,
|
||
quota_key=quota_key,
|
||
period_key=period_key,
|
||
used=0,
|
||
limit_value=limit_value,
|
||
bonus_granted=0,
|
||
)
|
||
.on_conflict_do_nothing(index_elements=["user_id", "quota_key", "period_key"])
|
||
)
|
||
result = await session.execute(
|
||
select(Quota)
|
||
.where(
|
||
Quota.user_id == user_id,
|
||
Quota.quota_key == quota_key,
|
||
Quota.period_key == period_key,
|
||
)
|
||
.with_for_update()
|
||
)
|
||
row = result.scalar_one()
|
||
return row
|
||
|
||
|
||
async def get_ledger_by_idempotency_key(session: AsyncSession, key: str) -> Ledger | None:
|
||
result = await session.execute(select(Ledger).where(Ledger.idempotency_key == key))
|
||
return result.scalar_one_or_none()
|
||
|
||
|
||
async def get_ledger(session: AsyncSession, tx_id: int) -> Ledger | None:
|
||
return await session.get(Ledger, tx_id)
|
||
|
||
|
||
async def list_ledger(session: AsyncSession, user_id: int, limit: int = 50) -> list[Ledger]:
|
||
result = await session.execute(
|
||
select(Ledger)
|
||
.where(Ledger.user_id == user_id)
|
||
.order_by(Ledger.created_at.desc(), Ledger.tx_id.desc())
|
||
.limit(limit)
|
||
)
|
||
return list(result.scalars().all())
|
||
|
||
|
||
def add_ledger(session: AsyncSession, ledger: Ledger) -> Ledger:
|
||
session.add(ledger)
|
||
return ledger
|
||
|
||
|
||
# -- 任务完成凭据探测（T2b，rp_* 只读） --------------------------------------
|
||
# 分层契约（pyproject import-linter layers）禁止 packages.billing 依赖
|
||
# packages.report，而 claim 完成校验又必须读 rp_feedback/rp_report 的服务端
|
||
# 可信信号 —— 直接查 shared.models 的只读 exists 探测是唯一合规通道。
|
||
# 纪律：billing 对 rp_* 永远只读，绝不写入；「完成判定」语义仍归 report 域。
|
||
|
||
#: rp_report.status 的「已完成」取值：与 packages/report/repo.py 的
|
||
#: STATUS_SUCCEEDED 对齐（同上，分层契约不允许 import 该常量，故复制字面量）。
|
||
_REPORT_STATUS_SUCCEEDED = "succeeded"
|
||
|
||
|
||
async def has_feedback_between(
|
||
session: AsyncSession, *, user_id: int, start: datetime, end: datetime
|
||
) -> bool:
|
||
"""``[start, end)`` 内该用户是否存在反馈行（report_feedback 完成凭据）。"""
|
||
result = await session.execute(
|
||
select(RpFeedback.feedback_id)
|
||
.where(
|
||
RpFeedback.user_id == user_id,
|
||
RpFeedback.created_at >= start,
|
||
RpFeedback.created_at < end,
|
||
)
|
||
.limit(1)
|
||
)
|
||
return result.first() is not None
|
||
|
||
|
||
async def has_completed_report(session: AsyncSession, *, user_id: int) -> bool:
|
||
"""该用户是否存在已完成报告（first_report 完成凭据）。"""
|
||
result = await session.execute(
|
||
select(RpReport.report_id)
|
||
.where(RpReport.user_id == user_id, RpReport.status == _REPORT_STATUS_SUCCEEDED)
|
||
.limit(1)
|
||
)
|
||
return result.first() is not None
|