mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2025-12-15 14:22:55 +08:00
🐛 修复web zip更新路径问题
This commit is contained in:
parent
1de1ded65c
commit
d7747e3b0e
@ -65,26 +65,29 @@ def init_mocked_api(mocked_api: MockRouter) -> None:
|
|||||||
tar_buffer = io.BytesIO()
|
tar_buffer = io.BytesIO()
|
||||||
zip_bytes = io.BytesIO()
|
zip_bytes = io.BytesIO()
|
||||||
|
|
||||||
from zhenxun.builtin_plugins.auto_update.config import (
|
from zhenxun.utils.manager.zhenxun_repo_manager import ZhenxunRepoManager
|
||||||
PYPROJECT_FILE_STRING,
|
|
||||||
PYPROJECT_LOCK_FILE_STRING,
|
|
||||||
REPLACE_FOLDERS,
|
|
||||||
REQ_TXT_FILE_STRING,
|
|
||||||
)
|
|
||||||
|
|
||||||
# 指定要添加到压缩文件中的文件路径列表
|
# 指定要添加到压缩文件中的文件路径列表
|
||||||
file_paths: list[str] = [
|
file_paths: list[str] = [
|
||||||
PYPROJECT_FILE_STRING,
|
ZhenxunRepoManager.config.PYPROJECT_FILE_STRING,
|
||||||
PYPROJECT_LOCK_FILE_STRING,
|
ZhenxunRepoManager.config.PYPROJECT_LOCK_FILE_STRING,
|
||||||
REQ_TXT_FILE_STRING,
|
ZhenxunRepoManager.config.REQUIREMENTS_FILE_STRING,
|
||||||
]
|
]
|
||||||
|
|
||||||
# 打开一个tarfile对象,写入到上面创建的BytesIO对象中
|
# 打开一个tarfile对象,写入到上面创建的BytesIO对象中
|
||||||
with tarfile.open(mode="w:gz", fileobj=tar_buffer) as tar:
|
with tarfile.open(mode="w:gz", fileobj=tar_buffer) as tar:
|
||||||
add_files_and_folders_to_tar(tar, file_paths, folders=REPLACE_FOLDERS)
|
add_files_and_folders_to_tar(
|
||||||
|
tar,
|
||||||
|
file_paths,
|
||||||
|
folders=ZhenxunRepoManager.config.ZHENXUN_BOT_UPDATE_FOLDERS,
|
||||||
|
)
|
||||||
|
|
||||||
with zipfile.ZipFile(zip_bytes, mode="w", compression=zipfile.ZIP_DEFLATED) as zipf:
|
with zipfile.ZipFile(zip_bytes, mode="w", compression=zipfile.ZIP_DEFLATED) as zipf:
|
||||||
add_files_and_folders_to_zip(zipf, file_paths, folders=REPLACE_FOLDERS)
|
add_files_and_folders_to_zip(
|
||||||
|
zipf,
|
||||||
|
file_paths,
|
||||||
|
folders=ZhenxunRepoManager.config.ZHENXUN_BOT_UPDATE_FOLDERS,
|
||||||
|
)
|
||||||
|
|
||||||
mocked_api.get(
|
mocked_api.get(
|
||||||
url="https://codeload.github.com/HibiKier/zhenxun_bot/legacy.tar.gz/refs/tags/v0.2.2",
|
url="https://codeload.github.com/HibiKier/zhenxun_bot/legacy.tar.gz/refs/tags/v0.2.2",
|
||||||
@ -199,52 +202,47 @@ def add_directory_to_tar(tarinfo, tar):
|
|||||||
|
|
||||||
|
|
||||||
def init_mocker_path(mocker: MockerFixture, tmp_path: Path):
|
def init_mocker_path(mocker: MockerFixture, tmp_path: Path):
|
||||||
from zhenxun.builtin_plugins.auto_update.config import (
|
from zhenxun.utils.manager.zhenxun_repo_manager import ZhenxunRepoManager
|
||||||
PYPROJECT_FILE_STRING,
|
|
||||||
PYPROJECT_LOCK_FILE_STRING,
|
|
||||||
REQ_TXT_FILE_STRING,
|
|
||||||
VERSION_FILE_STRING,
|
|
||||||
)
|
|
||||||
|
|
||||||
mocker.patch(
|
mocker.patch(
|
||||||
"zhenxun.utils.manager.virtual_env_package_manager.VirtualEnvPackageManager.install_requirement",
|
"zhenxun.utils.manager.virtual_env_package_manager.VirtualEnvPackageManager.install_requirement",
|
||||||
return_value=None,
|
return_value=None,
|
||||||
)
|
)
|
||||||
mock_tmp_path = mocker.patch(
|
mock_tmp_path = mocker.patch(
|
||||||
"zhenxun.builtin_plugins.auto_update._data_source.TMP_PATH",
|
"zhenxun.configs.path_config.TEMP_PATH",
|
||||||
new=tmp_path / "auto_update",
|
new=tmp_path / "auto_update",
|
||||||
)
|
)
|
||||||
mock_base_path = mocker.patch(
|
mock_base_path = mocker.patch(
|
||||||
"zhenxun.builtin_plugins.auto_update._data_source.BASE_PATH",
|
"zhenxun.utils.manager.zhenxun_repo_manager.ZhenxunRepoManager.config.ZHENXUN_BOT_CODE_PATH",
|
||||||
new=tmp_path / "zhenxun",
|
new=tmp_path / "zhenxun",
|
||||||
)
|
)
|
||||||
mock_backup_path = mocker.patch(
|
mock_backup_path = mocker.patch(
|
||||||
"zhenxun.builtin_plugins.auto_update._data_source.BACKUP_PATH",
|
"zhenxun.utils.manager.zhenxun_repo_manager.ZhenxunRepoManager.config.ZHENXUN_BOT_BACKUP_PATH",
|
||||||
new=tmp_path / "backup",
|
new=tmp_path / "backup",
|
||||||
)
|
)
|
||||||
mock_download_gz_file = mocker.patch(
|
mock_download_gz_file = mocker.patch(
|
||||||
"zhenxun.builtin_plugins.auto_update._data_source.DOWNLOAD_GZ_FILE",
|
"zhenxun.utils.manager.zhenxun_repo_manager.ZhenxunRepoManager.config.ZHENXUN_BOT_DOWNLOAD_FILE",
|
||||||
new=mock_tmp_path / "download_latest_file.tar.gz",
|
new=mock_tmp_path / "download_latest_file.tar.gz",
|
||||||
)
|
)
|
||||||
mock_download_zip_file = mocker.patch(
|
mock_download_zip_file = mocker.patch(
|
||||||
"zhenxun.builtin_plugins.auto_update._data_source.DOWNLOAD_ZIP_FILE",
|
"zhenxun.utils.manager.zhenxun_repo_manager.ZhenxunRepoManager.config.ZHENXUN_BOT_UNZIP_PATH",
|
||||||
new=mock_tmp_path / "download_latest_file.zip",
|
new=mock_tmp_path / "download_latest_file.zip",
|
||||||
)
|
)
|
||||||
mock_pyproject_file = mocker.patch(
|
mock_pyproject_file = mocker.patch(
|
||||||
"zhenxun.builtin_plugins.auto_update._data_source.PYPROJECT_FILE",
|
"zhenxun.utils.manager.zhenxun_repo_manager.ZhenxunRepoManager.config.PYPROJECT_FILE",
|
||||||
new=tmp_path / PYPROJECT_FILE_STRING,
|
new=tmp_path / ZhenxunRepoManager.config.PYPROJECT_FILE_STRING,
|
||||||
)
|
)
|
||||||
mock_pyproject_lock_file = mocker.patch(
|
mock_pyproject_lock_file = mocker.patch(
|
||||||
"zhenxun.builtin_plugins.auto_update._data_source.PYPROJECT_LOCK_FILE",
|
"zhenxun.utils.manager.zhenxun_repo_manager.ZhenxunRepoManager.config.PYPROJECT_LOCK_FILE",
|
||||||
new=tmp_path / PYPROJECT_LOCK_FILE_STRING,
|
new=tmp_path / ZhenxunRepoManager.config.PYPROJECT_LOCK_FILE_STRING,
|
||||||
)
|
)
|
||||||
mock_req_txt_file = mocker.patch(
|
mock_req_txt_file = mocker.patch(
|
||||||
"zhenxun.builtin_plugins.auto_update._data_source.REQ_TXT_FILE",
|
"zhenxun.utils.manager.zhenxun_repo_manager.ZhenxunRepoManager.config.REQUIREMENTS_FILE",
|
||||||
new=tmp_path / REQ_TXT_FILE_STRING,
|
new=tmp_path / ZhenxunRepoManager.config.REQUIREMENTS_FILE_STRING,
|
||||||
)
|
)
|
||||||
mock_version_file = mocker.patch(
|
mock_version_file = mocker.patch(
|
||||||
"zhenxun.builtin_plugins.auto_update._data_source.VERSION_FILE",
|
"zhenxun.utils.manager.zhenxun_repo_manager.ZhenxunRepoManager.config.ZHENXUN_BOT_VERSION_FILE_STRING",
|
||||||
new=tmp_path / VERSION_FILE_STRING,
|
new=tmp_path / ZhenxunRepoManager.config.ZHENXUN_BOT_VERSION_FILE_STRING,
|
||||||
)
|
)
|
||||||
open(mock_version_file, "w").write("__version__: v0.2.2")
|
open(mock_version_file, "w").write("__version__: v0.2.2")
|
||||||
return (
|
return (
|
||||||
@ -271,12 +269,7 @@ async def test_check_update_release(
|
|||||||
测试检查更新(release)
|
测试检查更新(release)
|
||||||
"""
|
"""
|
||||||
from zhenxun.builtin_plugins.auto_update import _matcher
|
from zhenxun.builtin_plugins.auto_update import _matcher
|
||||||
from zhenxun.builtin_plugins.auto_update.config import (
|
from zhenxun.utils.manager.zhenxun_repo_manager import ZhenxunRepoManager
|
||||||
PYPROJECT_FILE_STRING,
|
|
||||||
PYPROJECT_LOCK_FILE_STRING,
|
|
||||||
REPLACE_FOLDERS,
|
|
||||||
REQ_TXT_FILE_STRING,
|
|
||||||
)
|
|
||||||
|
|
||||||
init_mocked_api(mocked_api=mocked_api)
|
init_mocked_api(mocked_api=mocked_api)
|
||||||
|
|
||||||
@ -295,7 +288,7 @@ async def test_check_update_release(
|
|||||||
# 确保目录下有一个子目录,以便 os.listdir() 能返回一个目录名
|
# 确保目录下有一个子目录,以便 os.listdir() 能返回一个目录名
|
||||||
mock_tmp_path.mkdir(parents=True, exist_ok=True)
|
mock_tmp_path.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
for folder in REPLACE_FOLDERS:
|
for folder in ZhenxunRepoManager.config.ZHENXUN_BOT_UPDATE_FOLDERS:
|
||||||
(mock_base_path / folder).mkdir(parents=True, exist_ok=True)
|
(mock_base_path / folder).mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
mock_pyproject_file.write_bytes(b"")
|
mock_pyproject_file.write_bytes(b"")
|
||||||
@ -305,7 +298,7 @@ async def test_check_update_release(
|
|||||||
async with app.test_matcher(_matcher) as ctx:
|
async with app.test_matcher(_matcher) as ctx:
|
||||||
bot = create_bot(ctx)
|
bot = create_bot(ctx)
|
||||||
bot = cast(Bot, bot)
|
bot = cast(Bot, bot)
|
||||||
raw_message = "检查更新 release"
|
raw_message = "检查更新 release -z"
|
||||||
event = _v11_group_message_event(
|
event = _v11_group_message_event(
|
||||||
raw_message,
|
raw_message,
|
||||||
self_id=BotId.QQ_BOT,
|
self_id=BotId.QQ_BOT,
|
||||||
@ -324,14 +317,14 @@ async def test_check_update_release(
|
|||||||
ctx.should_call_api(
|
ctx.should_call_api(
|
||||||
"send_msg",
|
"send_msg",
|
||||||
_v11_private_message_send(
|
_v11_private_message_send(
|
||||||
message="检测真寻已更新,版本更新:v0.2.2\n开始更新...",
|
message="检测真寻已更新,当前版本:v0.2.2\n开始更新...",
|
||||||
user_id=UserId.SUPERUSER,
|
user_id=UserId.SUPERUSER,
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
ctx.should_call_send(
|
ctx.should_call_send(
|
||||||
event=event,
|
event=event,
|
||||||
message=Message(
|
message=Message(
|
||||||
"版本更新完成\n版本: v0.2.2 -> v0.2.2\n请重新启动真寻以完成更新!"
|
"版本更新完成!\n版本: v0.2.2 -> v0.2.2\n请重新启动真寻以完成更新!"
|
||||||
),
|
),
|
||||||
result=None,
|
result=None,
|
||||||
bot=bot,
|
bot=bot,
|
||||||
@ -340,9 +333,13 @@ async def test_check_update_release(
|
|||||||
assert mocked_api["release_latest"].called
|
assert mocked_api["release_latest"].called
|
||||||
assert mocked_api["release_download_url_redirect"].called
|
assert mocked_api["release_download_url_redirect"].called
|
||||||
|
|
||||||
assert (mock_backup_path / PYPROJECT_FILE_STRING).exists()
|
assert (mock_backup_path / ZhenxunRepoManager.config.PYPROJECT_FILE_STRING).exists()
|
||||||
assert (mock_backup_path / PYPROJECT_LOCK_FILE_STRING).exists()
|
assert (
|
||||||
assert (mock_backup_path / REQ_TXT_FILE_STRING).exists()
|
mock_backup_path / ZhenxunRepoManager.config.PYPROJECT_LOCK_FILE_STRING
|
||||||
|
).exists()
|
||||||
|
assert (
|
||||||
|
mock_backup_path / ZhenxunRepoManager.config.REQUIREMENTS_FILE_STRING
|
||||||
|
).exists()
|
||||||
|
|
||||||
assert not mock_download_gz_file.exists()
|
assert not mock_download_gz_file.exists()
|
||||||
assert not mock_download_zip_file.exists()
|
assert not mock_download_zip_file.exists()
|
||||||
@ -351,9 +348,9 @@ async def test_check_update_release(
|
|||||||
assert mock_pyproject_lock_file.read_bytes() == b"new"
|
assert mock_pyproject_lock_file.read_bytes() == b"new"
|
||||||
assert mock_req_txt_file.read_bytes() == b"new"
|
assert mock_req_txt_file.read_bytes() == b"new"
|
||||||
|
|
||||||
for folder in REPLACE_FOLDERS:
|
for folder in ZhenxunRepoManager.config.ZHENXUN_BOT_UPDATE_FOLDERS:
|
||||||
assert not (mock_base_path / folder).exists()
|
assert not (mock_base_path / folder).exists()
|
||||||
for folder in REPLACE_FOLDERS:
|
for folder in ZhenxunRepoManager.config.ZHENXUN_BOT_UPDATE_FOLDERS:
|
||||||
assert (mock_backup_path / folder).exists()
|
assert (mock_backup_path / folder).exists()
|
||||||
|
|
||||||
|
|
||||||
@ -368,12 +365,7 @@ async def test_check_update_main(
|
|||||||
测试检查更新(正式环境)
|
测试检查更新(正式环境)
|
||||||
"""
|
"""
|
||||||
from zhenxun.builtin_plugins.auto_update import _matcher
|
from zhenxun.builtin_plugins.auto_update import _matcher
|
||||||
from zhenxun.builtin_plugins.auto_update.config import (
|
from zhenxun.utils.manager.zhenxun_repo_manager import ZhenxunRepoManager
|
||||||
PYPROJECT_FILE_STRING,
|
|
||||||
PYPROJECT_LOCK_FILE_STRING,
|
|
||||||
REPLACE_FOLDERS,
|
|
||||||
REQ_TXT_FILE_STRING,
|
|
||||||
)
|
|
||||||
|
|
||||||
init_mocked_api(mocked_api=mocked_api)
|
init_mocked_api(mocked_api=mocked_api)
|
||||||
|
|
||||||
@ -391,7 +383,7 @@ async def test_check_update_main(
|
|||||||
|
|
||||||
# 确保目录下有一个子目录,以便 os.listdir() 能返回一个目录名
|
# 确保目录下有一个子目录,以便 os.listdir() 能返回一个目录名
|
||||||
mock_tmp_path.mkdir(parents=True, exist_ok=True)
|
mock_tmp_path.mkdir(parents=True, exist_ok=True)
|
||||||
for folder in REPLACE_FOLDERS:
|
for folder in ZhenxunRepoManager.config.ZHENXUN_BOT_UPDATE_FOLDERS:
|
||||||
(mock_base_path / folder).mkdir(parents=True, exist_ok=True)
|
(mock_base_path / folder).mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
mock_pyproject_file.write_bytes(b"")
|
mock_pyproject_file.write_bytes(b"")
|
||||||
@ -401,7 +393,7 @@ async def test_check_update_main(
|
|||||||
async with app.test_matcher(_matcher) as ctx:
|
async with app.test_matcher(_matcher) as ctx:
|
||||||
bot = create_bot(ctx)
|
bot = create_bot(ctx)
|
||||||
bot = cast(Bot, bot)
|
bot = cast(Bot, bot)
|
||||||
raw_message = "检查更新 main -r"
|
raw_message = "检查更新 main -r -z"
|
||||||
event = _v11_group_message_event(
|
event = _v11_group_message_event(
|
||||||
raw_message,
|
raw_message,
|
||||||
self_id=BotId.QQ_BOT,
|
self_id=BotId.QQ_BOT,
|
||||||
@ -420,26 +412,30 @@ async def test_check_update_main(
|
|||||||
ctx.should_call_api(
|
ctx.should_call_api(
|
||||||
"send_msg",
|
"send_msg",
|
||||||
_v11_private_message_send(
|
_v11_private_message_send(
|
||||||
message="检测真寻已更新,版本更新:v0.2.2\n开始更新...",
|
message="检测真寻已更新,当前版本:v0.2.2\n开始更新...",
|
||||||
user_id=UserId.SUPERUSER,
|
user_id=UserId.SUPERUSER,
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
ctx.should_call_send(
|
ctx.should_call_send(
|
||||||
event=event,
|
event=event,
|
||||||
message=Message(
|
message=Message(
|
||||||
"版本更新完成\n"
|
"版本更新完成!\n"
|
||||||
"版本: v0.2.2 -> v0.2.2-e6f17c4\n"
|
"版本: v0.2.2 -> v0.2.2-e6f17c4\n"
|
||||||
"请重新启动真寻以完成更新!\n"
|
"请重新启动真寻以完成更新!\n"
|
||||||
"资源文件更新成功!"
|
"真寻资源更新完成!"
|
||||||
),
|
),
|
||||||
result=None,
|
result=None,
|
||||||
bot=bot,
|
bot=bot,
|
||||||
)
|
)
|
||||||
ctx.should_finished(_matcher)
|
ctx.should_finished(_matcher)
|
||||||
assert mocked_api["main_download_url"].called
|
assert mocked_api["main_download_url"].called
|
||||||
assert (mock_backup_path / PYPROJECT_FILE_STRING).exists()
|
assert (mock_backup_path / ZhenxunRepoManager.config.PYPROJECT_FILE_STRING).exists()
|
||||||
assert (mock_backup_path / PYPROJECT_LOCK_FILE_STRING).exists()
|
assert (
|
||||||
assert (mock_backup_path / REQ_TXT_FILE_STRING).exists()
|
mock_backup_path / ZhenxunRepoManager.config.PYPROJECT_LOCK_FILE_STRING
|
||||||
|
).exists()
|
||||||
|
assert (
|
||||||
|
mock_backup_path / ZhenxunRepoManager.config.REQUIREMENTS_FILE_STRING
|
||||||
|
).exists()
|
||||||
|
|
||||||
assert not mock_download_gz_file.exists()
|
assert not mock_download_gz_file.exists()
|
||||||
assert not mock_download_zip_file.exists()
|
assert not mock_download_zip_file.exists()
|
||||||
@ -448,7 +444,7 @@ async def test_check_update_main(
|
|||||||
assert mock_pyproject_lock_file.read_bytes() == b"new"
|
assert mock_pyproject_lock_file.read_bytes() == b"new"
|
||||||
assert mock_req_txt_file.read_bytes() == b"new"
|
assert mock_req_txt_file.read_bytes() == b"new"
|
||||||
|
|
||||||
for folder in REPLACE_FOLDERS:
|
for folder in ZhenxunRepoManager.config.ZHENXUN_BOT_UPDATE_FOLDERS:
|
||||||
assert (mock_base_path / folder).exists()
|
assert (mock_base_path / folder).exists()
|
||||||
for folder in REPLACE_FOLDERS:
|
for folder in ZhenxunRepoManager.config.ZHENXUN_BOT_UPDATE_FOLDERS:
|
||||||
assert (mock_backup_path / folder).exists()
|
assert (mock_backup_path / folder).exists()
|
||||||
|
|||||||
@ -110,7 +110,8 @@ async def _(
|
|||||||
try:
|
try:
|
||||||
result += await UpdateManager.update_webui(
|
result += await UpdateManager.update_webui(
|
||||||
source_str, # type: ignore
|
source_str, # type: ignore
|
||||||
"dist",
|
"test",
|
||||||
|
True,
|
||||||
)
|
)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error("WebUI更新失败...", "检查更新", session=session, e=e)
|
logger.error("WebUI更新失败...", "检查更新", session=session, e=e)
|
||||||
|
|||||||
@ -34,7 +34,7 @@ class UpdateManager:
|
|||||||
async def update_webui(
|
async def update_webui(
|
||||||
cls,
|
cls,
|
||||||
source: Literal["git", "ali"] | None,
|
source: Literal["git", "ali"] | None,
|
||||||
branch: str = "main",
|
branch: str = "dist",
|
||||||
force: bool = False,
|
force: bool = False,
|
||||||
):
|
):
|
||||||
"""更新WebUI
|
"""更新WebUI
|
||||||
|
|||||||
@ -8,16 +8,6 @@ if sys.version_info >= (3, 11):
|
|||||||
else:
|
else:
|
||||||
from strenum import StrEnum
|
from strenum import StrEnum
|
||||||
|
|
||||||
from zhenxun.configs.path_config import DATA_PATH, TEMP_PATH
|
|
||||||
|
|
||||||
WEBUI_STRING = "web_ui"
|
|
||||||
PUBLIC_STRING = "public"
|
|
||||||
|
|
||||||
WEBUI_DATA_PATH = DATA_PATH / WEBUI_STRING
|
|
||||||
PUBLIC_PATH = WEBUI_DATA_PATH / PUBLIC_STRING
|
|
||||||
TMP_PATH = TEMP_PATH / WEBUI_STRING
|
|
||||||
|
|
||||||
WEBUI_DIST_GITHUB_URL = "https://github.com/HibiKier/zhenxun_bot_webui/tree/dist"
|
|
||||||
|
|
||||||
app = nonebot.get_app()
|
app = nonebot.get_app()
|
||||||
|
|
||||||
|
|||||||
@ -3,41 +3,38 @@ from fastapi.responses import FileResponse
|
|||||||
from fastapi.staticfiles import StaticFiles
|
from fastapi.staticfiles import StaticFiles
|
||||||
|
|
||||||
from zhenxun.services.log import logger
|
from zhenxun.services.log import logger
|
||||||
|
from zhenxun.utils.manager.zhenxun_repo_manager import ZhenxunRepoManager
|
||||||
from ..config import PUBLIC_PATH
|
|
||||||
from .data_source import COMMAND_NAME, update_webui_assets
|
|
||||||
|
|
||||||
router = APIRouter()
|
router = APIRouter()
|
||||||
|
|
||||||
|
|
||||||
@router.get("/")
|
@router.get("/")
|
||||||
async def index():
|
async def index():
|
||||||
return FileResponse(PUBLIC_PATH / "index.html")
|
return FileResponse(ZhenxunRepoManager.config.WEBUI_PATH / "index.html")
|
||||||
|
|
||||||
|
|
||||||
@router.get("/favicon.ico")
|
@router.get("/favicon.ico")
|
||||||
async def favicon():
|
async def favicon():
|
||||||
return FileResponse(PUBLIC_PATH / "favicon.ico")
|
return FileResponse(ZhenxunRepoManager.config.WEBUI_PATH / "favicon.ico")
|
||||||
|
|
||||||
|
|
||||||
@router.get("/79edfa81f3308a9f.jfif")
|
|
||||||
async def _():
|
|
||||||
return FileResponse(PUBLIC_PATH / "79edfa81f3308a9f.jfif")
|
|
||||||
|
|
||||||
|
|
||||||
async def init_public(app: FastAPI):
|
async def init_public(app: FastAPI):
|
||||||
try:
|
try:
|
||||||
if not PUBLIC_PATH.exists():
|
if not ZhenxunRepoManager.check_webui_exists():
|
||||||
folders = await update_webui_assets()
|
await ZhenxunRepoManager.webui_update(branch="test")
|
||||||
else:
|
folders = [
|
||||||
folders = [x.name for x in PUBLIC_PATH.iterdir() if x.is_dir()]
|
x.name for x in ZhenxunRepoManager.config.WEBUI_PATH.iterdir() if x.is_dir()
|
||||||
|
]
|
||||||
app.include_router(router)
|
app.include_router(router)
|
||||||
for pathname in folders:
|
for pathname in folders:
|
||||||
logger.debug(f"挂载文件夹: {pathname}")
|
logger.debug(f"挂载文件夹: {pathname}")
|
||||||
app.mount(
|
app.mount(
|
||||||
f"/{pathname}",
|
f"/{pathname}",
|
||||||
StaticFiles(directory=PUBLIC_PATH / pathname, check_dir=True),
|
StaticFiles(
|
||||||
|
directory=ZhenxunRepoManager.config.WEBUI_PATH / pathname,
|
||||||
|
check_dir=True,
|
||||||
|
),
|
||||||
name=f"public_{pathname}",
|
name=f"public_{pathname}",
|
||||||
)
|
)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error("初始化 WebUI资源 失败", COMMAND_NAME, e=e)
|
logger.error("初始化 WebUI资源 失败", "WebUI", e=e)
|
||||||
|
|||||||
@ -1,44 +0,0 @@
|
|||||||
from pathlib import Path
|
|
||||||
import shutil
|
|
||||||
import zipfile
|
|
||||||
|
|
||||||
from nonebot.utils import run_sync
|
|
||||||
|
|
||||||
from zhenxun.services.log import logger
|
|
||||||
from zhenxun.utils.github_utils import GithubUtils
|
|
||||||
from zhenxun.utils.http_utils import AsyncHttpx
|
|
||||||
|
|
||||||
from ..config import PUBLIC_PATH, TMP_PATH, WEBUI_DIST_GITHUB_URL
|
|
||||||
|
|
||||||
COMMAND_NAME = "WebUI资源管理"
|
|
||||||
|
|
||||||
|
|
||||||
async def update_webui_assets():
|
|
||||||
webui_assets_path = TMP_PATH / "webui_assets.zip"
|
|
||||||
download_url = await GithubUtils.parse_github_url(
|
|
||||||
WEBUI_DIST_GITHUB_URL
|
|
||||||
).get_archive_download_urls()
|
|
||||||
logger.info("开始下载 webui_assets 资源...", COMMAND_NAME)
|
|
||||||
if await AsyncHttpx.download_file(
|
|
||||||
download_url, webui_assets_path, follow_redirects=True
|
|
||||||
):
|
|
||||||
logger.info("下载 webui_assets 成功...", COMMAND_NAME)
|
|
||||||
return await _file_handle(webui_assets_path)
|
|
||||||
raise Exception("下载 webui_assets 失败", COMMAND_NAME)
|
|
||||||
|
|
||||||
|
|
||||||
@run_sync
|
|
||||||
def _file_handle(webui_assets_path: Path):
|
|
||||||
logger.debug("开始解压 webui_assets...", COMMAND_NAME)
|
|
||||||
if webui_assets_path.exists():
|
|
||||||
tf = zipfile.ZipFile(webui_assets_path)
|
|
||||||
tf.extractall(TMP_PATH)
|
|
||||||
logger.debug("解压 webui_assets 成功...", COMMAND_NAME)
|
|
||||||
else:
|
|
||||||
raise Exception("解压 webui_assets 失败,文件不存在...", COMMAND_NAME)
|
|
||||||
download_file_path = next(f for f in TMP_PATH.iterdir() if f.is_dir())
|
|
||||||
shutil.rmtree(PUBLIC_PATH, ignore_errors=True)
|
|
||||||
shutil.copytree(download_file_path / "dist", PUBLIC_PATH, dirs_exist_ok=True)
|
|
||||||
logger.debug("复制 webui_assets 成功...", COMMAND_NAME)
|
|
||||||
shutil.rmtree(TMP_PATH, ignore_errors=True)
|
|
||||||
return [x.name for x in PUBLIC_PATH.iterdir() if x.is_dir()]
|
|
||||||
@ -65,7 +65,7 @@ class ZhenxunRepoConfig:
|
|||||||
|
|
||||||
# WEB UI 相关配置
|
# WEB UI 相关配置
|
||||||
WEBUI_GIT = "https://github.com/HibiKier/zhenxun_bot_webui.git"
|
WEBUI_GIT = "https://github.com/HibiKier/zhenxun_bot_webui.git"
|
||||||
WEBUI_DIST_GITHUB_URL = "https://github.com/HibiKier/zhenxun_bot_webui/tree/dist"
|
WEBUI_DIST_GITHUB_URL = "https://github.com/HibiKier/zhenxun_bot_webui/tree/test"
|
||||||
WEBUI_DOWNLOAD_FILE_STRING = "webui_assets.zip"
|
WEBUI_DOWNLOAD_FILE_STRING = "webui_assets.zip"
|
||||||
WEBUI_DOWNLOAD_FILE = TEMP_PATH / WEBUI_DOWNLOAD_FILE_STRING
|
WEBUI_DOWNLOAD_FILE = TEMP_PATH / WEBUI_DOWNLOAD_FILE_STRING
|
||||||
WEBUI_UNZIP_PATH = TEMP_PATH / "web_ui"
|
WEBUI_UNZIP_PATH = TEMP_PATH / "web_ui"
|
||||||
@ -85,6 +85,12 @@ class ZhenxunRepoConfig:
|
|||||||
REQUIREMENTS_FILE_STRING = "requirements.txt"
|
REQUIREMENTS_FILE_STRING = "requirements.txt"
|
||||||
REQUIREMENTS_FILE = Path() / REQUIREMENTS_FILE_STRING
|
REQUIREMENTS_FILE = Path() / REQUIREMENTS_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
|
||||||
|
|
||||||
|
|
||||||
class ZhenxunRepoManagerClass:
|
class ZhenxunRepoManagerClass:
|
||||||
"""真寻仓库管理器"""
|
"""真寻仓库管理器"""
|
||||||
@ -99,6 +105,8 @@ class ZhenxunRepoManagerClass:
|
|||||||
参数:
|
参数:
|
||||||
folder_path: 文件夹路径
|
folder_path: 文件夹路径
|
||||||
"""
|
"""
|
||||||
|
if not folder_path.exists():
|
||||||
|
return
|
||||||
for filename in os.listdir(folder_path):
|
for filename in os.listdir(folder_path):
|
||||||
file_path = folder_path / filename
|
file_path = folder_path / filename
|
||||||
try:
|
try:
|
||||||
@ -347,7 +355,7 @@ class ZhenxunRepoManagerClass:
|
|||||||
bool: 是否存在
|
bool: 是否存在
|
||||||
"""
|
"""
|
||||||
if self.config.RESOURCE_PATH.exists():
|
if self.config.RESOURCE_PATH.exists():
|
||||||
font_path = self.config.RESOURCE_PATH / "fonts"
|
font_path = self.config.RESOURCE_PATH / "font"
|
||||||
if font_path.exists() and os.listdir(font_path):
|
if font_path.exists() and os.listdir(font_path):
|
||||||
return True
|
return True
|
||||||
return False
|
return False
|
||||||
@ -438,10 +446,16 @@ class ZhenxunRepoManagerClass:
|
|||||||
|
|
||||||
# ==================== Web UI 管理相关方法 ====================
|
# ==================== Web UI 管理相关方法 ====================
|
||||||
|
|
||||||
|
def check_webui_exists(self) -> bool:
|
||||||
|
"""检查 Web UI 资源是否存在"""
|
||||||
|
return bool(
|
||||||
|
self.config.WEBUI_PATH.exists() and os.listdir(self.config.WEBUI_PATH)
|
||||||
|
)
|
||||||
|
|
||||||
async def webui_download_zip(self):
|
async def webui_download_zip(self):
|
||||||
"""下载 WEBUI_ASSETS 资源"""
|
"""下载 WEBUI_ASSETS 资源"""
|
||||||
download_url = await GithubUtils.parse_github_url(
|
download_url = await GithubUtils.parse_github_url(
|
||||||
self.config.WEBUI_GIT
|
self.config.WEBUI_DIST_GITHUB_URL
|
||||||
).get_archive_download_urls()
|
).get_archive_download_urls()
|
||||||
logger.info("开始下载 WEBUI_ASSETS 资源...", LOG_COMMAND)
|
logger.info("开始下载 WEBUI_ASSETS 资源...", LOG_COMMAND)
|
||||||
if await AsyncHttpx.download_file(
|
if await AsyncHttpx.download_file(
|
||||||
@ -469,16 +483,17 @@ class ZhenxunRepoManagerClass:
|
|||||||
str: 更新结果
|
str: 更新结果
|
||||||
"""
|
"""
|
||||||
if not self.config.WEBUI_DOWNLOAD_FILE.exists():
|
if not self.config.WEBUI_DOWNLOAD_FILE.exists():
|
||||||
raise FileNotFoundError("备份webui文件夹不存在")
|
raise FileNotFoundError("webui文件压缩包不存在")
|
||||||
tf = None
|
tf = None
|
||||||
try:
|
try:
|
||||||
self.__backup_webui()
|
self.__backup_webui()
|
||||||
self.__clear_folder(self.config.WEBUI_PATH)
|
self.__clear_folder(self.config.WEBUI_PATH)
|
||||||
tf = zipfile.ZipFile(self.config.WEBUI_DOWNLOAD_FILE)
|
tf = zipfile.ZipFile(self.config.WEBUI_DOWNLOAD_FILE)
|
||||||
tf.extractall(self.config.WEBUI_UNZIP_PATH)
|
tf.extractall(self.config.WEBUI_UNZIP_PATH)
|
||||||
logger.debug("解压文件压缩包完成...", LOG_COMMAND)
|
logger.debug("Web UI 解压文件压缩包完成...", LOG_COMMAND)
|
||||||
self.__copy_files(self.config.WEBUI_UNZIP_PATH, self.config.WEBUI_PATH)
|
unzip_dir = next(self.config.WEBUI_UNZIP_PATH.iterdir())
|
||||||
logger.debug("复制 WEBUI_ASSETS 成功!", LOG_COMMAND)
|
self.__copy_files(unzip_dir, self.config.WEBUI_PATH)
|
||||||
|
logger.debug("Web UI 复制 WEBUI_ASSETS 成功!", LOG_COMMAND)
|
||||||
shutil.rmtree(self.config.WEBUI_UNZIP_PATH, ignore_errors=True)
|
shutil.rmtree(self.config.WEBUI_UNZIP_PATH, ignore_errors=True)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
if self.config.WEBUI_BACKUP_PATH.exists():
|
if self.config.WEBUI_BACKUP_PATH.exists():
|
||||||
@ -497,7 +512,7 @@ class ZhenxunRepoManagerClass:
|
|||||||
await self.webui_unzip()
|
await self.webui_unzip()
|
||||||
|
|
||||||
async def webui_git_update(
|
async def webui_git_update(
|
||||||
self, source: Literal["git", "ali"], branch: str = "main", force: bool = False
|
self, source: Literal["git", "ali"], branch: str = "dist", force: bool = False
|
||||||
) -> RepoUpdateResult:
|
) -> RepoUpdateResult:
|
||||||
"""使用git或阿里云更新 Web UI
|
"""使用git或阿里云更新 Web UI
|
||||||
|
|
||||||
@ -524,7 +539,7 @@ class ZhenxunRepoManagerClass:
|
|||||||
async def webui_update(
|
async def webui_update(
|
||||||
self,
|
self,
|
||||||
source: Literal["git", "ali"] = "ali",
|
source: Literal["git", "ali"] = "ali",
|
||||||
branch: str = "main",
|
branch: str = "dist",
|
||||||
force: bool = False,
|
force: bool = False,
|
||||||
):
|
):
|
||||||
"""更新 Web UI
|
"""更新 Web UI
|
||||||
|
|||||||
@ -285,9 +285,8 @@ class BaseRepoManager(ABC):
|
|||||||
# 如果目录存在,检查是否是Git仓库
|
# 如果目录存在,检查是否是Git仓库
|
||||||
# 首先检查目录本身是否有.git文件夹
|
# 首先检查目录本身是否有.git文件夹
|
||||||
git_dir = local_path / ".git"
|
git_dir = local_path / ".git"
|
||||||
is_git_repo = git_dir.exists() and git_dir.is_dir()
|
|
||||||
|
|
||||||
if not is_git_repo:
|
if not git_dir.is_dir():
|
||||||
# 如果不是Git仓库,尝试初始化它
|
# 如果不是Git仓库,尝试初始化它
|
||||||
logger.info(f"目录 {local_path} 不是Git仓库,尝试初始化", LOG_COMMAND)
|
logger.info(f"目录 {local_path} 不是Git仓库,尝试初始化", LOG_COMMAND)
|
||||||
init_success, _, init_stderr = await run_git_command(
|
init_success, _, init_stderr = await run_git_command(
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user