223 lines
9.5 KiB
Python
223 lines
9.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
from datetime import datetime
|
|
from typing import Dict, Optional, List
|
|
|
|
from db.base import BaseDBOperator
|
|
from db.connection import DBConnectionManager
|
|
|
|
|
|
class SignInDB(BaseDBOperator):
|
|
"""签到系统相关数据库操作"""
|
|
|
|
def __init__(self, db_manager: DBConnectionManager):
|
|
super().__init__(db_manager)
|
|
|
|
def initialize_table(self) -> bool:
|
|
"""初始化签到表"""
|
|
sql = """
|
|
CREATE TABLE IF NOT EXISTS t_sign_record (
|
|
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
|
wx_id VARCHAR(100) NOT NULL,
|
|
group_id VARCHAR(100) NOT NULL,
|
|
wx_nick_name VARCHAR(100) NOT NULL,
|
|
points INT DEFAULT 0,
|
|
sign_stat DATETIME,
|
|
signin_streak INT DEFAULT 0,
|
|
create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
UNIQUE KEY unique_sign (wx_id, group_id)
|
|
)
|
|
"""
|
|
return self.execute_update(sql)
|
|
|
|
def get_user_record(self, wx_id: str, group_id: str) -> Optional[Dict]:
|
|
"""获取用户签到记录"""
|
|
sql = """
|
|
SELECT wx_id, group_id, wx_nick_name, points, sign_stat, signin_streak,last_sign_date,previous_streak
|
|
FROM t_sign_record
|
|
WHERE wx_id = %s AND group_id = %s
|
|
"""
|
|
return self.execute_query(sql, (wx_id, group_id), fetch_one=True)
|
|
|
|
def update_sign_record(self, wx_id: str, group_id: str, wx_nick_name: str,
|
|
points_to_add: int, sign_time: datetime, streak: int) -> bool:
|
|
"""更新签到记录"""
|
|
sql = """
|
|
UPDATE t_sign_record
|
|
SET wx_nick_name = %s, points = points + %s,
|
|
sign_stat = %s, signin_streak = %s,
|
|
update_time = %s
|
|
WHERE wx_id = %s AND group_id = %s
|
|
"""
|
|
params = (wx_nick_name, points_to_add, sign_time, streak, sign_time, wx_id, group_id)
|
|
return self.execute_update(sql, params)
|
|
|
|
def create_sign_record(self, wx_id: str, group_id: str, wx_nick_name: str,
|
|
points: int, sign_time: datetime, streak: int) -> bool:
|
|
"""创建签到记录"""
|
|
sql = """
|
|
INSERT INTO t_sign_record
|
|
(wx_id, group_id, wx_nick_name, points, sign_stat, signin_streak)
|
|
VALUES (%s, %s, %s, %s, %s, %s)
|
|
"""
|
|
params = (wx_id, group_id, wx_nick_name, points, sign_time, streak)
|
|
return self.execute_update(sql, params)
|
|
|
|
def update_sign_record_with_last_date(self, wx_id: str, group_id: str, wx_nick_name: str,
|
|
points_to_add: int, sign_time: datetime, streak: int,
|
|
last_sign_date: datetime) -> bool:
|
|
"""更新签到记录，包括上次签到日期"""
|
|
sql = """
|
|
UPDATE t_sign_record
|
|
SET wx_nick_name = %s, points = points + %s,
|
|
sign_stat = %s, signin_streak = %s,
|
|
last_sign_date = %s, update_time = %s
|
|
WHERE wx_id = %s AND group_id = %s
|
|
"""
|
|
params = (wx_nick_name, points_to_add,
|
|
sign_time, streak,
|
|
last_sign_date, datetime.now(),
|
|
wx_id, group_id)
|
|
self.execute_update(sql, params)
|
|
|
|
def create_sign_record_with_last_date(self, wx_id: str, group_id: str, wx_nick_name: str,
|
|
points: int, sign_time: datetime, streak: int,
|
|
last_sign_date: datetime) -> bool:
|
|
"""创建新的签到记录，包括上次签到日期"""
|
|
sql = """
|
|
INSERT INTO t_sign_record
|
|
(wx_id, group_id, wx_nick_name, points, sign_stat, signin_streak, last_sign_date, create_time, update_time)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
"""
|
|
params = (
|
|
wx_id, group_id, wx_nick_name, points, sign_time, streak, last_sign_date, datetime.now(), datetime.now())
|
|
self.execute_update(sql, params)
|
|
|
|
def update_makeup_sign(self, wx_id: str, group_id: str, wx_nick_name: str,
|
|
points_to_add: int, sign_time: datetime, streak: int, last_sign_date: datetime) -> bool:
|
|
"""更新补签记录，专门处理今天已签到的情况"""
|
|
sql = """
|
|
UPDATE t_sign_record
|
|
SET wx_nick_name = %s, points = points + %s,
|
|
signin_streak = %s, last_sign_date = %s,
|
|
update_time = %s
|
|
WHERE wx_id = %s AND group_id = %s
|
|
"""
|
|
params = (wx_nick_name, points_to_add,
|
|
streak, last_sign_date,
|
|
datetime.now(),
|
|
wx_id, group_id
|
|
)
|
|
self.execute_update(sql, params)
|
|
|
|
def update_sign_record_with_previous_streak(self, wx_id: str, group_id: str, wx_nick_name: str,
|
|
points_to_add: int, sign_time: datetime, streak: int,
|
|
last_sign_date: datetime, previous_streak: int) -> bool:
|
|
"""更新签到记录，包括上次签到日期和断签前连签天数"""
|
|
sql = """
|
|
UPDATE t_sign_record
|
|
SET wx_nick_name = %s, points = points + %s,
|
|
sign_stat = %s, signin_streak = %s,
|
|
last_sign_date = %s, previous_streak = %s, update_time = %s
|
|
WHERE wx_id = %s AND group_id = %s
|
|
"""
|
|
params = (wx_nick_name, points_to_add,
|
|
sign_time, streak,
|
|
last_sign_date, previous_streak, datetime.now(),
|
|
wx_id, group_id)
|
|
return self.execute_update(sql, params)
|
|
|
|
def update_makeup_sign_with_streak_recovery(self, wx_id: str, group_id: str, wx_nick_name: str,
|
|
points_to_add: int, sign_time: datetime, streak: int,
|
|
last_sign_date: datetime, previous_streak: int) -> bool:
|
|
"""更新补签记录，支持连签恢复"""
|
|
sql = """
|
|
UPDATE t_sign_record
|
|
SET wx_nick_name = %s, points = points + %s,
|
|
signin_streak = %s, last_sign_date = %s,
|
|
previous_streak = %s, update_time = %s
|
|
WHERE wx_id = %s AND group_id = %s
|
|
"""
|
|
params = (wx_nick_name, points_to_add,
|
|
streak, last_sign_date,
|
|
previous_streak, datetime.now(),
|
|
wx_id, group_id
|
|
)
|
|
return self.execute_update(sql, params)
|
|
|
|
def update_sign_record_with_streak_recovery(self, wx_id: str, group_id: str, wx_nick_name: str,
|
|
points_to_add: int, sign_time: datetime, streak: int,
|
|
last_sign_date: datetime, previous_streak: int) -> bool:
|
|
"""更新签到记录，支持连签恢复"""
|
|
sql = """
|
|
UPDATE t_sign_record
|
|
SET wx_nick_name = %s, points = points + %s,
|
|
sign_stat = %s, signin_streak = %s,
|
|
last_sign_date = %s, previous_streak = %s, update_time = %s
|
|
WHERE wx_id = %s AND group_id = %s
|
|
"""
|
|
params = (wx_nick_name, points_to_add,
|
|
sign_time, streak,
|
|
last_sign_date, previous_streak, datetime.now(),
|
|
wx_id, group_id)
|
|
return self.execute_update(sql, params)
|
|
|
|
def create_sign_history(self, wx_id: str, group_id: str, sign_date: datetime,
|
|
sign_time: datetime, is_makeup: bool, points_earned: int,
|
|
streak_count: int) -> bool:
|
|
"""创建签到历史记录"""
|
|
sql = """
|
|
INSERT INTO t_sign_history
|
|
(wx_id, group_id, sign_date, sign_time, is_makeup, points_earned, streak_count)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s)
|
|
"""
|
|
params = (wx_id, group_id, sign_date.date(), sign_time,
|
|
1 if is_makeup else 0, points_earned, streak_count)
|
|
return self.execute_update(sql, params)
|
|
|
|
def get_user_sign_history(self, wx_id: str, group_id: str,
|
|
start_date: datetime = None,
|
|
end_date: datetime = None,
|
|
limit: int = 30) -> List[Dict]:
|
|
"""获取用户签到历史记录"""
|
|
sql = """
|
|
SELECT id, wx_id, group_id, sign_date, sign_time,
|
|
is_makeup, points_earned, streak_count, create_time
|
|
FROM t_sign_history
|
|
WHERE wx_id = %s AND group_id = %s
|
|
"""
|
|
params = [wx_id, group_id]
|
|
|
|
if start_date:
|
|
sql += " AND sign_date >= %s"
|
|
params.append(start_date.date())
|
|
if end_date:
|
|
sql += " AND sign_date <= %s"
|
|
params.append(end_date.date())
|
|
|
|
sql += " ORDER BY sign_date DESC, sign_time DESC LIMIT %s"
|
|
params.append(limit)
|
|
|
|
return self.execute_query(sql, tuple(params))
|
|
|
|
def get_user_sign_count(self, wx_id: str, group_id: str,
|
|
start_date: datetime = None,
|
|
end_date: datetime = None) -> int:
|
|
"""获取用户签到总次数"""
|
|
sql = """
|
|
SELECT COUNT(*) as count
|
|
FROM t_sign_history
|
|
WHERE wx_id = %s AND group_id = %s
|
|
"""
|
|
params = [wx_id, group_id]
|
|
|
|
if start_date:
|
|
sql += " AND sign_date >= %s"
|
|
params.append(start_date.date())
|
|
if end_date:
|
|
sql += " AND sign_date <= %s"
|
|
params.append(end_date.date())
|
|
|
|
result = self.execute_query(sql, tuple(params), fetch_one=True)
|
|
return result['count'] if result else 0
|