232 lines
7.8 KiB
Python
232 lines
7.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
签到接口
|
||
功能：每日签到、签到状态查询、签到日历
|
||
"""
|
||
|
||
from fastapi import APIRouter, HTTPException, Depends
|
||
from pydantic import BaseModel
|
||
from datetime import date, datetime, timedelta
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import func
|
||
from app.db import get_db
|
||
from app.models.user import User
|
||
from app.models.business import CheckInConfig, VipConfig, CheckInRecord, PointsHistory
|
||
from app.core.security import get_current_user
|
||
|
||
router = APIRouter()
|
||
|
||
# ==================== 签到功能 ====================
|
||
|
||
@router.post("/checkin/daily")
|
||
async def daily_checkin(db: Session = Depends(get_db), current_username: str = Depends(get_current_user)):
|
||
"""每日签到"""
|
||
# 1. 获取用户信息
|
||
user = db.query(User).filter(User.username == current_username).first()
|
||
if not user:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
|
||
# 2. 检查今日是否已签到
|
||
today = date.today()
|
||
if user.last_check_in_date == today:
|
||
raise HTTPException(status_code=400, detail="今日已签到，明天再来吧")
|
||
|
||
# 3. 计算连续天数
|
||
consecutive_days = user.consecutive_check_in if user.consecutive_check_in else 0
|
||
total_days = user.total_check_in_days if user.total_check_in_days else 0
|
||
|
||
yesterday = today - timedelta(days=1)
|
||
if user.last_check_in_date == yesterday:
|
||
# 连续签到
|
||
consecutive_days += 1
|
||
else:
|
||
# 中断了，重新开始
|
||
consecutive_days = 1
|
||
|
||
total_days += 1
|
||
|
||
# 4. 从配置表获取奖励
|
||
reward_config = db.query(CheckInConfig)\
|
||
.filter(CheckInConfig.consecutive_days <= consecutive_days, CheckInConfig.enabled == True)\
|
||
.order_by(CheckInConfig.consecutive_days.desc())\
|
||
.first()
|
||
|
||
if not reward_config:
|
||
# 如果没有配置，默认给10积分
|
||
base_points = 10
|
||
bonus_points = 0
|
||
total_points = 10
|
||
else:
|
||
base_points = reward_config.base_points
|
||
bonus_points = reward_config.bonus_points
|
||
total_points = reward_config.total_points
|
||
|
||
# 5. 应用VIP倍数
|
||
vip_multiplier = 1.0
|
||
if user.vip_type and user.vip_type in ['vip', 'svip']:
|
||
vip_config = db.query(VipConfig).filter(VipConfig.vip_type == user.vip_type).first()
|
||
if vip_config:
|
||
vip_multiplier = float(vip_config.points_multiplier)
|
||
|
||
points_earned = int(total_points * vip_multiplier)
|
||
current_points = user.points if user.points else 0
|
||
new_balance = current_points + points_earned
|
||
|
||
# 6. 更新用户数据
|
||
user.points = new_balance
|
||
user.total_check_in_days = total_days
|
||
user.consecutive_check_in = consecutive_days
|
||
user.last_check_in_date = today
|
||
|
||
# 7. 记录签到记录
|
||
checkin_record = CheckInRecord(
|
||
user_id=user.id,
|
||
username=current_username,
|
||
check_in_date=today,
|
||
points_earned=points_earned,
|
||
consecutive_days=consecutive_days,
|
||
vip_multiplier=vip_multiplier
|
||
)
|
||
db.add(checkin_record)
|
||
|
||
# 8. 记录积分历史
|
||
points_history = PointsHistory(
|
||
user_id=user.id,
|
||
username=current_username,
|
||
type='checkin',
|
||
amount=points_earned,
|
||
balance=new_balance,
|
||
description=f"每日签到奖励（连续{consecutive_days}天）"
|
||
)
|
||
db.add(points_history)
|
||
|
||
db.commit()
|
||
|
||
# 9. 返回结果
|
||
return {
|
||
"code": 200,
|
||
"data": {
|
||
"success": True,
|
||
"points_earned": points_earned,
|
||
"base_points": base_points,
|
||
"bonus_points": bonus_points,
|
||
"vip_multiplier": vip_multiplier,
|
||
"consecutive_days": consecutive_days,
|
||
"total_check_in_days": total_days,
|
||
"total_points": new_balance,
|
||
"message": f"签到成功！连续签到{consecutive_days}天，获得{points_earned}积分"
|
||
}
|
||
}
|
||
|
||
@router.get("/checkin/config")
|
||
async def get_checkin_config(db: Session = Depends(get_db)):
|
||
"""获取签到奖励配置（公开）"""
|
||
configs = db.query(CheckInConfig)\
|
||
.filter(CheckInConfig.enabled == True)\
|
||
.order_by(CheckInConfig.consecutive_days)\
|
||
.all()
|
||
|
||
result = []
|
||
for config in configs:
|
||
result.append({
|
||
"consecutive_days": config.consecutive_days,
|
||
"base_points": config.base_points,
|
||
"bonus_points": config.bonus_points,
|
||
"total_points": config.total_points
|
||
})
|
||
|
||
return {
|
||
"code": 200,
|
||
"data": result
|
||
}
|
||
|
||
@router.get("/checkin/status")
|
||
async def get_checkin_status(db: Session = Depends(get_db), current_username: str = Depends(get_current_user)):
|
||
"""获取签到状态"""
|
||
user = db.query(User).filter(User.username == current_username).first()
|
||
if not user:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
|
||
today = date.today()
|
||
today_checked = (user.last_check_in_date == today)
|
||
|
||
return {
|
||
"code": 200,
|
||
"data": {
|
||
"today_checked": today_checked,
|
||
"consecutive_days": user.consecutive_check_in if user.consecutive_check_in else 0,
|
||
"total_days": user.total_check_in_days if user.total_check_in_days else 0,
|
||
"total_points": user.points if user.points else 0,
|
||
"last_check_in_date": user.last_check_in_date.isoformat() if user.last_check_in_date else None
|
||
}
|
||
}
|
||
|
||
@router.get("/checkin/calendar/{year}/{month}")
|
||
async def get_checkin_calendar(year: int, month: int, db: Session = Depends(get_db), current_username: str = Depends(get_current_user)):
|
||
"""获取签到日历"""
|
||
user = db.query(User).filter(User.username == current_username).first()
|
||
if not user:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
|
||
# 使用 SQLAlchemy 提取日期部分
|
||
# 注意：SQLite 和 MySQL 的日期函数不同
|
||
# 为了兼容性，这里我们查询该月范围内的所有记录，然后在 Python 中处理
|
||
|
||
import calendar
|
||
_, last_day = calendar.monthrange(year, month)
|
||
start_date = date(year, month, 1)
|
||
end_date = date(year, month, last_day)
|
||
|
||
records = db.query(CheckInRecord)\
|
||
.filter(
|
||
CheckInRecord.user_id == user.id,
|
||
CheckInRecord.check_in_date >= start_date,
|
||
CheckInRecord.check_in_date <= end_date
|
||
).all()
|
||
|
||
checked_dates = [record.check_in_date.day for record in records]
|
||
|
||
return {
|
||
"code": 200,
|
||
"data": {
|
||
"year": year,
|
||
"month": month,
|
||
"checked_dates": checked_dates
|
||
}
|
||
}
|
||
|
||
@router.get("/checkin/history")
|
||
async def get_checkin_history(page: int = 1, limit: int = 10, db: Session = Depends(get_db), current_username: str = Depends(get_current_user)):
|
||
"""签到记录"""
|
||
user = db.query(User).filter(User.username == current_username).first()
|
||
if not user:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
|
||
offset = (page - 1) * limit
|
||
|
||
total = db.query(func.count(CheckInRecord.id)).filter(CheckInRecord.user_id == user.id).scalar()
|
||
|
||
records = db.query(CheckInRecord)\
|
||
.filter(CheckInRecord.user_id == user.id)\
|
||
.order_by(CheckInRecord.check_in_date.desc())\
|
||
.offset(offset)\
|
||
.limit(limit)\
|
||
.all()
|
||
|
||
result_records = []
|
||
for record in records:
|
||
result_records.append({
|
||
"check_in_date": record.check_in_date.isoformat(),
|
||
"points_earned": record.points_earned,
|
||
"consecutive_days": record.consecutive_days,
|
||
"created_at": record.created_at.isoformat() if record.created_at else None
|
||
})
|
||
|
||
return {
|
||
"code": 200,
|
||
"data": {
|
||
"total": total,
|
||
"records": result_records
|
||
}
|
||
}
|