164 lines
7.2 KiB
Python
164 lines
7.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
from datetime import datetime
|
|
from typing import Dict, Optional
|
|
|
|
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)
|