64 lines
2.5 KiB
Python
64 lines
2.5 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
|
|
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) |