import redis import datetime import calendar ? r = redis.Redis( host="127.0.0.1", port=6379, db=1, password="" ) ? ? def check_input(func): def wrapper(*args, **kwargs): if not isinstance(args[1], int): raise ValueError(f"User_id must be int, and your input is {type(args[1])}") return func(*args, **kwargs) ? return wrapper ? ? class RedisCheckIn: ? _private_key = "_check_in_" ? def __init__(self): pass ? @check_input def sign(self, user_id: int) -> int: # 用户签到 today = str(datetime.datetime.now())[:10] return r.setbit(self._private_key+today, user_id, 1) ? @check_input def sign_status(self, user_id: int) -> int: # 用户今日签到状态 today = str(datetime.datetime.now())[:10] return r.getbit(self._private_key+today, user_id) ? @check_input def week_sign_status(self, user_id: int) -> list: # 求出这个周的签到状况 now = datetime.datetime.now() # 周一是1 周日是7 now.weekday()则是周一是0,周日是6 weekday = now.isoweekday() with r.pipeline(transaction=False) as p: for d in range(weekday): check_day = str(now-datetime.timedelta(days=1)*d)[:10] p.getbit(self._private_key + check_day, user_id) data = p.execute()[::-1] # 比如周三的时候我们只查3次getbit,然后剩下补0 data.extend([0] * (7 - len(data))) return data ? @check_input def month_sing_status(self, user_id: int) -> list: # 求出这个月的某个用户签到状况 now = datetime.datetime.now() day = now.day with r.pipeline(transaction=False) as p: for d in range(day): check_day = str(now - datetime.timedelta(days=1) * d)[:10] p.getbit(self._private_key + check_day, user_id) data = p.execute()[::-1] # 获取当月天数,还没到的天数补0 month_range = calendar.monthrange(now.year, now.month) data.extend([0] * (month_range[1] - len(data))) return data ? @check_input def week_sign_num(self, user_id: int) -> int: # 求出这个周的签到次数 return sum(self.week_sign_status(user_id)) ? @check_input def month_sign_num(self, user_id: int) -> int: # 求出这个月的签到次数 return sum(self.month_sing_status(user_id)) ? @check_input def today_sign_all_num(self) -> int: # 求出当天有多少用户签到 today = str(datetime.datetime.now())[:10] return r.bitcount(self._private_key + today) ? ? redis_sign_in = RedisCheckIn()