feature: 数据库连接与SQL集中管理,提高代码可读性
This commit is contained in:
77
db/sign_in_redis.py
Normal file
77
db/sign_in_redis.py
Normal file
@@ -0,0 +1,77 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from datetime import datetime
|
||||
from typing import Dict, Optional
|
||||
|
||||
from db.connection import DBConnectionManager
|
||||
|
||||
|
||||
class SignInRedisDB:
|
||||
"""签到系统Redis相关操作"""
|
||||
|
||||
def __init__(self, db_manager: DBConnectionManager):
|
||||
self.db_manager = db_manager
|
||||
self.prefix = "group:sign_in:"
|
||||
|
||||
def get_redis_connection(self):
|
||||
"""获取Redis连接"""
|
||||
return self.db_manager.get_redis_connection()
|
||||
|
||||
def load_signin_count(self) -> Dict[str, int]:
|
||||
"""加载签到人数数据"""
|
||||
signin_count = {}
|
||||
with self.get_redis_connection() as redis_client:
|
||||
keys = redis_client.keys(f'{self.prefix}*')
|
||||
for key in keys:
|
||||
if isinstance(key, bytes):
|
||||
key = key.decode('utf-8')
|
||||
if key == f'{self.prefix}last_reset_date':
|
||||
continue
|
||||
group_id = key.replace(self.prefix, '')
|
||||
count = redis_client.get(key)
|
||||
if count is not None:
|
||||
if isinstance(count, bytes):
|
||||
count = count.decode('utf-8')
|
||||
signin_count[group_id] = int(count)
|
||||
return signin_count
|
||||
|
||||
def save_signin_count(self, group_id: str, count: int) -> bool:
|
||||
"""保存签到人数"""
|
||||
try:
|
||||
with self.get_redis_connection() as redis_client:
|
||||
redis_client.set(f'{self.prefix}{group_id}', count)
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def get_last_reset_date(self) -> Optional[datetime.date]:
|
||||
"""获取最后重置日期"""
|
||||
with self.get_redis_connection() as redis_client:
|
||||
date_str = redis_client.get(f'{self.prefix}last_reset_date')
|
||||
if date_str:
|
||||
if isinstance(date_str, bytes):
|
||||
date_str = date_str.decode('utf-8')
|
||||
return datetime.strptime(date_str, '%Y-%m-%d').date()
|
||||
return None
|
||||
|
||||
def save_last_reset_date(self, date: datetime.date) -> bool:
|
||||
"""保存最后重置日期"""
|
||||
try:
|
||||
with self.get_redis_connection() as redis_client:
|
||||
redis_client.set(f'{self.prefix}last_reset_date', date.strftime('%Y-%m-%d'))
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def reset_daily_counts(self) -> bool:
|
||||
"""重置每日签到计数"""
|
||||
try:
|
||||
with self.get_redis_connection() as redis_client:
|
||||
keys = redis_client.keys(f'{self.prefix}*')
|
||||
for key in keys:
|
||||
if isinstance(key, bytes):
|
||||
key = key.decode('utf-8')
|
||||
if key != f'{self.prefix}last_reset_date':
|
||||
redis_client.delete(key)
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
Reference in New Issue
Block a user