Compare commits

..
Author SHA1 Message Date
HibiKier e08d89350f ✨ feat(help): 添加帮助功能的快捷方式支持
- 在帮助插件中新增了对“帮助”命令的快捷方式支持,允许用户通过简化的输入方式获取功能信息。
- 移除了不必要的别名,优化了命令的可用性。
2025-11-03 10:50:45 +08:00
Rumioandwebjoin111 07be73c1b7 ✨ feat(avatar): 引入头像缓存服务并优化头像获取 (#2055)
检查bot是否运行正常 / bot check (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, javascript-typescript) (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, python) (push) Has been cancelled
Sequential Lint and Type Check / ruff-call (push) Has been cancelled
Release Drafter / Update Release Draft (push) Has been cancelled
Force Sync to Aliyun / sync (push) Has been cancelled
Update Version / update-version (push) Has been cancelled
Sequential Lint and Type Check / pyright-call (push) Has been cancelled
Co-authored-by: webjoin111 <455457521@qq.com>
2025-09-28 08:53:10 +08:00
molanp 7e6896fa01 🚑fix(data_source): 修复插件商店更新路径错误 (#2056)
* 🚑fix(data_source): 修复插件商店更新路径错误

* fix(plugin_store): 修复插件模块路径处理逻辑

简化了插件模块路径的赋值逻辑,直接使用插件对象的模块路径,避免不必要的路径分割操作。
同时修复了目标目录判断条件,确保只有在模块路径为根目录时才使用插件名称作为目录。
2025-09-28 08:50:54 +08:00
Rumioandwebjoin111 3cc882b116 ✨ feat(auto_update): 增强自动更新与版本检查 (#2042)
检查bot是否运行正常 / bot check (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, javascript-typescript) (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, python) (push) Has been cancelled
Sequential Lint and Type Check / ruff-call (push) Has been cancelled
Release Drafter / Update Release Draft (push) Has been cancelled
Force Sync to Aliyun / sync (push) Has been cancelled
Update Version / update-version (push) Has been cancelled
Sequential Lint and Type Check / pyright-call (push) Has been cancelled
- 优化 `检查更新` 默认行为,未指定类型时直接显示版本信息
- 扩展版本详情显示:当前版本、最新开发版/正式版(含日期)、资源版本及更新提示
- 新增更新后资源兼容性检查,自动读取 `resources.spec` 并提示更新
- 使用 `asyncio.gather` 并发获取版本信息,引入 `packaging` 库提高比较准确性
- 优化错误处理与日志记录

Co-authored-by: webjoin111 <455457521@qq.com>
2025-09-12 17:38:41 +08:00
molanpandHibiKier ee699fb345 fix(plugin_store): 修复插件商店的安装与卸载逻辑 (#2050)
* fix(plugin_store): 修复插件商店的安装与卸载逻辑

- 优化了插件安装、更新和移除的逻辑
- 调整了插件路径的处理方式,支持更灵活的安装位置
- 重构了 `install_plugin_with_repo` 方法,使用 `StorePluginInfo` 对象作为参数
- 修复了一些潜在的路径问题和模块命名问题

* refactor(zhenxun): 优化插件信息获取逻辑

- 将 PluginInfo.get_or_none 替换为 get_plugin 方法,简化插件信息获取逻辑
- 优化了插件移除操作中的插件信息获取流程

* refactor(zhenxun): 优化 sparse_checkout_clone 函数的实现

- 将 git 操作移至临时目录中执行,避免影响目标目录中的现有内容
- 简化了稀疏检出的配置和执行过程
- 改进了错误处理和回退逻辑
- 优化了文件移动和目录清理的操作

* 🐛 添加移除插件时二次查询

* ✨ plugin_info.get_plugin参数包含plugin_type时无效过滤

---------

Co-authored-by: HibiKier <45528451+HibiKier@users.noreply.github.com>
2025-09-12 17:38:24 +08:00
molanp 631e66d54f fix(htmlrender): 更新htmlrender 导入 路径 (#2051)
- 将 get_browser 的导入路径从 nonebot_plugin_htmlrender 更新为 nonebot_plugin_htmlrender.browser
2025-09-12 16:41:43 +08:00
17 changed files with 279 additions and 105 deletions
@@ -19,12 +19,12 @@ from zhenxun.configs.config import Config
from zhenxun.configs.utils import Command, PluginExtraData, RegisterConfig
from zhenxun.models.chat_history import ChatHistory
from zhenxun.models.group_member_info import GroupInfoUser
from zhenxun.services import avatar_service
from zhenxun.services.log import logger
from zhenxun.ui.builders import TableBuilder
from zhenxun.ui.models import ImageCell, TextCell
from zhenxun.utils.enum import PluginType
from zhenxun.utils.message import MessageUtils
from zhenxun.utils.platform import PlatformUtils
__plugin_meta__ = PluginMetadata(
name="消息统计",
@@ -147,12 +147,14 @@ async def _(
user_in_group.user_name if user_in_group else f"{uid_str}(已退群)"
)
avatar_url = PlatformUtils.get_user_avatar_url(uid_str, platform)
avatar_path = await avatar_service.get_avatar_path(platform, uid_str)
rows_data.append(
[
TextCell(content=str(len(rows_data) + 1)),
ImageCell(src=avatar_url or "", shape="circle"),
ImageCell(
src=avatar_path.as_uri() if avatar_path else "", shape="circle"
),
TextCell(content=user_name),
TextCell(content=str(num), bold=True),
]
+7 -1
View File
@@ -78,12 +78,18 @@ _matcher = on_alconna(
Option("-s|--superuser", action=store_true, help_text="超级用户帮助"),
Option("-d|--detail", action=store_true, help_text="详细帮助"),
),
aliases={"help", "帮助", "菜单"},
aliases={"help", "菜单"},
rule=to_me(),
priority=1,
block=True,
)
_matcher.shortcut(
r"帮助(?P<name>.*?)",
command="功能",
arguments=["{name}"],
prefix=True,
)
_matcher.shortcut(
r"详细帮助",
+3 -1
View File
@@ -13,6 +13,7 @@ from zhenxun.models.statistics import Statistics
from zhenxun.services import (
LLMException,
LLMMessage,
avatar_service,
generate,
)
from zhenxun.services.log import logger
@@ -105,7 +106,8 @@ async def create_help_img(
platform = PlatformUtils.get_platform(session)
bot_id = BotConfig.get_qbot_uid(session.self_id) or session.self_id
bot_avatar_url = PlatformUtils.get_user_avatar_url(bot_id, platform) or ""
bot_avatar_path = await avatar_service.get_avatar_path(platform, bot_id)
bot_avatar_url = bot_avatar_path.as_uri() if bot_avatar_path else ""
builder = PluginMenuBuilder(
bot_name=BotConfig.self_nickname,
+3 -3
View File
@@ -11,6 +11,7 @@ from zhenxun.models.level_user import LevelUser
from zhenxun.models.sign_user import SignUser
from zhenxun.models.statistics import Statistics
from zhenxun.models.user_console import UserConsole
from zhenxun.services import avatar_service
from zhenxun.utils.platform import PlatformUtils
RACE = [
@@ -139,9 +140,8 @@ async def get_user_info(
bytes: 图片数据
"""
platform = PlatformUtils.get_platform(session) or "qq"
avatar_url = (
PlatformUtils.get_user_avatar_url(user_id, platform, session.self_id) or ""
)
avatar_path = await avatar_service.get_avatar_path(platform, user_id)
avatar_url = avatar_path.as_uri() if avatar_path else ""
user = await UserConsole.get_user(user_id, platform)
permission_level = await LevelUser.get_user_level(user_id, group_id)
@@ -11,6 +11,7 @@ from zhenxun.models.mahiro_bank import MahiroBank
from zhenxun.models.mahiro_bank_log import MahiroBankLog
from zhenxun.models.sign_user import SignUser
from zhenxun.models.user_console import UserConsole
from zhenxun.services import avatar_service
from zhenxun.utils.enum import BankHandleType, GoldHandle
from zhenxun.utils.platform import PlatformUtils
@@ -210,9 +211,8 @@ class BankManager:
for deposit in user_today_deposit
]
platform = PlatformUtils.get_platform(session)
avatar_url = PlatformUtils.get_user_avatar_url(
user_id, platform, session.self_id
)
avatar_path = await avatar_service.get_avatar_path(platform, user_id)
avatar_url = avatar_path.as_uri() if avatar_path else ""
return {
"name": uname,
"rank": rank + 1,
@@ -1,3 +1,4 @@
import os
from pathlib import Path
import random
import shutil
@@ -10,6 +11,7 @@ from zhenxun.configs.path_config import TEMP_PATH
from zhenxun.models.plugin_info import PluginInfo
from zhenxun.services.log import logger
from zhenxun.services.plugin_init import PluginInitManager
from zhenxun.utils.enum import PluginType
from zhenxun.utils.image_utils import BuildImage, ImageTemplate, RowStyle
from zhenxun.utils.manager.virtual_env_package_manager import VirtualEnvPackageManager
from zhenxun.utils.repo_utils import RepoFileManager
@@ -183,6 +185,8 @@ class StoreManager:
StorePluginInfo: 插件信息
bool: 是否是外部插件
"""
plugin_list: list[StorePluginInfo]
extra_plugin_list: list[StorePluginInfo]
plugin_list, extra_plugin_list = await cls.get_data()
plugin_info = None
is_external = False
@@ -206,6 +210,12 @@ class StoreManager:
if is_remove:
if plugin_info.module not in modules:
raise PluginStoreException(f"插件 {plugin_info.name} 未安装,无法移除")
if plugin_obj := await PluginInfo.get_plugin(
module=plugin_info.module, plugin_type=PluginType.PARENT
):
plugin_info.module_path = plugin_obj.module_path
elif plugin_obj := await PluginInfo.get_plugin(module=plugin_info.module):
plugin_info.module_path = plugin_obj.module_path
return plugin_info, is_external
if is_update:
@@ -237,9 +247,7 @@ class StoreManager:
plugin_info.github_url = f"{github_url_split[0]}/tree/{version_split[1]}"
logger.info(f"正在安装插件 {plugin_info.name}...", LOG_COMMAND)
await cls.install_plugin_with_repo(
plugin_info.github_url,
plugin_info.module_path,
plugin_info.is_dir,
plugin_info,
is_external,
source,
)
@@ -248,9 +256,7 @@ class StoreManager:
@classmethod
async def install_plugin_with_repo(
cls,
github_url: str,
module_path: str,
is_dir: bool,
plugin_info: StorePluginInfo,
is_external: bool = False,
source: str | None = None,
):
@@ -267,18 +273,26 @@ class StoreManager:
repo_type = RepoType.ALIYUN
elif source == "git":
repo_type = RepoType.GITHUB
replace_module_path = module_path.replace(".", "/")
plugin_name = module_path.split(".")[-1]
module_path = plugin_info.module_path
is_dir = plugin_info.is_dir
github_url = plugin_info.github_url
assert github_url
replace_module_path = module_path.replace(".", "/").lstrip("/")
plugin_name = module_path.split(".")[-1] or plugin_info.module
if is_dir:
files = await RepoFileManager.list_directory_files(
github_url, replace_module_path, repo_type=repo_type
)
else:
files = [RepoFileInfo(path=f"{replace_module_path}.py", is_dir=False)]
local_path = BASE_PATH / "plugins" if is_external else BASE_PATH
target_dir = BASE_PATH / "plugins" / plugin_name
if not is_external:
target_dir = BASE_PATH
elif is_dir and module_path == ".":
target_dir = BASE_PATH / "plugins" / plugin_name
else:
target_dir = BASE_PATH / "plugins"
files = [file for file in files if not file.is_dir]
download_files = [(file.path, local_path / file.path) for file in files]
download_files = [(file.path, target_dir / file.path) for file in files]
result = await RepoFileManager.download_files(
github_url,
download_files,
@@ -298,7 +312,7 @@ class StoreManager:
is_install_req = False
for requirement_path in requirement_paths:
requirement_file = local_path / requirement_path.path
requirement_file = target_dir / requirement_path.path
if requirement_file.exists():
is_install_req = True
await VirtualEnvPackageManager.install_requirement(requirement_file)
@@ -341,13 +355,11 @@ class StoreManager:
str: 返回消息
"""
plugin_info, _ = await cls.get_plugin_by_value(index_or_module, is_remove=True)
path = BASE_PATH
if plugin_info.github_url:
path = BASE_PATH / "plugins"
for p in plugin_info.module_path.split("."):
path = path / p
module_path = plugin_info.module_path
module = module_path.split(".")[-1]
path = BASE_PATH.parent / Path(module_path.replace(".", os.sep))
if not plugin_info.is_dir:
path = Path(f"{path}.py")
path = path.parent / f"{module}.py"
if not path.exists():
return f"插件 {plugin_info.name} 不存在..."
logger.debug(f"尝试移除插件 {plugin_info.name} 文件: {path}", LOG_COMMAND)
@@ -356,7 +368,7 @@ class StoreManager:
shutil.rmtree(path, onerror=win_on_rm_error)
else:
path.unlink()
await PluginInitManager.remove(f"zhenxun.{plugin_info.module_path}")
await PluginInitManager.remove(module_path)
return f"插件 {plugin_info.name} 移除成功! 重启后生效"
@classmethod
@@ -423,9 +435,7 @@ class StoreManager:
if plugin_info.github_url is None:
plugin_info.github_url = DEFAULT_GITHUB_URL
await cls.install_plugin_with_repo(
plugin_info.github_url,
plugin_info.module_path,
plugin_info.is_dir,
plugin_info,
is_external,
)
return f"插件 {plugin_info.name} 更新成功! 重启后生效"
@@ -473,9 +483,7 @@ class StoreManager:
plugin_info.github_url = DEFAULT_GITHUB_URL
is_external = False
await cls.install_plugin_with_repo(
plugin_info.github_url,
plugin_info.module_path,
plugin_info.is_dir,
plugin_info,
is_external,
)
update_success_list.append(plugin_info.name)
+6 -3
View File
@@ -21,6 +21,7 @@ from zhenxun.models.group_member_info import GroupInfoUser
from zhenxun.models.user_console import UserConsole
from zhenxun.models.user_gold_log import UserGoldLog
from zhenxun.models.user_props_log import UserPropsLog
from zhenxun.services import avatar_service
from zhenxun.services.log import logger
from zhenxun.ui.models import ImageCell, TextCell
from zhenxun.utils.enum import GoldHandle, PropHandle
@@ -123,12 +124,14 @@ async def gold_rank(session: Uninfo, group_id: str | None, num: int) -> bytes |
data_list = []
platform = PlatformUtils.get_platform(session)
for i, user in enumerate(user_list):
ava_url = PlatformUtils.get_user_avatar_url(user[0], platform, session.self_id)
avatar_path = await avatar_service.get_avatar_path(platform, user[0])
data_list.append(
[
TextCell(content=f"{i + 1}"),
ImageCell(src=ava_url or "", shape="circle")
if platform == "qq"
ImageCell(
src=avatar_path.as_uri() if avatar_path else "", shape="circle"
)
if avatar_path
else TextCell(content=""),
TextCell(content=uid2name.get(user[0]) or user[0]),
TextCell(content=str(user[1]), bold=True),
@@ -13,6 +13,7 @@ from zhenxun.models.group_member_info import GroupInfoUser
from zhenxun.models.sign_log import SignLog
from zhenxun.models.sign_user import SignUser
from zhenxun.models.user_console import UserConsole
from zhenxun.services.avatar_service import avatar_service
from zhenxun.services.log import logger
from zhenxun.ui.models import ImageCell, TextCell
from zhenxun.utils.platform import PlatformUtils
@@ -79,14 +80,16 @@ class SignManage:
data_list = []
platform = PlatformUtils.get_platform(session)
for i, user in enumerate(user_list):
ava_url = PlatformUtils.get_user_avatar_url(
user[0], platform, session.self_id
avatar_path = await avatar_service.get_avatar_path(
platform=user[3] or "qq", identifier=user[0]
)
data_list.append(
[
TextCell(content=f"{i + 1}"),
ImageCell(src=ava_url or "", shape="circle")
if user[3] == "qq"
ImageCell(
src=avatar_path.as_uri() if avatar_path else "", shape="circle"
)
if avatar_path
else TextCell(content=""),
TextCell(content=uid2name.get(user[0]) or user[0]),
TextCell(content=str(user[1]), bold=True),
+5 -4
View File
@@ -11,6 +11,7 @@ from nonebot_plugin_uninfo import Uninfo
from zhenxun import ui
from zhenxun.configs.config import BotConfig, Config
from zhenxun.models.sign_user import SignUser
from zhenxun.services import avatar_service
from zhenxun.utils.manager.priority_manager import PriorityLifecycle
from zhenxun.utils.platform import PlatformUtils
@@ -212,13 +213,13 @@ async def _generate_html_card(
if len(nickname) > 6:
font_size = 27
avatar_path = await avatar_service.get_avatar_path(
PlatformUtils.get_platform(session), user.user_id
)
user_info = {
"nickname": nickname,
"uid_str": uid_formatted,
"avatar_url": PlatformUtils.get_user_avatar_url(
user.user_id, PlatformUtils.get_platform(session), session.self_id
)
or "",
"avatar_url": avatar_path.as_uri() if avatar_path else "",
"sign_count": user.sign_count,
"font_size": font_size,
}
+2 -2
View File
@@ -77,7 +77,7 @@ class PluginInfo(Model):
返回:
Self | None: 插件
"""
if filter_parent:
if not kwargs.get("plugin_type") and filter_parent:
return await cls.get_or_none(
load_status=load_status, plugin_type__not=PluginType.PARENT, **kwargs
)
@@ -96,7 +96,7 @@ class PluginInfo(Model):
返回:
list[Self]: 插件列表
"""
if filter_parent:
if not kwargs.get("plugin_type") and filter_parent:
return await cls.filter(
load_status=load_status, plugin_type__not=PluginType.PARENT, **kwargs
).all()
+2
View File
@@ -18,6 +18,7 @@ require("nonebot_plugin_htmlrender")
require("nonebot_plugin_uninfo")
require("nonebot_plugin_waiter")
from .avatar_service import avatar_service
from .db_context import Model, disconnect, with_db_timeout
from .llm import (
AI,
@@ -57,6 +58,7 @@ __all__ = [
"Model",
"PluginInit",
"PluginInitManager",
"avatar_service",
"chat",
"clear_model_cache",
"code",
+141
View File
@@ -0,0 +1,141 @@
"""
头像缓存服务
提供一个统一的、带缓存的头像获取服务,支持多平台和可配置的过期策略。
"""
import os
from pathlib import Path
import time
from nonebot_plugin_apscheduler import scheduler
from zhenxun.configs.config import Config
from zhenxun.configs.path_config import DATA_PATH
from zhenxun.services.log import logger
from zhenxun.utils.http_utils import AsyncHttpx
from zhenxun.utils.platform import PlatformUtils
Config.add_plugin_config(
"avatar_cache",
"ENABLED",
True,
help="是否启用头像缓存功能",
default_value=True,
type=bool,
)
Config.add_plugin_config(
"avatar_cache",
"TTL_DAYS",
7,
help="头像缓存的有效期(天)",
default_value=7,
type=int,
)
Config.add_plugin_config(
"avatar_cache",
"CLEANUP_INTERVAL_HOURS",
24,
help="后台清理过期缓存的间隔时间(小时)",
default_value=24,
type=int,
)
class AvatarService:
"""
一个集中式的头像缓存服务,提供L1(内存)和L2(文件)两级缓存。
"""
def __init__(self):
self.cache_path = (DATA_PATH / "cache" / "avatars").resolve()
self.cache_path.mkdir(parents=True, exist_ok=True)
self._memory_cache: dict[str, Path] = {}
def _get_cache_path(self, platform: str, identifier: str) -> Path:
"""
根据平台和ID生成存储的文件路径。
例如: data/cache/avatars/qq/123456789.png
"""
identifier = str(identifier)
return self.cache_path / platform / f"{identifier}.png"
async def get_avatar_path(
self, platform: str, identifier: str, force_refresh: bool = False
) -> Path | None:
"""
获取用户或群组的头像本地路径。
参数:
platform: 平台名称 (e.g., 'qq')
identifier: 用户ID或群组ID
force_refresh: 是否强制刷新缓存
返回:
Path | None: 头像的本地文件路径,如果获取失败则返回None。
"""
if not Config.get_config("avatar_cache", "ENABLED"):
return None
cache_key = f"{platform}-{identifier}"
if not force_refresh and cache_key in self._memory_cache:
if self._memory_cache[cache_key].exists():
return self._memory_cache[cache_key]
local_path = self._get_cache_path(platform, identifier)
ttl_seconds = Config.get_config("avatar_cache", "TTL_DAYS", 7) * 86400
if not force_refresh and local_path.exists():
try:
file_mtime = os.path.getmtime(local_path)
if time.time() - file_mtime < ttl_seconds:
self._memory_cache[cache_key] = local_path
return local_path
except FileNotFoundError:
pass
avatar_url = PlatformUtils.get_user_avatar_url(identifier, platform)
if not avatar_url:
return None
local_path.parent.mkdir(parents=True, exist_ok=True)
if await AsyncHttpx.download_file(avatar_url, local_path):
self._memory_cache[cache_key] = local_path
return local_path
else:
logger.warning(f"下载头像失败: {avatar_url}", "AvatarService")
return None
async def _cleanup_cache(self):
"""后台定时清理过期的缓存文件"""
if not Config.get_config("avatar_cache", "ENABLED"):
return
logger.info("开始执行头像缓存清理任务...", "AvatarService")
ttl_seconds = Config.get_config("avatar_cache", "TTL_DAYS", 7) * 86400
now = time.time()
deleted_count = 0
for root, _, files in os.walk(self.cache_path):
for name in files:
file_path = Path(root) / name
try:
if now - os.path.getmtime(file_path) > ttl_seconds:
file_path.unlink()
deleted_count += 1
except FileNotFoundError:
continue
logger.info(
f"头像缓存清理完成,共删除 {deleted_count} 个过期文件。", "AvatarService"
)
avatar_service = AvatarService()
@scheduler.scheduled_job(
"interval", hours=Config.get_config("avatar_cache", "CLEANUP_INTERVAL_HOURS", 24)
)
async def _run_avatar_cache_cleanup():
await avatar_service._cleanup_cache()
+1 -1
View File
@@ -87,7 +87,7 @@ class PluginInitManager:
@classmethod
async def remove(cls, module_path: str):
"""运行指定插件安装方法"""
"""运行指定插件移除方法"""
if model := cls.plugins.get(module_path):
if model.remove:
class_ = model.class_()
+1 -1
View File
@@ -4,7 +4,7 @@ from pathlib import Path
from typing import Any, Literal
from nonebot_plugin_alconna import UniMessage
from nonebot_plugin_htmlrender import get_browser
from nonebot_plugin_htmlrender.browser import get_browser
from playwright.async_api import Page
from zhenxun.utils.message import MessageUtils
+1 -1
View File
@@ -247,7 +247,7 @@ class PlatformUtils:
if platform != "qq":
return None
if user_id.isdigit():
return f"http://q1.qlogo.cn/g?b=qq&nk={user_id}&s=160"
return f"http://q1.qlogo.cn/g?b=qq&nk={user_id}&s=640"
else:
return f"https://q.qlogo.cn/qqapp/{appid}/{user_id}/640"
+2 -2
View File
@@ -326,7 +326,7 @@ class RepoFileManager:
# 获取仓库树信息
strategy = GitHubStrategy()
strategy.body = await GitHubStrategy.parse_repo_info(repo_info)
strategy.body = await strategy.parse_repo_info(repo_info)
# 处理目录路径,确保格式正确
if directory_path and not directory_path.endswith("/") and recursive:
@@ -480,7 +480,7 @@ class RepoFileManager:
target_dir: Path | None = None,
) -> FileDownloadResult:
"""
下载单个文件
下载多个文件
参数:
repo_url: 仓库URL
+57 -51
View File
@@ -7,6 +7,7 @@ import base64
from pathlib import Path
import re
import shutil
import tempfile
from zhenxun.services.log import logger
@@ -145,80 +146,85 @@ async def sparse_checkout_clone(
target_dir: Path,
) -> None:
"""
使用 git 稀疏检出克隆指定路径到目标目录(完全独立于主项目 git)。
使用 git 稀疏检出克隆指定路径到目标目录(在临时目录中操作)。
关键保障:
- 在 target_dir 下检测/初始化 .git,所有 git 操作均以 cwd=target_dir 执行
- 强制拉取与工作区覆盖: fetch --force、checkout -B、reset --hard、clean -xdf
- 反复设置 sparse-checkout 路径,确保路径更新生效
- 在临时目录中执行所有 git 操作,避免影响 target_dir 中的现有内容
- 只操作 target_dir/sparse_path 路径,不影响 target_dir 其他内容
"""
target_dir.mkdir(parents=True, exist_ok=True)
if not await check_git():
raise GitUnavailableError()
git_dir = target_dir / ".git"
if not git_dir.exists():
success, out, err = await run_git_command("init", target_dir)
# 在临时目录中进行 git 操作
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
# 初始化临时目录为 git 仓库
success, out, err = await run_git_command("init", temp_path)
if not success:
raise RuntimeError(f"git init 失败: {err or out}")
success, out, err = await run_git_command(
f"remote add origin {repo_url}", target_dir
f"remote add origin {repo_url}", temp_path
)
if not success:
raise RuntimeError(f"添加远程失败: {err or out}")
else:
# 启用稀疏检出(使用 --no-cone 模式以获得更精确的控制)
await run_git_command("config core.sparseCheckout true", temp_path)
await run_git_command("sparse-checkout init --no-cone", temp_path)
# 设置需要检出的路径(每次都覆盖配置)
if not sparse_path:
raise RuntimeError("sparse-checkout 路径不能为空")
# 使用 --no-cone 模式,直接指定要检出的具体路径
success, out, err = await run_git_command(
f"remote set-url origin {repo_url}", target_dir
f"sparse-checkout set {sparse_path}/", temp_path
)
if not success:
# 兜底尝试添加
await run_git_command(f"remote add origin {repo_url}", target_dir)
raise RuntimeError(f"配置稀疏路径失败: {err or out}")
# 启用稀疏检出(使用 --no-cone 模式以获得更精确的控制)
await run_git_command("config core.sparseCheckout true", target_dir)
await run_git_command("sparse-checkout init --no-cone", target_dir)
# 强制拉取并同步到远端
success, out, err = await run_git_command(
f"fetch --force --depth 1 origin {branch}", temp_path
)
if not success:
raise RuntimeError(f"fetch 失败: {err or out}")
# 设置需要检出的路径(每次都覆盖配置)
if not sparse_path:
raise RuntimeError("sparse-checkout 路径不能为空")
# 使用远端强制更新本地分支并覆盖工作区
success, out, err = await run_git_command(
f"checkout -B {branch} origin/{branch}", temp_path
)
if not success:
# 回退方案
success2, out2, err2 = await run_git_command(
f"checkout {branch}", temp_path
)
if not success2:
raise RuntimeError(f"checkout 失败: {(err or out) or (err2 or out2)}")
# 使用 --no-cone 模式,直接指定要检出的具体路径
# 例如:sparse_path="plugins/mahiro" -> 只检出 plugins/mahiro/ 下的内容
success, out, err = await run_git_command(
f"sparse-checkout set {sparse_path}/", target_dir
)
if not success:
raise RuntimeError(f"配置稀疏路径失败: {err or out}")
# 强制对齐工作区
await run_git_command(f"reset --hard origin/{branch}", temp_path)
await run_git_command("clean -xdf", temp_path)
# 强制拉取并同步到远端
success, out, err = await run_git_command(
f"fetch --force --depth 1 origin {branch}", target_dir
)
if not success:
raise RuntimeError(f"fetch 失败: {err or out}")
# 将检出的文件移动到目标位置
source_path = temp_path / sparse_path
if source_path.exists():
# 确保目标路径存在
target_path = target_dir / sparse_path
target_path.parent.mkdir(parents=True, exist_ok=True)
# 使用远端强制更新本地分支并覆盖工作区
success, out, err = await run_git_command(
f"checkout -B {branch} origin/{branch}", target_dir
)
if not success:
# 回退方案
success2, out2, err2 = await run_git_command(f"checkout {branch}", target_dir)
if not success2:
raise RuntimeError(f"checkout 失败: {(err or out) or (err2 or out2)}")
# 如果目标路径已存在,先清理
if target_path.exists():
if target_path.is_dir():
shutil.rmtree(target_path)
else:
target_path.unlink()
# 强制对齐工作区
await run_git_command(f"reset --hard origin/{branch}", target_dir)
await run_git_command("clean -xdf", target_dir)
dir_path = target_dir / Path(sparse_path)
for f in dir_path.iterdir():
shutil.move(f, target_dir / f.name)
dir_name = sparse_path.split("/")[0]
rm_path = target_dir / dir_name
if rm_path.exists():
shutil.rmtree(rm_path)
# 移动整个目录结构到目标位置
shutil.move(str(source_path), str(target_path))
def prepare_aliyun_url(repo_url: str) -> str: