feat: Try read credential from json file when not set in config
This commit is contained in:
LWR
@@ -2,6 +2,7 @@ import asyncio
|
||||
import json
|
||||
import signal
|
||||
import sys
|
||||
from json import JSONDecodeError
|
||||
|
||||
from creart import create
|
||||
from graia.ariadne import Ariadne
|
||||
@@ -127,6 +128,28 @@ class StarBot:
|
||||
except asyncio.exceptions.TimeoutError:
|
||||
logger.warning("等待连接所有直播间超时, 请检查是否存在未连接成功的直播间")
|
||||
|
||||
# 未设置登录凭据时尝试从 JSON 文件中读取
|
||||
if config.get("SESSDATA") is None or config.get("BILI_JCT") is None or config.get("BUVID3") is None:
|
||||
logger.info("未设置 B 站登录凭据, 将尝试从 credential.json 文件中读取")
|
||||
try:
|
||||
with open("credential.json", "r", encoding="utf-8") as file:
|
||||
credential = json.loads(file.read())
|
||||
config.set("SESSDATA", credential["sessdata"])
|
||||
config.set("BILI_JCT", credential["bili_jct"])
|
||||
config.set("BUVID3", credential["buvid3"])
|
||||
logger.success("成功从 JSON 文件中读取了 B 站登录凭据")
|
||||
except FileNotFoundError:
|
||||
logger.warning("登录凭据 JSON 文件不存在")
|
||||
except UnicodeDecodeError:
|
||||
logger.warning("登录凭据 JSON 文件编码不正确, 请将其转换为 UTF-8 格式编码")
|
||||
except (JSONDecodeError, KeyError):
|
||||
logger.warning("登录凭据 JSON 文件格式不正确")
|
||||
except Exception as ex:
|
||||
logger.warning(f"读取登录凭据 JSON 文件异常 {ex}")
|
||||
|
||||
if config.get("SESSDATA") is None or config.get("BILI_JCT") is None or config.get("BUVID3") is None:
|
||||
logger.warning("读取 B 站登录凭据失败, 动态推送等部分功能将不可用")
|
||||
|
||||
# 启动动态推送模块
|
||||
asyncio.get_event_loop().create_task(dynamic_spider(self.__datasource))
|
||||
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import os
|
||||
import abc
|
||||
import asyncio
|
||||
import json
|
||||
@@ -195,19 +194,17 @@ class JsonDataSource(DataSource):
|
||||
"""
|
||||
从 JSON 字符串初始化的 Bot 推送配置数据源
|
||||
"""
|
||||
def __init__(self, json_file: Optional[str] = None, json_str: Optional[str] = None, credential_file: Optional[str] = None):
|
||||
def __init__(self, json_file: Optional[str] = None, json_str: Optional[str] = None):
|
||||
"""
|
||||
Args:
|
||||
json_file: JSON 文件路径，两个参数任选其一传入，全部传入优先使用 json_str
|
||||
json_str: JSON 配置字符串，两个参数任选其一传入，全部传入优先使用 json_str
|
||||
credential_file: B站凭据 JSON 文件路径，不填默认从运行目录下的credential.json中读取凭据
|
||||
"""
|
||||
super().__init__()
|
||||
self.__config = None
|
||||
|
||||
self.__json_file = json_file
|
||||
self.__json_str = json_str
|
||||
self.__credential_file = credential_file
|
||||
|
||||
async def load(self):
|
||||
"""
|
||||
@@ -252,15 +249,6 @@ class JsonDataSource(DataSource):
|
||||
super().format_data()
|
||||
logger.success(f"成功从 JSON 中导入了 {len(self.get_up_list())} 个 UP 主")
|
||||
|
||||
# 判断用户是否已通过config.set_credential设置凭据，若已设置则跳过设置
|
||||
if config.get("SESSDATA") is None or config.get("BILI_JCT") is None or config.get("BUVID3") is None:
|
||||
# 用户不填credential_file字段时默认运行目录下credential.json
|
||||
if self.__credential_file is None:
|
||||
# 判断运行目录下是否存在credential.json，若不存在则不调set_credential_from_json
|
||||
if os.path.exists("credential.json"):
|
||||
config.set_credential_from_json("credential.json")
|
||||
else:
|
||||
config.set_credential_from_json(self.__credential_file)
|
||||
|
||||
class MySQLDataSource(DataSource):
|
||||
"""
|
||||
|
||||
@@ -1,11 +0,0 @@
|
||||
from .ApiException import ApiException
|
||||
|
||||
|
||||
class CredentialFromJSONException(ApiException):
|
||||
"""
|
||||
从JSON文件读取Credential时发生的异常
|
||||
"""
|
||||
|
||||
def __init__(self, msg: str):
|
||||
super().__init__()
|
||||
self.msg = msg
|
||||
@@ -1,9 +1,4 @@
|
||||
import json
|
||||
from typing import Any, Optional
|
||||
|
||||
from loguru import logger
|
||||
|
||||
from ..exception.CredentialFromJSONException import CredentialFromJSONException
|
||||
from typing import Any
|
||||
|
||||
DEFAULT_CONFIG = {
|
||||
# 是否检测最新 StarBot 版本
|
||||
@@ -156,43 +151,6 @@ def set_credential(sessdata: str, bili_jct: str, buvid3: str):
|
||||
set("BUVID3", buvid3)
|
||||
|
||||
|
||||
def set_credential_from_json(json_file: Optional[str] = None, json_str: Optional[str] = None):
|
||||
"""
|
||||
从JSON读取B站credential
|
||||
|
||||
Args:
|
||||
json_file: JSON 文件路径，两个参数任选其一传入，全部传入优先使用 json_str
|
||||
json_str: JSON 配置字符串，两个参数任选其一传入，全部传入优先使用 json_str
|
||||
Raises:
|
||||
CredentialFromJSONException: JSON 格式错误或缺少必要参数
|
||||
"""
|
||||
if json_str is None:
|
||||
try:
|
||||
with open(json_file, "r", encoding="utf-8") as file:
|
||||
json_str = file.read()
|
||||
except FileNotFoundError:
|
||||
logger.error("B站凭据 JSON 文件不存在, 请检查文件路径是否正确")
|
||||
raise CredentialFromJSONException("B站凭据 JSON 文件不存在, 请检查文件路径是否正确")
|
||||
except UnicodeDecodeError:
|
||||
logger.error("B站凭据 JSON 文件编码不正确, 请将其转换为 UTF-8 格式编码后重试")
|
||||
raise CredentialFromJSONException("B站凭据 JSON 文件编码不正确, 请将其转换为 UTF-8 格式编码后重试")
|
||||
except Exception as ex:
|
||||
logger.error(f"读取B站凭据 JSON 文件异常 {ex}")
|
||||
raise CredentialFromJSONException(f"读取B站凭据 JSON 文件异常 {ex}")
|
||||
|
||||
try:
|
||||
config = json.loads(json_str)
|
||||
except Exception:
|
||||
logger.error("提供的B站凭据 JSON 字符串格式不正确")
|
||||
raise CredentialFromJSONException("提供的B站凭据 JSON 字符串格式不正确")
|
||||
|
||||
set("SESSDATA", config["sessdata"])
|
||||
set("BILI_JCT", config["bili_jct"])
|
||||
set("BUVID3", config["buvid3"])
|
||||
|
||||
logger.success("成功从JSON中导入了B站凭据")
|
||||
|
||||
|
||||
def get(key: str) -> Any:
|
||||
"""
|
||||
获取配置项的值
|
||||
|
||||
Reference in New Issue
Block a user