mirror of
https://github.com/PaiGramTeam/GramCore.git
synced 2024-11-21 13:48:20 +00:00
63 lines
1.7 KiB
Python
63 lines
1.7 KiB
Python
from typing import List
|
|
|
|
from gram_core.base_service import BaseService
|
|
from gram_core.dependence.redisdb import RedisDB
|
|
|
|
__all__ = (
|
|
"GroupBanCache",
|
|
"GroupIgnoreCache",
|
|
"GroupUpdateCache",
|
|
)
|
|
|
|
|
|
class IGroupCache:
|
|
qname = ""
|
|
|
|
def __init__(self, redis: RedisDB):
|
|
self.client = redis.client
|
|
|
|
async def ismember(self, chat_id: int) -> bool:
|
|
return await self.client.sismember(self.qname, chat_id)
|
|
|
|
async def get_all(self) -> List[int]:
|
|
return [int(str_data) for str_data in await self.client.smembers(self.qname)]
|
|
|
|
async def set(self, chat_id: int) -> bool:
|
|
return await self.client.sadd(self.qname, chat_id)
|
|
|
|
async def remove(self, chat_id: int) -> bool:
|
|
return await self.client.srem(self.qname, chat_id)
|
|
|
|
async def remove_all(self) -> bool:
|
|
return await self.client.delete(self.qname)
|
|
|
|
|
|
class GroupBanCache(BaseService.Component, IGroupCache):
|
|
qname = "groups:ban"
|
|
|
|
|
|
class GroupIgnoreCache(BaseService.Component, IGroupCache):
|
|
qname = "groups:ignore"
|
|
|
|
|
|
class GroupUpdateCache(BaseService.Component):
|
|
def __init__(self, redis: RedisDB):
|
|
self.client = redis.client
|
|
self.qname = "groups:update"
|
|
self.expire = 24 * 60 * 60 # 24h
|
|
|
|
def get_key(self, chat_id: int) -> str:
|
|
return f"{self.qname}:{chat_id}"
|
|
|
|
async def set(self, chat_id: int) -> bool:
|
|
key = self.get_key(chat_id)
|
|
return await self.client.set(key, 1, ex=self.expire)
|
|
|
|
async def remove(self, chat_id: int) -> bool:
|
|
key = self.get_key(chat_id)
|
|
return await self.client.delete(key)
|
|
|
|
async def is_member(self, chat_id: int) -> bool:
|
|
key = self.get_key(chat_id)
|
|
return await self.client.exists(key)
|