🎨代码优化

zhenxun/builtin_plugins/auto_update/_data_source.py

@@ -1,9 +1,8 @@
 import os
 import shutil
-import subprocess
 import tarfile
 import zipfile
-from pathlib import Path
+import subprocess
 
 from nonebot.adapters import Bot
 from nonebot.utils import run_sync
@@ -13,24 +12,28 @@
 from zhenxun.utils.platform import PlatformUtils
 
 from .config import (
- BACKUP_PATH,
- BASE_PATH,
  DEV_URL,
- DOWNLOAD_GZ_FILE,
- DOWNLOAD_ZIP_FILE,
  MAIN_URL,
- PYPROJECT_FILE,
- PYPROJECT_LOCK_FILE,
+ TMP_PATH,
+ BASE_PATH,
+ BACKUP_PATH,
  RELEASE_URL,
- REPLACE_FOLDERS,
  REQ_TXT_FILE,
- TMP_PATH,
  VERSION_FILE,
+ PYPROJECT_FILE,
+ REPLACE_FOLDERS,
+ BASE_PATH_STRING,
+ DOWNLOAD_GZ_FILE,
+ DOWNLOAD_ZIP_FILE,
+ PYPROJECT_LOCK_FILE,
+ REQ_TXT_FILE_STRING,
+ PYPROJECT_FILE_STRING,
+ PYPROJECT_LOCK_FILE_STRING,
 )
 
 
 def install_requirement():
