7.6 KiB

Raw Permalink Blame History

Credit Service 完整实现

日期：2026-01-28
类型：Feature Implementation
影响范围：Credit Service 完整代码实现

实施内容

基于 docs/requirements/backend/04-services/user/credit-service.md 文档，完成 Credit Service 的完整代码实现。

1. Model 层（5 个模型）

文件位置：server/app/models/credit/

✅ transaction.py - CreditTransaction 模型（积分流水表）
✅ consumption_log.py - CreditConsumptionLog 模型（消耗记录表）
✅ package.py - CreditPackage 模型（套餐表）
✅ pricing.py - CreditPricing 模型（定价配置表）
✅ gift.py - CreditGift 模型（赠送记录表）
✅ __init__.py - 模型导出

技术特性：

UUID v7 主键（应用层生成）
SMALLINT 存储枚举值
无物理外键约束
使用 primaryjoin 配置 Relationship
完整的索引定义

2. Repository 层

文件位置：server/app/repositories/credit_repository.py

实现方法：

用户相关：get_user(), user_exists()
交易记录：create_transaction(), get_transactions(), count_transactions()
消耗记录：create_consumption_log(), get_consumption_log(), get_consumption_logs(), count_consumption_logs(), get_expired_pending_consumptions(), get_consumption_logs_by_status()
定价配置：get_pricing()
套餐管理：get_active_packages(), get_package()
赠送记录：create_gift()

3. Service 层

文件位置：server/app/services/credit_service.py

枚举定义（6 个）：

PaymentMethod - 支付方式（1=微信, 2=支付宝）
PaymentStatus - 支付状态（1=待支付, 2=已支付, 3=失败, 4=已退款, 5=已取消）
TransactionType - 交易类型（1=充值, 2=消耗, 3=退款, 4=赠送, 5=过期, 6=管理员调整）
FeatureType - 功能类型（1-10，涵盖图片、视频、文本等）
TaskStatus - 任务状态（1=待处理, 2=处理中, 3=成功, 4=失败, 5=已退款, 6=已过期）
GiftType - 赠送类型（1=注册, 2=邀请, 3=活动, 4=补偿, 5=管理员）

核心方法：

积分查询：get_balance(), get_transactions(), get_consumption_logs()
积分操作：add_credits(), consume_credits(), refund_credits(), confirm_consumption()
积分定价：calculate_credits()
超时管理：expire_pending_consumptions()
积分套餐：get_packages(), get_package()
积分赠送：gift_credits()

4. Schema 层

文件位置：server/app/schemas/credit.py

响应模型：

CreditBalanceResponse - 余额响应
CreditTransactionResponse - 交易响应
CreditConsumptionResponse - 消耗响应
CreditPackageResponse - 套餐响应
CalculateCreditsResponse - 计算积分响应

请求模型：

CalculateCreditsRequest - 计算积分请求
ConsumeCreditsRequest - 消耗积分请求
ConfirmConsumptionRequest - 确认消耗请求
RefundCreditsRequest - 退款请求
GiftCreditsRequest - 赠送积分请求

特性：

使用 camelCase 别名（前端友好）
枚举值自动转换为字符串
完整的类型提示

5. API 路由层

文件位置：server/app/api/v1/credits.py

端点列表（10 个）：

GET /credits/balance - 查询积分余额
GET /credits/transactions - 查询积分流水
GET /credits/consumption - 查询消耗记录
POST /credits/calculate - 计算所需积分
POST /credits/consume - 消耗积分（预扣）
POST /credits/consumption/{consumption_id}/confirm - 确认消耗
POST /credits/consumption/{consumption_id}/refund - 退还积分
GET /credits/packages - 获取充值套餐列表
GET /credits/packages/{package_id} - 获取套餐详情
POST /credits/gift - 赠送积分（管理员）

特性：

统一的 SuccessResponse 响应格式
分页支持（PaginatedResponse）
完整的依赖注入
用户认证保护

6. Celery 任务层

文件位置：server/app/tasks/credit_tasks.py

定时任务：

expire_pending_consumptions() - 超时自动退款（每小时执行）
cleanup_old_transactions() - 清理旧交易记录（每天执行）

7. 数据库迁移

文件位置：server/alembic/versions/20260128_2200_create_credit_service_tables.py

迁移内容：

创建 5 个表（credit_transactions, credit_consumption_logs, credit_packages, credit_pricing, credit_gifts）
创建 30+ 个索引（单列、组合、条件索引）
添加表级和列级注释
创建 updated_at 触发器

技术规范遵循

1. 数据库设计

✅ UUID v7 主键：所有表使用 UUID v7 作为主键，应用层生成
✅ SMALLINT 枚举：所有枚举值使用 SMALLINT 存储
✅ 无物理外键：禁止使用 FOREIGN KEY 约束，应用层验证
✅ 完整索引：单列、组合、条件索引覆盖所有查询场景
✅ 完整注释：所有表和列都有中文注释，时间字段标注 UTC

2. 代码规范

✅ 异步优先：所有数据库操作使用 async/await
✅ 类型提示：完整的 Python 类型提示
✅ Repository 模式：数据访问层与业务逻辑层分离
✅ 枚举转换：提供 to_name/from_name 方法
✅ 错误处理：使用自定义异常（NotFoundError, ValidationError, InsufficientCreditsError）

3. API 设计

✅ 统一响应：使用 SuccessResponse 包装
✅ camelCase：前端友好的字段命名
✅ 分页支持：PaginatedResponse 统一分页格式
✅ 认证保护：所有端点使用 get_current_user 依赖

文件清单

新建文件（11 个）

Model 层：

server/app/models/credit/__init__.py
server/app/models/credit/transaction.py
server/app/models/credit/consumption_log.py
server/app/models/credit/package.py
server/app/models/credit/pricing.py
server/app/models/credit/gift.py

Repository 层：

server/app/repositories/credit_repository.py

Service 层：

server/app/services/credit_service.py

Schema 层：

server/app/schemas/credit.py

API 层：

server/app/api/v1/credits.py

任务层：

server/app/tasks/credit_tasks.py

迁移脚本：

server/alembic/versions/20260128_2200_create_credit_service_tables.py

下一步工作

1. 数据库迁移

# 在 Docker 容器中执行迁移
docker exec jointo-server-app alembic upgrade head

2. 注册 API 路由

在 server/app/api/v1/__init__.py 中注册 credits 路由：

from app.api.v1.credits import router as credits_router

api_router.include_router(credits_router)

3. 配置 Celery Beat

在 server/app/core/celery_app.py 中添加定时任务配置：

from celery.schedules import crontab

celery_app.conf.beat_schedule = {
    'expire-pending-consumptions': {
        'task': 'app.tasks.credit_tasks.expire_pending_consumptions',
        'schedule': crontab(minute=0),  # 每小时
    },
    'cleanup-old-transactions': {
        'task': 'app.tasks.credit_tasks.cleanup_old_transactions',
        'schedule': crontab(hour=2, minute=0),  # 每天凌晨 2 点
    },
}

4. 测试

单元测试：Service 层方法测试
集成测试：API 端点测试
性能测试：查询性能验证

5. 文档

API 文档：Swagger/OpenAPI 自动生成
使用指南：开发者文档
部署文档：生产环境配置

7.6 KiB Raw Permalink Blame History