mirror of
https://github.com/PaiGramTeam/PamGram.git
synced 2024-11-27 00:05:03 +00:00
059bcd5e70
* 🔧 使用 dotenv 重构 config
默认配置从 config.json 移动到 config.py 中。如果要覆盖默认配置，在根目录创建
.env 文件按照 .env.example 的例子编辑。
这个方案的优点是：
* 支持写注释
* 以后如果新增配置项，如果用默认值就可以，不需要修改 .env 文件
* 如果通过 serverless、docker 或者 k8s 部署，方便不用修改文件，直接注入环境变量
修改配置
61 lines
2.2 KiB
Python
61 lines
2.2 KiB
Python
from typing import List
|
|
|
|
from pymysql import IntegrityError
|
|
from telegram import Bot
|
|
|
|
from config import config
|
|
from logger import Log
|
|
from .cache import BotAdminCache, GroupAdminCache
|
|
from .repositories import BotAdminRepository
|
|
|
|
|
|
class BotAdminService:
|
|
def __init__(self, repository: BotAdminRepository, cache: BotAdminCache):
|
|
self._repository = repository
|
|
self._cache = cache
|
|
|
|
async def get_admin_list(self) -> List[int]:
|
|
admin_list = await self._cache.get_list()
|
|
if len(admin_list) == 0:
|
|
admin_list = await self._repository.get_all_user_id()
|
|
for config_admin in config.admins:
|
|
admin_list.append(config_admin["user_id"])
|
|
await self._cache.set_list(admin_list)
|
|
return admin_list
|
|
|
|
async def add_admin(self, user_id: int) -> bool:
|
|
try:
|
|
await self._repository.add_by_user_id(user_id)
|
|
except IntegrityError as error:
|
|
Log.warning(f"{user_id} 已经存在数据库 \n", error)
|
|
admin_list = await self._repository.get_all_user_id()
|
|
for config_admin in config.admins:
|
|
admin_list.append(config_admin["user_id"])
|
|
await self._cache.set_list(admin_list)
|
|
return True
|
|
|
|
async def delete_admin(self, user_id: int) -> bool:
|
|
try:
|
|
await self._repository.delete_by_user_id(user_id)
|
|
except ValueError:
|
|
return False
|
|
admin_list = await self._repository.get_all_user_id()
|
|
for config_admin in config.admins:
|
|
admin_list.append(config_admin["user_id"])
|
|
await self._cache.set_list(admin_list)
|
|
return True
|
|
|
|
|
|
class GroupAdminService:
|
|
def __init__(self, cache: GroupAdminCache):
|
|
self._cache = cache
|
|
|
|
async def get_admins(self, bot: Bot, chat_id: int, extra_user: List[int]) -> List[int]:
|
|
admin_id_list = await self._cache.get_chat_admin(chat_id)
|
|
if len(admin_id_list) == 0:
|
|
admin_list = await bot.get_chat_administrators(chat_id)
|
|
admin_id_list = [admin.user.id for admin in admin_list]
|
|
await self._cache.set_chat_admin(chat_id, admin_id_list)
|
|
admin_id_list += extra_user
|
|
return admin_id_list
|