- requirement_path = (Path() / "requirements.txt").absolute()
+ requirement_path = (REQ_TXT_FILE).absolute()
 
  if not requirement_path.exists():
  logger.debug(
@@ -41,8 +44,7 @@ def install_requirement():
  result = subprocess.run(
  ["pip", "install", "-r", str(requirement_path)],
  check=True,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
+ capture_output=True,
  text=True,
  )
  logger.debug(f"成功安装真寻依赖，日志:\n{result.stdout}", "插件管理")
@@ -67,32 +69,32 @@ def _file_handle(latest_version: str | None):
  tf = zipfile.ZipFile(DOWNLOAD_ZIP_FILE)
  tf.extractall(TMP_PATH)
  logger.debug("解压文件压缩包完成...", "检查更新")
- download_file_path = (
- TMP_PATH / [x for x in os.listdir(TMP_PATH) if (TMP_PATH / x).is_dir()][0]
+ download_file_path = TMP_PATH / next(
+ x for x in os.listdir(TMP_PATH) if (TMP_PATH / x).is_dir()
  )
- _pyproject = download_file_path / "pyproject.toml"
- _lock_file = download_file_path / "poetry.lock"
- _req_file = download_file_path / "requirements.txt"
- extract_path = download_file_path / "zhenxun"
+ _pyproject = download_file_path / PYPROJECT_FILE_STRING
+ _lock_file = download_file_path / PYPROJECT_LOCK_FILE_STRING
+ _req_file = download_file_path / REQ_TXT_FILE_STRING
+ extract_path = download_file_path / BASE_PATH_STRING
  target_path = BASE_PATH
  if PYPROJECT_FILE.exists():
  logger.debug(f"移除备份文件: {PYPROJECT_FILE}", "检查更新")
- shutil.move(PYPROJECT_FILE, BACKUP_PATH / "pyproject.toml")
+ shutil.move(PYPROJECT_FILE, BACKUP_PATH / PYPROJECT_FILE_STRING)
  if PYPROJECT_LOCK_FILE.exists():
  logger.debug(f"移除备份文件: {PYPROJECT_LOCK_FILE}", "检查更新")
- shutil.move(PYPROJECT_LOCK_FILE, BACKUP_PATH / "poetry.lock")
+ shutil.move(PYPROJECT_LOCK_FILE, BACKUP_PATH / PYPROJECT_LOCK_FILE_STRING)
  if REQ_TXT_FILE.exists():
  logger.debug(f"移除备份文件: {REQ_TXT_FILE}", "检查更新")
- shutil.move(REQ_TXT_FILE, BACKUP_PATH / "requirements.txt")
+ shutil.move(REQ_TXT_FILE, BACKUP_PATH / REQ_TXT_FILE_STRING)
  if _pyproject.exists():
  logger.debug("移动文件: pyproject.toml", "检查更新")
- shutil.move(_pyproject, Path() / "pyproject.toml")
+ shutil.move(_pyproject, PYPROJECT_FILE)
  if _lock_file.exists():
  logger.debug("移动文件: poetry.lock", "检查更新")
- shutil.move(_lock_file, Path() / "poetry.lock")
+ shutil.move(_lock_file, PYPROJECT_LOCK_FILE)
  if _req_file.exists():
  logger.debug("移动文件: requirements.txt", "检查更新")
- shutil.move(_req_file, Path() / "requirements.txt")
+ shutil.move(_req_file, REQ_TXT_FILE)
  for folder in REPLACE_FOLDERS:
  """移动指定文件夹"""
  _dir = BASE_PATH / folder
@@ -132,7 +134,6 @@ def _file_handle(latest_version: str | None):
 
 
 class UpdateManage:
-
  @classmethod
  async def check_version(cls) -> str:
  """检查更新版本
@@ -144,7 +145,13 @@ async def check_version(cls) -> str:
  data = await cls.__get_latest_data()
  if not data:
  return "检查更新获取版本失败..."
- return f"检测到当前版本更新\n当前版本：{cur_version}\n最新版本：{data.get('name')}\n创建日期：{data.get('created_at')}\n更新内容：\n{data.get('body')}"
+ return (
+ "检测到当前版本更新\n"
+ f"当前版本：{cur_version}\n"
+ f"最新版本：{data.get('name')}\n"
+ f"创建日期：{data.get('created_at')}\n"
+ f"更新内容：\n{data.get('body')}"
+ )
 
  @classmethod
  async def update(cls, bot: Bot, user_id: str, version_type: str) -> str | None:
@@ -158,8 +165,10 @@ async def update(cls, bot: Bot, user_id: str, version_type: str) -> str | None:
  返回:
  str | None: 返回消息
  """
- logger.info(f"开始下载真寻最新版文件....", "检查更新")
+ logger.info("开始下载真寻最新版文件....", "检查更新")
  cur_version = cls.__get_version()
+ url = None
+ new_version = None
  if version_type == "dev":
  url = DEV_URL
  new_version = await cls.__get_version_from_branch("dev")
@@ -197,7 +206,11 @@ async def update(cls, bot: Bot, user_id: str, version_type: str) -> str | None:
  if await AsyncHttpx.download_file(url, download_file):
  logger.debug("下载真寻最新版文件完成...", "检查更新")
  await _file_handle(new_version)
- return f"版本更新完成\n版本: {cur_version} -> {new_version}\n请重新启动真寻以完成更新!"
+ return (
+ f"版本更新完成\n"
+ f"版本: {cur_version} -> {new_version}\n"
+ "请重新启动真寻以完成更新!"
+ )
  else:
  logger.debug("下载真寻最新版文件失败...", "检查更新")
  return None
@@ -211,8 +224,7 @@ def __get_version(cls) -> str:
  """
  _version = "v0.0.0"
  if VERSION_FILE.exists():
- text = VERSION_FILE.open(encoding="utf8").readline()
- if text:
+ if text := VERSION_FILE.open(encoding="utf8").readline():
  _version = text.split(":")[-1].strip()
  return _version
 
@@ -231,7 +243,7 @@ async def __get_latest_data(cls) -> dict:
  except TimeoutError:
  pass
  except Exception as e:
- logger.error(f"检查更新真寻获取版本失败", e=e)
+ logger.error("检查更新真寻获取版本失败", e=e)
  return {}
 
  @classmethod

zhenxun/builtin_plugins/auto_update/config.py

@@ -6,14 +6,18 @@
 MAIN_URL = "https://ghproxy.cc/https://github.com/HibiKier/zhenxun_bot/archive/refs/heads/main.zip"
 RELEASE_URL = "https://api.github.com/repos/HibiKier/zhenxun_bot/releases/latest"
 
-
-VERSION_FILE = Path() / "__version__"
-
-PYPROJECT_FILE = Path() / "pyproject.toml"
-PYPROJECT_LOCK_FILE = Path() / "poetry.lock"
-REQ_TXT_FILE = Path() / "requirements.txt"
-
-BASE_PATH = Path() / "zhenxun"
+VERSION_FILE_STRING = "__version__"
+VERSION_FILE = Path() / VERSION_FILE_STRING
+
+PYPROJECT_FILE_STRING = "pyproject.toml"
+PYPROJECT_FILE = Path() / PYPROJECT_FILE_STRING
+PYPROJECT_LOCK_FILE_STRING = "poetry.lock"
+PYPROJECT_LOCK_FILE = Path() / PYPROJECT_LOCK_FILE_STRING
+REQ_TXT_FILE_STRING = "requirements.txt"
+REQ_TXT_FILE = Path() / REQ_TXT_FILE_STRING
+
+BASE_PATH_STRING = "zhenxun"
+BASE_PATH = Path() / BASE_PATH_STRING
 
 TMP_PATH = TEMP_PATH / "auto_update"
 

zhenxun/builtin_plugins/web_ui/__init__.py

@@ -2,26 +2,26 @@
 import secrets
 
 import nonebot
-from fastapi import APIRouter, FastAPI
-from nonebot.log import default_filter, default_format
+from fastapi import FastAPI, APIRouter
 from nonebot.plugin import PluginMetadata
+from nonebot.log import default_filter, default_format
 
-from zhenxun.configs.config import Config as gConfig
-from zhenxun.configs.utils import PluginExtraData, RegisterConfig
-from zhenxun.services.log import logger, logger_
 from zhenxun.utils.enum import PluginType
+from zhenxun.services.log import logger, logger_
+from zhenxun.configs.config import Config as gConfig
+from zhenxun.configs.utils import RegisterConfig, PluginExtraData
 
+from .public import init_public
+from .auth import router as auth_router
 from .api.logs import router as ws_log_routes
 from .api.logs.log_manager import LOG_STORAGE
-from .api.tabs.database import router as database_router
 from .api.tabs.main import router as main_router
-from .api.tabs.main import ws_router as status_routes
 from .api.tabs.manage import router as manage_router
+from .api.tabs.system import router as system_router
+from .api.tabs.main import ws_router as status_routes
+from .api.tabs.database import router as database_router
 from .api.tabs.manage.chat import ws_router as chat_routes
 from .api.tabs.plugin_manage import router as plugin_router
-from .api.tabs.system import router as system_router
-from .auth import router as auth_router
-from .public import init_public
 
 __plugin_meta__ = PluginMetadata(
  name="WebUi",
@@ -32,7 +32,7 @@
  author="HibiKier",
  version="0.1",
  plugin_type=PluginType.HIDDEN,
- Configs=[
+ configs=[
  RegisterConfig(
  module="web-ui",
  key="username",
@@ -57,7 +57,7 @@
  type=str,
  default_value=None,
  ),
-  ],
+ ],
  ).dict(),
 )
 
@@ -98,7 +98,7 @@ async def log_sink(message: str):
  logger.warning("Web Ui log_sink", e=e)
  if not loop:
  loop = asyncio.new_event_loop()
- loop.create_task(LOG_STORAGE.add(message.rstrip("\n")))
+ loop.create_task(LOG_STORAGE.add(message.rstrip("\n"))) # noqa: RUF006
 
  logger_.add(
  log_sink, colorize=True, filter=default_filter, format=default_format

zhenxun/plugins/word_bank/_data_source.py

@@ -1,13 +1,13 @@
 from nonebot_plugin_alconna import At
-from nonebot_plugin_alconna import At as alcAt
 from nonebot_plugin_alconna import Image
-from nonebot_plugin_alconna import Image as alcImage
+from nonebot_plugin_alconna import At as alcAt
 from nonebot_plugin_alconna import Text as alcText
-from nonebot_plugin_alconna import UniMessage, UniMsg
+from nonebot_plugin_alconna import Image as alcImage
+from nonebot_plugin_alconna import UniMsg, UniMessage
 
-from zhenxun.plugins.word_bank._config import ScopeType
-from zhenxun.utils.image_utils import ImageTemplate
 from zhenxun.utils.message import MessageUtils
+from zhenxun.utils.image_utils import ImageTemplate
+from zhenxun.plugins.word_bank._config import ScopeType
 
 from ._model import WordBank
 
@@ -42,9 +42,9 @@ def get_problem(message: UniMsg) -> str:
  problem = ""
  a, b = True, True
  for msg in message:
- if isinstance(msg, alcText) or isinstance(msg, str):
+ if isinstance(msg, alcText | str):
  msg = str(msg)
- if "问" in str(msg) and a:
+ if "问" in msg and a:
  a = False
  split_text = msg.split("问")
  if len(split_text) > 1:
@@ -53,7 +53,7 @@ def get_problem(message: UniMsg) -> str:
  if "答" in problem:
  b = False
  problem = problem.split("答")[0]
- elif "答" in msg and b:
+ elif "答" in msg:
  b = False
  # problem += "答".join(msg.split("答")[:-1])
  problem += msg.split("答")[0]
@@ -78,7 +78,7 @@ def get_answer(message: UniMsg) -> UniMessage | None:
  index = 0
  for msg in message:
  index += 1
- if isinstance(msg, alcText) or isinstance(msg, str):
+ if isinstance(msg, alcText | str):
  msg = str(msg)
  if "答" in msg:
  answer += "答".join(msg.split("答")[1:])
@@ -90,7 +90,6 @@ def get_answer(message: UniMsg) -> UniMessage | None:
 
 
 class WordBankManage:
-
  @classmethod
  async def update_word(
  cls,
@@ -175,10 +174,14 @@ async def __word_handle(
  )
  if not _problem_list:
  return problem, ""
- if await WordBank.delete_group_problem(problem, group_id, aid, word_scope): # type: ignore
- return "删除词条成功!", ""
- return "词条不存在", ""
- if handle_type == "update":
+ return (
+ ("删除词条成功!", "")
+ if await WordBank.delete_group_problem(
+ problem, group_id, aid, word_scope
+ )
+ else ("词条不存在", "")
+ )
+ elif handle_type == "update":
  old_problem = await WordBank.update_group_problem(
  problem, replace_problem, group_id, word_scope=word_scope
  )
@@ -187,7 +190,10 @@ async def __word_handle(
 
  @classmethod
  async def __get_problem_str(
- cls, idx: int, group_id: str | None = None, word_scope: ScopeType = ScopeType.GROUP
+ cls,
+ idx: int,
+ group_id: str | None = None,
+ word_scope: ScopeType = ScopeType.GROUP,
  ) -> tuple[str, int]:
  """通过id获取问题字符串
 
@@ -222,7 +228,7 @@ async def show_word(
  word_scope: 词条范围
  index: 指定回答下标
  """
- if problem or index != None:
+ if problem or index is not None:
  msg_list = []
  problem, _problem_list = await WordBank.get_problem_all_answer(
  problem, # type: ignore