Files
xiaoxia-saas/packages/domain/points_rules.py
T
xiaoxia c3e1ab62ed
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 1s
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 1s
CI/CD Pipeline / Integration Tests (pull_request) Failing after 1m57s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m26s
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Successful in 2m51s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 3m17s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 3m18s
CI/CD Pipeline / Validate - Style (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Security (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
AI Code Review / AI Code Review (pull_request) Has been cancelled
PR Automation / Auto Approve on CI Green (pull_request) Has been cancelled
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 30h47m23s
CI/CD Pipeline / PR Build Web Image (pull_request) Failing after 30h47m29s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 30h47m31s
CI/CD Pipeline / Check push changed paths (pull_request) Failing after 30h46m59s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 30h46m57s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 30h46m49s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 30h46m54s
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Failing after 30h46m54s
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Failing after 30h46m54s
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Failing after 30h46m55s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 30h46m55s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 30h46m49s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 30h46m56s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 30h46m57s
feat: add membership & points system backend (#1895)
- Add migration 008_membership_points.sql: users table new fields + 4 new tables
- Add domain objects: PointsAccount, PointsTransaction, PointsOrder, DailyUsageRecord
- Add repository ports & SQLAlchemy adapter implementations
- Add PointsService with balance/charge/refund/recharge/daily-usage logic
- Add points_rules configuration (9 scenes, packages, membership pricing)
- Add points_gate middleware decorator for AI feature entry points
- Add API routes: points balance/transactions/rules/packages/check/deduct/refund/recharge
- Add API routes: membership status, daily usage
- Add unit tests for points_rules (20 tests) and points_service (20 tests)
2026-09-15 08:35:55 +08:00

108 lines
3.7 KiB
Python

"""积分消耗规则配置 (#1895)"""
from __future__ import annotations
import math
# ============ 场景定义 ============
# 每个场景: base_points(基础积分), unit(计费单位), name(显示名称)
POINTS_SCENES: dict[str, dict] = {
"ai_voice": {"base_points": 1, "unit": "分钟", "name": "AI 配音"},
"ai_video": {
"base_points": 3,
"unit": "条",
"name": "智能混剪",
"extra_per_30s": 1,
},
"ai_digital_human": {"base_points": 15, "unit": "分钟", "name": "AI 数字人"},
"voice_clone_train": {"base_points": 0, "unit": "次", "name": "声音克隆训练"},
"voice_clone_synth": {"base_points": 1, "unit": "分钟", "name": "声音克隆合成"},
"douyin_extract": {"base_points": 1, "unit": "次", "name": "抖音链接提取"},
"ai_rewrite": {"base_points": 1, "unit": "次", "name": "AI 改写文案"},
"ai_title": {"base_points": 1, "unit": "次", "name": "AI 标题生成"},
"ai_cover": {"base_points": 1, "unit": "张", "name": "AI 封面生成"},
}
# 免费用户积分消耗上浮系数
FREE_USER_MULTIPLIER = 1.15
# ============ 积分包定义 ============
POINTS_PACKAGES: dict[str, dict] = {
"starter_pack": {"name": "体验包", "points": 100, "price_cents": 990},
"basic_pack": {"name": "基础包", "points": 500, "price_cents": 3900},
"pro_pack": {"name": "专业包", "points": 2000, "price_cents": 12900},
}
# ============ 会员定价 ============
MEMBERSHIP_PRICES: dict[str, dict] = {
"monthly": {"name": "月卡", "price_cents": 1990, "duration_days": 30},
"quarterly": {"name": "季卡", "price_cents": 3990, "duration_days": 90},
"yearly": {"name": "年卡", "price_cents": 15900, "duration_days": 365},
}
# 会员积分折扣(付费会员按此系数打折)
MEMBER_DISCOUNT: dict[str, float] = {
"monthly": 0.9,
"quarterly": 0.87,
"yearly": 0.8,
}
# 每日免费混剪次数(免费用户)
DAILY_FREE_CLIP_LIMIT = 2
def calculate_points_cost(
scene_key: str,
is_member: bool,
quantity: int = 1,
duration_minutes: float = 0,
member_type: str | None = None,
) -> int:
"""计算指定场景的积分消耗。
Args:
scene_key: 场景标识,如 "ai_voice"、"ai_video"
is_member: 是否付费会员
quantity: 数量(按次计费场景)
duration_minutes: 时长分钟数(按时长计费场景)
member_type: 会员类型 (monthly/quarterly/yearly),用于折扣
Returns:
实际消耗积分(已含免费用户 ×1.15 上浮或会员折扣)
Raises:
ValueError: 未知场景标识
"""
scene = POINTS_SCENES.get(scene_key)
if not scene:
raise ValueError(f"Unknown points scene: {scene_key}")
base = scene["base_points"]
if base == 0:
return 0
# —— 计算基础消耗 ——
unit = scene["unit"]
if unit == "分钟":
total_base = base * max(1, math.ceil(duration_minutes))
elif unit in ("条", "次", "张"):
total_base = base * quantity
# 混剪特殊逻辑:视频超过 30s 后每 +30s 额外加 1 积分
if scene_key == "ai_video" and duration_minutes > 0.5:
extra_segments = math.ceil((duration_minutes * 60 - 30) / 30)
if extra_segments > 0:
total_base += scene.get("extra_per_30s", 1) * extra_segments
else:
total_base = base
# —— 会员折扣 / 免费用户上浮 ——
if is_member and member_type and member_type in MEMBER_DISCOUNT:
total_base = max(1, math.floor(total_base * MEMBER_DISCOUNT[member_type]))
elif not is_member:
total_base = math.ceil(total_base * FREE_USER_MULTIPLIER)
return total_base