PaiGramTeam/PaiGram
3
0
Fork 0
mirror of https://github.com/PaiGramTeam/PaiGram.git synced 2024-11-25 09:37:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

🐛 修复 redis 中无数据时的错误

Browse Source 
* 🐛 兼容 python3.8

* 🐛 修复 redis 中无数据时的错误
This commit is contained in:
Karako 2022-08-31 16:48:09 +08:00 committed by GitHub
parent c44952fd3d
commit 8cb3ff9ca7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 11 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
core/wiki/cache.py
View File

@ -23,7 +23,11 @@ class WikiCache:
async def get(self, key: str) -> dict: async def get(self, key: str) -> dict:
qname = f"{self.qname}:{key}" qname = f"{self.qname}:{key}"
# noinspection PyBroadException
try:
result = json.loads(await self.client.get(qname)) result = json.loads(await self.client.get(qname))
except Exception:
result = []
if isinstance(result, list) and len(result) > 0: if isinstance(result, list) and len(result) > 0:
for num, item in enumerate(result): for num, item in enumerate(result):
result[num] = json.loads(item) result[num] = json.loads(item)

4
core/wiki/services.py
View File

@ -71,10 +71,10 @@ class WikiService:
if self.first_run: if self.first_run:
weapon_dict = await self._cache.get("weapon") weapon_dict = await self._cache.get("weapon")
self._weapon_list = [Weapon.parse_obj(obj) for obj in weapon_dict] self._weapon_list = [Weapon.parse_obj(obj) for obj in weapon_dict]
self._weapon_name_list = await Weapon.get_name_list() self._weapon_name_list = [weapon.name for weapon in self._weapon_list]
characters_dict = await self._cache.get("characters") characters_dict = await self._cache.get("characters")
self._character_list = [Character.parse_obj(obj) for obj in characters_dict] self._character_list = [Character.parse_obj(obj) for obj in characters_dict]
self._character_name_list = await Character.get_name_list() self._character_name_list = [character.name for character in self._character_list]
self.first_run = False self.first_run = False

8
plugins/system/auth.py
View File

@ -1,7 +1,7 @@
import asyncio import asyncio
import random import random
import time import time
from typing import Tuple, Union, Dict from typing import Tuple, Union, Dict, List
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, ChatPermissions, ChatMember from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, ChatPermissions, ChatMember
from telegram.constants import ParseMode from telegram.constants import ParseMode
@ -36,7 +36,7 @@ class GroupJoiningVerification:
self.kick_time = 120 self.kick_time = 120
self.random = MT19937_Random() self.random = MT19937_Random()
self.lock = asyncio.Lock() self.lock = asyncio.Lock()
self.chat_administrators_cache: Dict[Union[str, int], Tuple[float, list[ChatMember]]] = {} self.chat_administrators_cache: Dict[Union[str, int], Tuple[float, List[ChatMember]]] = {}
self.is_refresh_quiz = False self.is_refresh_quiz = False
async def refresh_quiz(self): async def refresh_quiz(self):
@ -45,7 +45,7 @@ class GroupJoiningVerification:
await self.quiz_service.refresh_quiz() await self.quiz_service.refresh_quiz()
self.is_refresh_quiz = True self.is_refresh_quiz = True
async def get_chat_administrators(self, context: CallbackContext, chat_id: Union[str, int]) -> list[ChatMember]: async def get_chat_administrators(self, context: CallbackContext, chat_id: Union[str, int]) -> List[ChatMember]:
async with self.lock: async with self.lock:
cache_data = self.chat_administrators_cache.get(f"{chat_id}") cache_data = self.chat_administrators_cache.get(f"{chat_id}")
if cache_data is not None: if cache_data is not None:
@ -57,7 +57,7 @@ class GroupJoiningVerification:
return chat_administrators return chat_administrators
@staticmethod @staticmethod
def is_admin(chat_administrators: list[ChatMember], user_id: int) -> bool: def is_admin(chat_administrators: List[ChatMember], user_id: int) -> bool:
return any(admin.user.id == user_id for admin in chat_administrators) return any(admin.user.id == user_id for admin in chat_administrators)
async def kick_member_job(self, context: CallbackContext): async def kick_member_job(self, context: CallbackContext):