mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2026-10-04 03:10:01 +08:00
Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6f7e9fc839 | ||
|
|
b7864e535d | ||
|
|
c495c5d9b5 | ||
|
|
e3d49c7105 |
@@ -6,7 +6,6 @@ from packaging.specifiers import SpecifierSet
|
||||
from packaging.version import InvalidVersion, Version
|
||||
|
||||
from zhenxun.services.log import logger
|
||||
from zhenxun.utils.http_utils import AsyncHttpx
|
||||
from zhenxun.utils.manager.virtual_env_package_manager import VirtualEnvPackageManager
|
||||
from zhenxun.utils.manager.zhenxun_repo_manager import (
|
||||
ZhenxunRepoConfig,
|
||||
@@ -19,20 +18,6 @@ LOG_COMMAND = "AutoUpdate"
|
||||
|
||||
|
||||
class UpdateManager:
|
||||
@staticmethod
|
||||
async def _get_latest_commit_date(owner: str, repo: str, path: str) -> str:
|
||||
"""获取文件最新 commit 日期"""
|
||||
api_url = f"https://api.github.com/repos/{owner}/{repo}/commits"
|
||||
params = {"path": path, "page": 1, "per_page": 1}
|
||||
try:
|
||||
data = await AsyncHttpx.get_json(api_url, params=params)
|
||||
if data and isinstance(data, list) and data[0]:
|
||||
date_str = data[0]["commit"]["committer"]["date"]
|
||||
return date_str.split("T")[0]
|
||||
except Exception as e:
|
||||
logger.warning(f"获取 {owner}/{repo}/{path} 的 commit 日期失败", e=e)
|
||||
return "获取失败"
|
||||
|
||||
@classmethod
|
||||
async def check_version(cls) -> str:
|
||||
"""检查真寻和资源的版本"""
|
||||
@@ -42,11 +27,11 @@ class UpdateManager:
|
||||
dev_version_task = RepoFileManager.get_file_content(
|
||||
ZhenxunRepoConfig.ZHENXUN_BOT_GITHUB_URL, "__version__"
|
||||
)
|
||||
bot_commit_date_task = cls._get_latest_commit_date(
|
||||
"HibiKier", "zhenxun_bot", "__version__"
|
||||
bot_commit_date_task = RepoFileManager.get_file_last_commit_date(
|
||||
ZhenxunRepoConfig.ZHENXUN_BOT_GITHUB_URL, "__version__"
|
||||
)
|
||||
res_commit_date_task = cls._get_latest_commit_date(
|
||||
"zhenxun-org", "zhenxun-bot-resources", "__version__"
|
||||
res_commit_date_task = RepoFileManager.get_file_last_commit_date(
|
||||
ZhenxunRepoConfig.RESOURCE_GITHUB_URL, "__version__"
|
||||
)
|
||||
|
||||
(
|
||||
|
||||
@@ -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,14 +147,12 @@ async def _(
|
||||
user_in_group.user_name if user_in_group else f"{uid_str}(已退群)"
|
||||
)
|
||||
|
||||
avatar_path = await avatar_service.get_avatar_path(platform, uid_str)
|
||||
avatar_url = PlatformUtils.get_user_avatar_url(uid_str, platform)
|
||||
|
||||
rows_data.append(
|
||||
[
|
||||
TextCell(content=str(len(rows_data) + 1)),
|
||||
ImageCell(
|
||||
src=avatar_path.as_uri() if avatar_path else "", shape="circle"
|
||||
),
|
||||
ImageCell(src=avatar_url or "", shape="circle"),
|
||||
TextCell(content=user_name),
|
||||
TextCell(content=str(num), bold=True),
|
||||
]
|
||||
|
||||
@@ -78,18 +78,12 @@ _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"详细帮助",
|
||||
|
||||
@@ -13,7 +13,6 @@ from zhenxun.models.statistics import Statistics
|
||||
from zhenxun.services import (
|
||||
LLMException,
|
||||
LLMMessage,
|
||||
avatar_service,
|
||||
generate,
|
||||
)
|
||||
from zhenxun.services.log import logger
|
||||
@@ -106,8 +105,7 @@ 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_path = await avatar_service.get_avatar_path(platform, bot_id)
|
||||
bot_avatar_url = bot_avatar_path.as_uri() if bot_avatar_path else ""
|
||||
bot_avatar_url = PlatformUtils.get_user_avatar_url(bot_id, platform) or ""
|
||||
|
||||
builder = PluginMenuBuilder(
|
||||
bot_name=BotConfig.self_nickname,
|
||||
|
||||
@@ -11,7 +11,6 @@ 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 = [
|
||||
@@ -140,8 +139,9 @@ async def get_user_info(
|
||||
bytes: 图片数据
|
||||
"""
|
||||
platform = PlatformUtils.get_platform(session) or "qq"
|
||||
avatar_path = await avatar_service.get_avatar_path(platform, user_id)
|
||||
avatar_url = avatar_path.as_uri() if avatar_path else ""
|
||||
avatar_url = (
|
||||
PlatformUtils.get_user_avatar_url(user_id, platform, session.self_id) or ""
|
||||
)
|
||||
|
||||
user = await UserConsole.get_user(user_id, platform)
|
||||
permission_level = await LevelUser.get_user_level(user_id, group_id)
|
||||
|
||||
@@ -11,7 +11,6 @@ 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
|
||||
|
||||
@@ -211,8 +210,9 @@ class BankManager:
|
||||
for deposit in user_today_deposit
|
||||
]
|
||||
platform = PlatformUtils.get_platform(session)
|
||||
avatar_path = await avatar_service.get_avatar_path(platform, user_id)
|
||||
avatar_url = avatar_path.as_uri() if avatar_path else ""
|
||||
avatar_url = PlatformUtils.get_user_avatar_url(
|
||||
user_id, platform, session.self_id
|
||||
)
|
||||
return {
|
||||
"name": uname,
|
||||
"rank": rank + 1,
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
import random
|
||||
import shutil
|
||||
@@ -11,7 +10,6 @@ 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
|
||||
@@ -185,8 +183,6 @@ 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
|
||||
@@ -210,12 +206,6 @@ 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:
|
||||
@@ -247,7 +237,9 @@ 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,
|
||||
plugin_info.github_url,
|
||||
plugin_info.module_path,
|
||||
plugin_info.is_dir,
|
||||
is_external,
|
||||
source,
|
||||
)
|
||||
@@ -256,7 +248,9 @@ class StoreManager:
|
||||
@classmethod
|
||||
async def install_plugin_with_repo(
|
||||
cls,
|
||||
plugin_info: StorePluginInfo,
|
||||
github_url: str,
|
||||
module_path: str,
|
||||
is_dir: bool,
|
||||
is_external: bool = False,
|
||||
source: str | None = None,
|
||||
):
|
||||
@@ -273,26 +267,18 @@ class StoreManager:
|
||||
repo_type = RepoType.ALIYUN
|
||||
elif source == "git":
|
||||
repo_type = RepoType.GITHUB
|
||||
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
|
||||
replace_module_path = module_path.replace(".", "/")
|
||||
plugin_name = module_path.split(".")[-1]
|
||||
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)]
|
||||
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"
|
||||
local_path = BASE_PATH / "plugins" if is_external else BASE_PATH
|
||||
target_dir = BASE_PATH / "plugins" / plugin_name
|
||||
files = [file for file in files if not file.is_dir]
|
||||
download_files = [(file.path, target_dir / file.path) for file in files]
|
||||
download_files = [(file.path, local_path / file.path) for file in files]
|
||||
result = await RepoFileManager.download_files(
|
||||
github_url,
|
||||
download_files,
|
||||
@@ -312,7 +298,7 @@ class StoreManager:
|
||||
|
||||
is_install_req = False
|
||||
for requirement_path in requirement_paths:
|
||||
requirement_file = target_dir / requirement_path.path
|
||||
requirement_file = local_path / requirement_path.path
|
||||
if requirement_file.exists():
|
||||
is_install_req = True
|
||||
await VirtualEnvPackageManager.install_requirement(requirement_file)
|
||||
@@ -355,11 +341,13 @@ class StoreManager:
|
||||
str: 返回消息
|
||||
"""
|
||||
plugin_info, _ = await cls.get_plugin_by_value(index_or_module, is_remove=True)
|
||||
module_path = plugin_info.module_path
|
||||
module = module_path.split(".")[-1]
|
||||
path = BASE_PATH.parent / Path(module_path.replace(".", os.sep))
|
||||
path = BASE_PATH
|
||||
if plugin_info.github_url:
|
||||
path = BASE_PATH / "plugins"
|
||||
for p in plugin_info.module_path.split("."):
|
||||
path = path / p
|
||||
if not plugin_info.is_dir:
|
||||
path = path.parent / f"{module}.py"
|
||||
path = Path(f"{path}.py")
|
||||
if not path.exists():
|
||||
return f"插件 {plugin_info.name} 不存在..."
|
||||
logger.debug(f"尝试移除插件 {plugin_info.name} 文件: {path}", LOG_COMMAND)
|
||||
@@ -368,7 +356,7 @@ class StoreManager:
|
||||
shutil.rmtree(path, onerror=win_on_rm_error)
|
||||
else:
|
||||
path.unlink()
|
||||
await PluginInitManager.remove(module_path)
|
||||
await PluginInitManager.remove(f"zhenxun.{plugin_info.module_path}")
|
||||
return f"插件 {plugin_info.name} 移除成功! 重启后生效"
|
||||
|
||||
@classmethod
|
||||
@@ -435,7 +423,9 @@ class StoreManager:
|
||||
if plugin_info.github_url is None:
|
||||
plugin_info.github_url = DEFAULT_GITHUB_URL
|
||||
await cls.install_plugin_with_repo(
|
||||
plugin_info,
|
||||
plugin_info.github_url,
|
||||
plugin_info.module_path,
|
||||
plugin_info.is_dir,
|
||||
is_external,
|
||||
)
|
||||
return f"插件 {plugin_info.name} 更新成功! 重启后生效"
|
||||
@@ -483,7 +473,9 @@ class StoreManager:
|
||||
plugin_info.github_url = DEFAULT_GITHUB_URL
|
||||
is_external = False
|
||||
await cls.install_plugin_with_repo(
|
||||
plugin_info,
|
||||
plugin_info.github_url,
|
||||
plugin_info.module_path,
|
||||
plugin_info.is_dir,
|
||||
is_external,
|
||||
)
|
||||
update_success_list.append(plugin_info.name)
|
||||
|
||||
@@ -153,7 +153,7 @@ async def _(session: Uninfo, arparma: Arparma, nickname: str = UserName()):
|
||||
nickname,
|
||||
PlatformUtils.get_platform(session),
|
||||
):
|
||||
await MessageUtils.build_message(image).finish(reply_to=True) # type: ignore
|
||||
await MessageUtils.build_message(image.pic2bytes()).finish(reply_to=True) # type: ignore
|
||||
return await MessageUtils.build_message("你的道具为空捏...").send(reply_to=True)
|
||||
|
||||
|
||||
|
||||
@@ -21,7 +21,6 @@ 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
|
||||
@@ -124,14 +123,12 @@ 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):
|
||||
avatar_path = await avatar_service.get_avatar_path(platform, user[0])
|
||||
ava_url = PlatformUtils.get_user_avatar_url(user[0], platform, session.self_id)
|
||||
data_list.append(
|
||||
[
|
||||
TextCell(content=f"{i + 1}"),
|
||||
ImageCell(
|
||||
src=avatar_path.as_uri() if avatar_path else "", shape="circle"
|
||||
)
|
||||
if avatar_path
|
||||
ImageCell(src=ava_url or "", shape="circle")
|
||||
if platform == "qq"
|
||||
else TextCell(content=""),
|
||||
TextCell(content=uid2name.get(user[0]) or user[0]),
|
||||
TextCell(content=str(user[1]), bold=True),
|
||||
@@ -532,10 +529,10 @@ class ShopManage:
|
||||
if not prop:
|
||||
continue
|
||||
|
||||
icon = None
|
||||
icon = ""
|
||||
if prop.icon:
|
||||
icon_path = ICON_PATH / prop.icon
|
||||
icon = icon_path if icon_path.exists() else None
|
||||
icon = (icon_path, 33, 33) if icon_path.exists() else ""
|
||||
|
||||
table_rows.append(
|
||||
[
|
||||
|
||||
@@ -13,7 +13,6 @@ 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
|
||||
@@ -80,16 +79,14 @@ class SignManage:
|
||||
data_list = []
|
||||
platform = PlatformUtils.get_platform(session)
|
||||
for i, user in enumerate(user_list):
|
||||
avatar_path = await avatar_service.get_avatar_path(
|
||||
platform=user[3] or "qq", identifier=user[0]
|
||||
ava_url = PlatformUtils.get_user_avatar_url(
|
||||
user[0], platform, session.self_id
|
||||
)
|
||||
data_list.append(
|
||||
[
|
||||
TextCell(content=f"{i + 1}"),
|
||||
ImageCell(
|
||||
src=avatar_path.as_uri() if avatar_path else "", shape="circle"
|
||||
)
|
||||
if avatar_path
|
||||
ImageCell(src=ava_url or "", shape="circle")
|
||||
if user[3] == "qq"
|
||||
else TextCell(content=""),
|
||||
TextCell(content=uid2name.get(user[0]) or user[0]),
|
||||
TextCell(content=str(user[1]), bold=True),
|
||||
|
||||
@@ -11,7 +11,6 @@ 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
|
||||
|
||||
@@ -213,13 +212,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": avatar_path.as_uri() if avatar_path else "",
|
||||
"avatar_url": PlatformUtils.get_user_avatar_url(
|
||||
user.user_id, PlatformUtils.get_platform(session), session.self_id
|
||||
)
|
||||
or "",
|
||||
"sign_count": user.sign_count,
|
||||
"font_size": font_size,
|
||||
}
|
||||
|
||||
@@ -77,7 +77,7 @@ class PluginInfo(Model):
|
||||
返回:
|
||||
Self | None: 插件
|
||||
"""
|
||||
if not kwargs.get("plugin_type") and filter_parent:
|
||||
if 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 not kwargs.get("plugin_type") and filter_parent:
|
||||
if filter_parent:
|
||||
return await cls.filter(
|
||||
load_status=load_status, plugin_type__not=PluginType.PARENT, **kwargs
|
||||
).all()
|
||||
|
||||
@@ -18,7 +18,6 @@ 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,
|
||||
@@ -58,7 +57,6 @@ __all__ = [
|
||||
"Model",
|
||||
"PluginInit",
|
||||
"PluginInitManager",
|
||||
"avatar_service",
|
||||
"chat",
|
||||
"clear_model_cache",
|
||||
"code",
|
||||
|
||||
@@ -1,141 +0,0 @@
|
||||
"""
|
||||
头像缓存服务
|
||||
|
||||
提供一个统一的、带缓存的头像获取服务,支持多平台和可配置的过期策略。
|
||||
"""
|
||||
|
||||
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()
|
||||
@@ -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,13 +1,6 @@
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
from typing import Literal
|
||||
|
||||
from ...models.core.table import (
|
||||
BaseCell,
|
||||
ImageCell,
|
||||
TableCell,
|
||||
TableData,
|
||||
TextCell,
|
||||
)
|
||||
from ...models.core.table import TableCell, TableData
|
||||
from ..base import BaseBuilder
|
||||
|
||||
__all__ = ["TableBuilder"]
|
||||
@@ -20,28 +13,6 @@ class TableBuilder(BaseBuilder[TableData]):
|
||||
data_model = TableData(title=title, tip=tip, headers=[], rows=[])
|
||||
super().__init__(data_model, template_name="components/core/table")
|
||||
|
||||
def _normalize_cell(self, cell_data: Any) -> TableCell:
|
||||
"""内部辅助方法,将各种原生数据类型转换为TableCell模型。"""
|
||||
if isinstance(cell_data, BaseCell):
|
||||
return cell_data # type: ignore
|
||||
if isinstance(cell_data, str | int | float):
|
||||
return TextCell(content=str(cell_data))
|
||||
if isinstance(cell_data, Path):
|
||||
return ImageCell(src=cell_data.resolve().as_uri())
|
||||
if isinstance(cell_data, tuple) and len(cell_data) == 3:
|
||||
if (
|
||||
isinstance(cell_data[0], Path)
|
||||
and isinstance(cell_data[1], int)
|
||||
and isinstance(cell_data[2], int)
|
||||
):
|
||||
return ImageCell(
|
||||
src=cell_data[0].resolve().as_uri(),
|
||||
width=cell_data[1],
|
||||
height=cell_data[2],
|
||||
)
|
||||
|
||||
return TextCell(content="")
|
||||
|
||||
def set_headers(self, headers: list[str]) -> "TableBuilder":
|
||||
"""
|
||||
设置表格的表头。
|
||||
@@ -86,13 +57,12 @@ class TableBuilder(BaseBuilder[TableData]):
|
||||
返回:
|
||||
TableBuilder: 当前构建器实例,以支持链式调用。
|
||||
"""
|
||||
normalized_row = [self._normalize_cell(cell) for cell in row]
|
||||
self._data.rows.append(normalized_row)
|
||||
self._data.rows.append(row)
|
||||
return self
|
||||
|
||||
def add_rows(self, rows: list[list[TableCell]]) -> "TableBuilder":
|
||||
"""
|
||||
向表格中批量添加多行数据, 并自动转换原生类型。
|
||||
向表格中批量添加多行数据。
|
||||
|
||||
参数:
|
||||
rows: 一个包含多行数据的列表。
|
||||
@@ -100,6 +70,5 @@ class TableBuilder(BaseBuilder[TableData]):
|
||||
返回:
|
||||
TableBuilder: 当前构建器实例,以支持链式调用。
|
||||
"""
|
||||
for row in rows:
|
||||
self.add_row(row)
|
||||
self._data.rows.extend(rows)
|
||||
return self
|
||||
|
||||
@@ -12,7 +12,6 @@ from .components import (
|
||||
from .core import (
|
||||
BaseCell,
|
||||
CodeElement,
|
||||
ComponentCell,
|
||||
HeadingElement,
|
||||
ImageCell,
|
||||
ImageElement,
|
||||
@@ -50,7 +49,6 @@ __all__ = [
|
||||
"BaseCell",
|
||||
"BaseChartData",
|
||||
"CodeElement",
|
||||
"ComponentCell",
|
||||
"Divider",
|
||||
"EChartsData",
|
||||
"HeadingElement",
|
||||
|
||||
+13
-53
@@ -11,68 +11,44 @@ from .core.base import RenderableComponent
|
||||
|
||||
class EChartsTitle(BaseModel):
|
||||
text: str
|
||||
"""图表主标题"""
|
||||
left: Literal["left", "center", "right"] = "center"
|
||||
"""标题水平对齐方式"""
|
||||
|
||||
|
||||
class EChartsAxis(BaseModel):
|
||||
type: Literal["category", "value", "time", "log"]
|
||||
"""坐标轴类型"""
|
||||
data: list[Any] | None = None
|
||||
"""类目数据"""
|
||||
show: bool = True
|
||||
"""是否显示坐标轴"""
|
||||
|
||||
|
||||
class EChartsSeries(BaseModel):
|
||||
type: str
|
||||
"""系列类型 (e.g., 'bar', 'line', 'pie')"""
|
||||
data: list[Any]
|
||||
"""系列数据"""
|
||||
name: str | None = None
|
||||
"""系列名称,用于 tooltip 的显示"""
|
||||
label: dict[str, Any] | None = None
|
||||
"""图形上的文本标签"""
|
||||
itemStyle: dict[str, Any] | None = None
|
||||
"""图形样式"""
|
||||
barMaxWidth: int | None = None
|
||||
"""柱条的最大宽度"""
|
||||
smooth: bool | None = None
|
||||
"""是否平滑显示折线"""
|
||||
|
||||
|
||||
class EChartsTooltip(BaseModel):
|
||||
trigger: Literal["item", "axis", "none"] = Field("item", description="触发类型")
|
||||
"""触发类型"""
|
||||
trigger: Literal["item", "axis", "none"] = "item"
|
||||
|
||||
|
||||
class EChartsGrid(BaseModel):
|
||||
left: str | None = None
|
||||
"""grid 组件离容器左侧的距离"""
|
||||
right: str | None = None
|
||||
"""grid 组件离容器右侧的距离"""
|
||||
top: str | None = None
|
||||
"""grid 组件离容器上侧的距离"""
|
||||
bottom: str | None = None
|
||||
"""grid 组件离容器下侧的距离"""
|
||||
containLabel: bool = True
|
||||
"""grid 区域是否包含坐标轴的刻度标签"""
|
||||
|
||||
|
||||
class BaseChartData(RenderableComponent, ABC):
|
||||
"""所有图表数据模型的基类"""
|
||||
|
||||
style_name: str | None = None
|
||||
"""组件的样式名称"""
|
||||
chart_id: str = Field(
|
||||
default_factory=lambda: f"chart-{uuid.uuid4().hex}",
|
||||
description="图表的唯一ID,用于前端渲染",
|
||||
)
|
||||
"""图表的唯一ID,用于前端渲染"""
|
||||
chart_id: str = Field(default_factory=lambda: f"chart-{uuid.uuid4().hex}")
|
||||
|
||||
echarts_options: dict[str, Any] | None = None
|
||||
"""原始ECharts选项,用于高级自定义"""
|
||||
|
||||
@abstractmethod
|
||||
def build_option(self) -> dict[str, Any]:
|
||||
@@ -94,37 +70,21 @@ class BaseChartData(RenderableComponent, ABC):
|
||||
class EChartsData(BaseChartData):
|
||||
"""统一的 ECharts 图表数据模型"""
|
||||
|
||||
template_path: str = Field(..., exclude=True, description="图表组件的模板路径")
|
||||
"""图表组件的模板路径"""
|
||||
title_model: EChartsTitle | None = Field(
|
||||
None, alias="title", description="标题组件"
|
||||
)
|
||||
"""标题组件"""
|
||||
grid_model: EChartsGrid | None = Field(None, alias="grid", description="网格组件")
|
||||
"""网格组件"""
|
||||
tooltip_model: EChartsTooltip | None = Field(
|
||||
None, alias="tooltip", description="提示框组件"
|
||||
)
|
||||
"""提示框组件"""
|
||||
x_axis_model: EChartsAxis | None = Field(None, alias="xAxis", description="X轴配置")
|
||||
"""X轴配置"""
|
||||
y_axis_model: EChartsAxis | None = Field(None, alias="yAxis", description="Y轴配置")
|
||||
"""Y轴配置"""
|
||||
series_models: list[EChartsSeries] = Field(
|
||||
default_factory=list, alias="series", description="系列列表"
|
||||
)
|
||||
"""系列列表"""
|
||||
legend_model: dict[str, Any] | None = Field(
|
||||
default_factory=dict, alias="legend", description="图例组件"
|
||||
)
|
||||
"""图例组件"""
|
||||
template_path: str = Field(..., exclude=True)
|
||||
title_model: EChartsTitle | None = Field(None, alias="title")
|
||||
grid_model: EChartsGrid | None = Field(None, alias="grid")
|
||||
tooltip_model: EChartsTooltip | None = Field(None, alias="tooltip")
|
||||
x_axis_model: EChartsAxis | None = Field(None, alias="xAxis")
|
||||
y_axis_model: EChartsAxis | None = Field(None, alias="yAxis")
|
||||
series_models: list[EChartsSeries] = Field(default_factory=list, alias="series")
|
||||
legend_model: dict[str, Any] | None = Field(default_factory=dict, alias="legend")
|
||||
raw_options: dict[str, Any] = Field(
|
||||
default_factory=dict, description="用于 set_option 的原始覆盖选项"
|
||||
)
|
||||
"""用于 set_option 的原始覆盖选项"""
|
||||
|
||||
background_image: str | None = Field(None, description="用于横向柱状图的背景图片")
|
||||
"""用于横向柱状图的背景图片"""
|
||||
background_image: str | None = Field(
|
||||
None, description="【兼容】用于横向柱状图的背景图片"
|
||||
)
|
||||
|
||||
def build_option(self) -> dict[str, Any]:
|
||||
"""将 Pydantic 模型序列化为 ECharts 的 option 字典。"""
|
||||
|
||||
@@ -14,13 +14,9 @@ class Alert(RenderableComponent):
|
||||
type: Literal["info", "success", "warning", "error"] = Field(
|
||||
default="info", description="提示框的类型,决定了颜色和图标"
|
||||
)
|
||||
"""提示框的类型,决定了颜色和图标"""
|
||||
title: str = Field(..., description="提示框的标题")
|
||||
"""提示框的标题"""
|
||||
content: str = Field(..., description="提示框的主要内容")
|
||||
"""提示框的主要内容"""
|
||||
show_icon: bool = Field(default=True, description="是否显示与类型匹配的图标")
|
||||
"""是否显示与类型匹配的图标"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -12,11 +12,8 @@ class Avatar(RenderableComponent):
|
||||
|
||||
component_type: Literal["avatar"] = "avatar"
|
||||
src: str = Field(..., description="头像的URL或Base64数据URI")
|
||||
"""头像的URL或Base64数据URI"""
|
||||
shape: Literal["circle", "square"] = Field("circle", description="头像形状")
|
||||
"""头像形状"""
|
||||
size: int = Field(50, description="头像尺寸(像素)")
|
||||
"""头像尺寸(像素)"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
@@ -28,13 +25,10 @@ class AvatarGroup(RenderableComponent):
|
||||
|
||||
component_type: Literal["avatar_group"] = "avatar_group"
|
||||
avatars: list[Avatar] = Field(default_factory=list, description="头像列表")
|
||||
"""头像列表"""
|
||||
spacing: int = Field(-15, description="头像间的间距(负数表示重叠)")
|
||||
"""头像间的间距(负数表示重叠)"""
|
||||
max_count: int | None = Field(
|
||||
None, description="最多显示的头像数量,超出部分会显示为'+N'"
|
||||
)
|
||||
"""最多显示的头像数量,超出部分会显示为'+N'"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -12,12 +12,10 @@ class Badge(RenderableComponent):
|
||||
|
||||
component_type: Literal["badge"] = "badge"
|
||||
text: str = Field(..., description="徽章上显示的文本")
|
||||
"""徽章上显示的文本"""
|
||||
color_scheme: Literal["primary", "success", "warning", "error", "info"] = Field(
|
||||
default="info",
|
||||
description="预设的颜色方案",
|
||||
)
|
||||
"""预设的颜色方案"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -12,13 +12,9 @@ class Divider(RenderableComponent):
|
||||
|
||||
component_type: Literal["divider"] = "divider"
|
||||
margin: str = Field("2em 0", description="CSS margin属性,控制分割线上下的间距")
|
||||
"""CSS margin属性,控制分割线上下的间距"""
|
||||
color: str = Field("#f7889c", description="分割线颜色")
|
||||
"""分割线颜色"""
|
||||
style: Literal["solid", "dashed", "dotted"] = Field("solid", description="线条样式")
|
||||
"""线条样式"""
|
||||
thickness: str = Field("1px", description="线条粗细")
|
||||
"""线条粗细"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
@@ -30,13 +26,9 @@ class Rectangle(RenderableComponent):
|
||||
|
||||
component_type: Literal["rectangle"] = "rectangle"
|
||||
height: str = Field("50px", description="矩形的高度 (CSS value)")
|
||||
"""矩形的高度 (CSS value)"""
|
||||
background_color: str = Field("#fdf1f5", description="背景颜色")
|
||||
"""背景颜色"""
|
||||
border: str = Field("1px solid #fce4ec", description="CSS border属性")
|
||||
"""CSS border属性"""
|
||||
border_radius: str = Field("8px", description="CSS border-radius属性")
|
||||
"""CSS border-radius属性"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -12,23 +12,17 @@ class KpiCard(RenderableComponent):
|
||||
|
||||
component_type: Literal["kpi_card"] = "kpi_card"
|
||||
label: str = Field(..., description="指标的标签或名称")
|
||||
"""指标的标签或名称"""
|
||||
value: Any = Field(..., description="指标的主要数值")
|
||||
"""指标的主要数值"""
|
||||
unit: str | None = Field(default=None, description="数值的单位,可选")
|
||||
"""数值的单位,可选"""
|
||||
change: str | None = Field(
|
||||
default=None, description="与上一周期的变化,例如 '+15%' 或 '-100'"
|
||||
)
|
||||
"""与上一周期的变化,例如 '+15%' 或 '-100'"""
|
||||
change_type: Literal["positive", "negative", "neutral"] = Field(
|
||||
default="neutral", description="变化的类型,用于决定颜色"
|
||||
)
|
||||
"""变化的类型,用于决定颜色"""
|
||||
icon_svg: str | None = Field(
|
||||
default=None, description="卡片中显示的可选图标 (SVG path data)"
|
||||
)
|
||||
"""卡片中显示的可选图标 (SVG path data)"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -12,16 +12,12 @@ class ProgressBar(RenderableComponent):
|
||||
|
||||
component_type: Literal["progress_bar"] = "progress_bar"
|
||||
progress: float = Field(..., ge=0, le=100, description="进度百分比 (0-100)")
|
||||
"""进度百分比 (0-100)"""
|
||||
label: str | None = Field(default=None, description="显示在进度条上的可选文本")
|
||||
"""显示在进度条上的可选文本"""
|
||||
color_scheme: Literal["primary", "success", "warning", "error", "info"] = Field(
|
||||
default="primary",
|
||||
description="预设的颜色方案",
|
||||
)
|
||||
"""预设的颜色方案"""
|
||||
animated: bool = Field(default=False, description="是否显示动画效果")
|
||||
"""是否显示动画效果"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -11,15 +11,10 @@ class TimelineItem(BaseModel):
|
||||
"""时间轴中的单个事件点。"""
|
||||
|
||||
timestamp: str = Field(..., description="显示在时间点旁边的时间或标签")
|
||||
"""显示在时间点旁边的时间或标签"""
|
||||
title: str = Field(..., description="事件的标题")
|
||||
"""事件的标题"""
|
||||
content: str = Field(..., description="事件的详细描述")
|
||||
"""事件的详细描述"""
|
||||
icon: str | None = Field(default=None, description="可选的自定义图标SVG路径")
|
||||
"""可选的自定义图标SVG路径"""
|
||||
color: str | None = Field(default=None, description="可选的自定义颜色,覆盖默认")
|
||||
"""可选的自定义颜色,覆盖默认"""
|
||||
|
||||
|
||||
class Timeline(RenderableComponent):
|
||||
@@ -29,7 +24,6 @@ class Timeline(RenderableComponent):
|
||||
items: list[TimelineItem] = Field(
|
||||
default_factory=list, description="时间轴项目列表"
|
||||
)
|
||||
"""时间轴项目列表"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -12,15 +12,11 @@ class UserInfoBlock(RenderableComponent):
|
||||
|
||||
component_type: Literal["user_info_block"] = "user_info_block"
|
||||
avatar_url: str = Field(..., description="用户头像的URL")
|
||||
"""用户头像的URL"""
|
||||
name: str = Field(..., description="用户的名称")
|
||||
"""用户的名称"""
|
||||
subtitle: str | None = Field(
|
||||
default=None, description="显示在名称下方的副标题 (如UID或角色)"
|
||||
)
|
||||
"""显示在名称下方的副标题 (如UID或角色)"""
|
||||
tags: list[str] = Field(default_factory=list, description="附加的标签列表")
|
||||
"""附加的标签列表"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -24,7 +24,6 @@ from .markdown import (
|
||||
from .notebook import NotebookData, NotebookElement
|
||||
from .table import (
|
||||
BaseCell,
|
||||
ComponentCell,
|
||||
ImageCell,
|
||||
RichTextCell,
|
||||
StatusBadgeCell,
|
||||
@@ -39,7 +38,6 @@ __all__ = [
|
||||
"BaseCell",
|
||||
"CardData",
|
||||
"CodeElement",
|
||||
"ComponentCell",
|
||||
"DetailsData",
|
||||
"DetailsItem",
|
||||
"HeadingElement",
|
||||
|
||||
@@ -20,15 +20,10 @@ class RenderableComponent(BaseModel, Renderable):
|
||||
"""
|
||||
|
||||
_is_standalone_template: bool = False
|
||||
"""标记此组件是否为独立模板"""
|
||||
inline_style: dict[str, str] | None = None
|
||||
"""应用于组件根元素的内联CSS样式"""
|
||||
component_css: str | None = None
|
||||
"""注入到页面的额外CSS字符串"""
|
||||
extra_classes: list[str] | None = None
|
||||
"""应用于组件根元素的额外CSS类名列表"""
|
||||
variant: str | None = None
|
||||
"""组件的变体/皮肤名称"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -7,11 +7,8 @@ class CardData(ContainerComponent):
|
||||
"""通用卡片的数据模型,可以包含头部、内容和尾部"""
|
||||
|
||||
header: RenderableComponent | None = None
|
||||
"""卡片的头部内容组件"""
|
||||
content: RenderableComponent
|
||||
"""卡片的主要内容组件"""
|
||||
footer: RenderableComponent | None = None
|
||||
"""卡片的尾部内容组件"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -9,18 +9,14 @@ class DetailsItem(BaseModel):
|
||||
"""描述列表中的单个项目"""
|
||||
|
||||
label: str = Field(..., description="项目的标签/键")
|
||||
"""项目的标签/键"""
|
||||
value: Any = Field(..., description="项目的值")
|
||||
"""项目的值"""
|
||||
|
||||
|
||||
class DetailsData(RenderableComponent):
|
||||
"""描述列表(键值对)的数据模型"""
|
||||
|
||||
title: str | None = Field(None, description="列表的可选标题")
|
||||
"""列表的可选标题"""
|
||||
items: list[DetailsItem] = Field(default_factory=list, description="键值对项目列表")
|
||||
"""键值对项目列表"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -12,26 +12,20 @@ class LayoutItem(BaseModel):
|
||||
"""布局中的单个项目,现在持有可渲染组件的数据模型"""
|
||||
|
||||
component: RenderableComponent = Field(..., description="要渲染的组件的数据模型")
|
||||
"""要渲染的组件的数据模型"""
|
||||
metadata: dict[str, Any] | None = Field(None, description="传递给模板的额外元数据")
|
||||
"""传递给模板的额外元数据"""
|
||||
|
||||
|
||||
class LayoutData(ContainerComponent):
|
||||
"""布局构建器的数据模型"""
|
||||
|
||||
style_name: str | None = None
|
||||
"""应用于布局容器的样式名称"""
|
||||
layout_type: str = "column"
|
||||
"""布局类型 (如 'column', 'row', 'grid')"""
|
||||
children: list[LayoutItem] = Field(
|
||||
default_factory=list, description="要布局的项目列表"
|
||||
)
|
||||
"""要布局的项目列表"""
|
||||
options: dict[str, Any] = Field(
|
||||
default_factory=dict, description="传递给模板的选项"
|
||||
)
|
||||
"""传递给模板的选项"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -12,7 +12,6 @@ class ListItem(BaseModel):
|
||||
"""列表中的单个项目,其内容可以是任何可渲染组件。"""
|
||||
|
||||
component: RenderableComponent = Field(..., description="要渲染的组件的数据模型")
|
||||
"""要渲染的组件的数据模型"""
|
||||
|
||||
|
||||
class ListData(ContainerComponent):
|
||||
@@ -20,9 +19,7 @@ class ListData(ContainerComponent):
|
||||
|
||||
component_type: Literal["list"] = "list"
|
||||
items: list[ListItem] = Field(default_factory=list, description="列表项目")
|
||||
"""列表项目"""
|
||||
ordered: bool = Field(default=False, description="是否为有序列表")
|
||||
"""是否为有序列表"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -44,9 +44,7 @@ class TextElement(MarkdownElement):
|
||||
class HeadingElement(MarkdownElement):
|
||||
type: Literal["heading"] = "heading"
|
||||
text: str
|
||||
"""标题文本"""
|
||||
level: int = Field(..., ge=1, le=6, description="标题级别 (1-6)")
|
||||
"""标题级别 (1-6)"""
|
||||
level: int = Field(..., ge=1, le=6)
|
||||
|
||||
def to_markdown(self) -> str:
|
||||
return f"{'#' * self.level} {self.text}"
|
||||
@@ -55,9 +53,7 @@ class HeadingElement(MarkdownElement):
|
||||
class ImageElement(MarkdownElement):
|
||||
type: Literal["image"] = "image"
|
||||
src: str
|
||||
"""图片来源 (URL或data URI)"""
|
||||
alt: str = "image"
|
||||
"""图片的替代文本"""
|
||||
|
||||
def to_markdown(self) -> str:
|
||||
return f""
|
||||
@@ -66,9 +62,7 @@ class ImageElement(MarkdownElement):
|
||||
class CodeElement(MarkdownElement):
|
||||
type: Literal["code"] = "code"
|
||||
code: str
|
||||
"""代码字符串"""
|
||||
language: str = ""
|
||||
"""代码语言,用于语法高亮"""
|
||||
|
||||
def to_markdown(self) -> str:
|
||||
return f"```{self.language}\n{self.code}\n```"
|
||||
@@ -77,7 +71,6 @@ class CodeElement(MarkdownElement):
|
||||
class RawHtmlElement(MarkdownElement):
|
||||
type: Literal["raw_html"] = "raw_html"
|
||||
html: str
|
||||
"""原始HTML字符串"""
|
||||
|
||||
def to_markdown(self) -> str:
|
||||
return self.html
|
||||
@@ -86,11 +79,8 @@ class RawHtmlElement(MarkdownElement):
|
||||
class TableElement(MarkdownElement):
|
||||
type: Literal["table"] = "table"
|
||||
headers: list[str]
|
||||
"""表格的表头列表"""
|
||||
rows: list[list[str]]
|
||||
"""表格的数据行列表"""
|
||||
alignments: list[Literal["left", "center", "right"]] | None = None
|
||||
"""每列的对齐方式"""
|
||||
|
||||
def to_markdown(self) -> str:
|
||||
header_row = "| " + " | ".join(self.headers) + " |"
|
||||
@@ -112,10 +102,7 @@ class TableElement(MarkdownElement):
|
||||
|
||||
|
||||
class ContainerElement(MarkdownElement):
|
||||
content: list[MarkdownElement] = Field(
|
||||
default_factory=list, description="容器内包含的Markdown元素列表"
|
||||
)
|
||||
"""容器内包含的Markdown元素列表"""
|
||||
content: list[MarkdownElement] = Field(default_factory=list)
|
||||
|
||||
|
||||
class QuoteElement(ContainerElement):
|
||||
@@ -134,7 +121,6 @@ class ListItemElement(ContainerElement):
|
||||
class ListElement(ContainerElement):
|
||||
type: Literal["list"] = "list"
|
||||
ordered: bool = False
|
||||
"""是否为有序列表 (例如 1., 2.)"""
|
||||
|
||||
def to_markdown(self) -> str:
|
||||
lines = []
|
||||
@@ -151,7 +137,6 @@ class ComponentElement(MarkdownElement):
|
||||
|
||||
type: Literal["component"] = "component"
|
||||
component: RenderableComponent
|
||||
"""嵌入在Markdown中的可渲染组件"""
|
||||
|
||||
def to_markdown(self) -> str:
|
||||
return ""
|
||||
@@ -161,15 +146,9 @@ class MarkdownData(ContainerComponent):
|
||||
"""Markdown转图片的数据模型"""
|
||||
|
||||
style_name: str | None = None
|
||||
"""Markdown内容的样式名称"""
|
||||
elements: list[MarkdownElement] = Field(
|
||||
default_factory=list, description="构成Markdown文档的元素列表"
|
||||
)
|
||||
"""构成Markdown文档的元素列表"""
|
||||
elements: list[MarkdownElement] = Field(default_factory=list)
|
||||
width: int = 800
|
||||
"""最终渲染图片的宽度"""
|
||||
css_path: str | None = None
|
||||
"""自定义CSS文件的绝对路径"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
@@ -201,6 +180,7 @@ class MarkdownData(ContainerComponent):
|
||||
logger.warning(f"Markdown自定义CSS文件不存在: {self.css_path}")
|
||||
else:
|
||||
style_name = self.style_name or "light"
|
||||
# 使用上下文对象来解析路径
|
||||
css_path = await context.theme_manager.resolve_markdown_style_path(
|
||||
style_name, context
|
||||
)
|
||||
|
||||
@@ -22,32 +22,21 @@ class NotebookElement(BaseModel):
|
||||
"component",
|
||||
]
|
||||
text: str | None = None
|
||||
"""元素的文本内容 (用于标题、段落、引用)"""
|
||||
level: int | None = None
|
||||
"""标题的级别 (1-4)"""
|
||||
src: str | None = None
|
||||
"""图片的来源 (URL或data URI)"""
|
||||
caption: str | None = None
|
||||
"""图片的说明文字"""
|
||||
code: str | None = None
|
||||
"""代码块的内容"""
|
||||
language: str | None = None
|
||||
"""代码块的语言"""
|
||||
data: list[str] | None = None
|
||||
"""列表项的内容列表"""
|
||||
ordered: bool | None = None
|
||||
"""是否为有序列表"""
|
||||
component: RenderableComponent | None = None
|
||||
"""嵌入的自定义可渲染组件"""
|
||||
|
||||
|
||||
class NotebookData(ContainerComponent):
|
||||
"""Notebook转图片的数据模型"""
|
||||
|
||||
style_name: str | None = None
|
||||
"""Notebook的样式名称"""
|
||||
elements: list[NotebookElement]
|
||||
"""构成Notebook页面的元素列表"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -8,7 +8,6 @@ from .text import TextSpan
|
||||
|
||||
__all__ = [
|
||||
"BaseCell",
|
||||
"ComponentCell",
|
||||
"ImageCell",
|
||||
"ProgressBarCell",
|
||||
"RichTextCell",
|
||||
@@ -64,18 +63,8 @@ class RichTextCell(BaseCell):
|
||||
|
||||
type: Literal["rich_text"] = "rich_text" # type: ignore
|
||||
spans: list[TextSpan] = Field(default_factory=list, description="文本片段列表")
|
||||
"""文本片段列表"""
|
||||
direction: Literal["column", "row"] = Field("column", description="片段排列方向")
|
||||
"""片段排列方向"""
|
||||
gap: str = Field("4px", description="片段之间的间距")
|
||||
"""片段之间的间距"""
|
||||
|
||||
|
||||
class ComponentCell(BaseCell):
|
||||
"""一个通用的单元格,可以容纳任何可渲染的组件。"""
|
||||
|
||||
type: str = "component"
|
||||
component: RenderableComponent
|
||||
|
||||
|
||||
TableCell = (
|
||||
@@ -84,7 +73,6 @@ TableCell = (
|
||||
| StatusBadgeCell
|
||||
| ProgressBarCell
|
||||
| RichTextCell
|
||||
| ComponentCell
|
||||
| str
|
||||
| int
|
||||
| float
|
||||
@@ -96,23 +84,16 @@ class TableData(RenderableComponent):
|
||||
"""通用表格的数据模型"""
|
||||
|
||||
style_name: str | None = None
|
||||
"""应用于表格容器的样式名称"""
|
||||
title: str = Field(..., description="表格主标题")
|
||||
"""表格主标题"""
|
||||
tip: str | None = Field(None, description="表格下方的提示信息")
|
||||
"""表格下方的提示信息"""
|
||||
headers: list[str] = Field(default_factory=list, description="表头列表")
|
||||
"""表头列表"""
|
||||
rows: list[list[TableCell]] = Field(default_factory=list, description="数据行列表")
|
||||
"""数据行列表"""
|
||||
column_alignments: list[Literal["left", "center", "right"]] | None = Field(
|
||||
default=None, description="每列的对齐方式"
|
||||
)
|
||||
"""每列的对齐方式"""
|
||||
column_widths: list[str | int] | None = Field(
|
||||
default=None, description="每列的宽度 (e.g., ['50px', 'auto', 100])"
|
||||
)
|
||||
"""每列的宽度 (e.g., ['50px', 'auto', 100])"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -1,8 +1,6 @@
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from pydantic import Field
|
||||
|
||||
from .base import RenderableComponent
|
||||
|
||||
__all__ = ["TemplateComponent"]
|
||||
@@ -12,11 +10,8 @@ class TemplateComponent(RenderableComponent):
|
||||
"""基于独立模板文件的UI组件"""
|
||||
|
||||
_is_standalone_template: bool = True
|
||||
"""标记此组件为独立模板"""
|
||||
template_path: str | Path = Field(..., description="指向HTML模板文件的路径")
|
||||
"""指向HTML模板文件的路径"""
|
||||
data: dict[str, Any] = Field(..., description="传递给模板的上下文数据字典")
|
||||
"""传递给模板的上下文数据字典"""
|
||||
template_path: str | Path
|
||||
data: dict[str, Any]
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -23,11 +23,9 @@ class TextData(RenderableComponent):
|
||||
"""轻量级富文本组件的数据模型"""
|
||||
|
||||
spans: list[TextSpan] = Field(default_factory=list, description="文本片段列表")
|
||||
"""文本片段列表"""
|
||||
align: Literal["left", "right", "center"] = Field(
|
||||
"left", description="整体文本对齐方式"
|
||||
)
|
||||
"""整体文本对齐方式"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -13,35 +13,25 @@ class HelpItem(BaseModel):
|
||||
"""帮助菜单中的单个功能项"""
|
||||
|
||||
name: str
|
||||
"""功能名称"""
|
||||
description: str
|
||||
"""功能描述"""
|
||||
usage: str
|
||||
"""功能用法说明"""
|
||||
|
||||
|
||||
class HelpCategory(BaseModel):
|
||||
"""帮助菜单中的一个功能类别"""
|
||||
|
||||
title: str
|
||||
"""分类标题"""
|
||||
icon_svg_path: str
|
||||
"""分类图标的SVG路径数据"""
|
||||
items: list[HelpItem]
|
||||
"""该分类下的功能项列表"""
|
||||
|
||||
|
||||
class PluginHelpPageData(RenderableComponent):
|
||||
"""通用插件帮助页面的数据模型"""
|
||||
|
||||
style_name: str | None = None
|
||||
"""页面样式名称"""
|
||||
bot_nickname: str
|
||||
"""机器人昵称"""
|
||||
page_title: str
|
||||
"""页面主标题"""
|
||||
categories: list[HelpCategory]
|
||||
"""帮助分类列表"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -13,43 +13,29 @@ class PluginMenuItem(BaseModel):
|
||||
"""插件菜单中的单个插件项"""
|
||||
|
||||
id: str
|
||||
"""插件的唯一ID"""
|
||||
name: str
|
||||
"""插件名称"""
|
||||
status: bool
|
||||
"""插件在当前群组的开关状态"""
|
||||
has_superuser_help: bool
|
||||
"""插件是否有超级用户专属帮助"""
|
||||
commands: list[str] = Field(default_factory=list, description="插件的主要命令列表")
|
||||
"""插件的主要命令列表"""
|
||||
commands: list[str] = Field(default_factory=list)
|
||||
|
||||
|
||||
class PluginMenuCategory(BaseModel):
|
||||
"""插件菜单中的一个分类"""
|
||||
|
||||
name: str
|
||||
"""插件分类名称"""
|
||||
items: list[PluginMenuItem] = Field(..., description="该分类下的插件项列表")
|
||||
"""该分类下的插件项列表"""
|
||||
items: list[PluginMenuItem]
|
||||
|
||||
|
||||
class PluginMenuData(RenderableComponent):
|
||||
"""通用插件帮助菜单的数据模型"""
|
||||
|
||||
style_name: str | None = None
|
||||
"""页面样式名称"""
|
||||
bot_name: str
|
||||
"""机器人名称"""
|
||||
bot_avatar_url: str
|
||||
"""机器人头像URL"""
|
||||
is_detail: bool
|
||||
"""是否为详细菜单模式"""
|
||||
plugin_count: int
|
||||
"""总插件数量"""
|
||||
active_count: int
|
||||
"""已启用插件数量"""
|
||||
categories: list[PluginMenuCategory]
|
||||
"""插件分类列表"""
|
||||
|
||||
@property
|
||||
def template_name(self) -> str:
|
||||
|
||||
@@ -4,7 +4,7 @@ from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
|
||||
from nonebot_plugin_alconna import UniMessage
|
||||
from nonebot_plugin_htmlrender.browser import get_browser
|
||||
from nonebot_plugin_htmlrender import get_browser
|
||||
from playwright.async_api import Page
|
||||
|
||||
from zhenxun.utils.message import MessageUtils
|
||||
|
||||
@@ -40,6 +40,9 @@ RELEASE_SOURCE_FORMAT = (
|
||||
GIT_API_COMMIT_FORMAT = "https://api.github.com/repos/{owner}/{repo}/commits/{branch}"
|
||||
"""git api commit地址格式"""
|
||||
|
||||
GIT_API_COMMIT_LIST_FORMAT = "https://api.github.com/repos/{owner}/{repo}/commits"
|
||||
"""git api 列出commits的地址格式"""
|
||||
|
||||
GIT_API_PROXY_COMMIT_FORMAT = (
|
||||
"https://git-api.zhenxun.org/repos/{owner}/{repo}/commits/{branch}"
|
||||
)
|
||||
|
||||
@@ -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=640"
|
||||
return f"http://q1.qlogo.cn/g?b=qq&nk={user_id}&s=160"
|
||||
else:
|
||||
return f"https://q.qlogo.cn/qqapp/{appid}/{user_id}/640"
|
||||
|
||||
|
||||
@@ -348,6 +348,11 @@ class AliyunCodeupManager(BaseRepoManager):
|
||||
if not self.config.aliyun_codeup.organization_id:
|
||||
raise AuthenticationError("阿里云CodeUp")
|
||||
|
||||
async def get_latest_commit(self, repo_url: str, branch: str = "main") -> str:
|
||||
"""获取阿里云CodeUp仓库指定分支的最新提交哈希值。"""
|
||||
repo_name = repo_url.split("/tree/")[0].split("/")[-1].replace(".git", "")
|
||||
return await self._get_newest_commit(repo_name, branch)
|
||||
|
||||
async def _get_newest_commit(self, repo_name: str, branch: str) -> str:
|
||||
"""
|
||||
获取仓库最新提交ID
|
||||
|
||||
@@ -117,6 +117,20 @@ class BaseRepoManager(ABC):
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
async def get_latest_commit(self, repo_url: str, branch: str = "main") -> str:
|
||||
"""
|
||||
获取仓库指定分支的最新提交哈希值。
|
||||
|
||||
参数:
|
||||
repo_url: 仓库URL或名称。
|
||||
branch: 分支名称。
|
||||
|
||||
返回:
|
||||
str: 最新的提交哈希值。
|
||||
"""
|
||||
pass
|
||||
|
||||
async def save_file_content(self, content: bytes, local_path: Path) -> int:
|
||||
"""
|
||||
保存文件内容
|
||||
|
||||
@@ -11,6 +11,7 @@ from httpx import Response
|
||||
|
||||
from zhenxun.services.log import logger
|
||||
from zhenxun.utils.github_utils import GithubUtils
|
||||
from zhenxun.utils.github_utils.const import GIT_API_COMMIT_LIST_FORMAT
|
||||
from zhenxun.utils.github_utils.models import AliyunTreeType, GitHubStrategy, TreeType
|
||||
from zhenxun.utils.http_utils import AsyncHttpx
|
||||
from zhenxun.utils.utils import is_binary_file
|
||||
@@ -326,7 +327,7 @@ class RepoFileManager:
|
||||
|
||||
# 获取仓库树信息
|
||||
strategy = GitHubStrategy()
|
||||
strategy.body = await strategy.parse_repo_info(repo_info)
|
||||
strategy.body = await GitHubStrategy.parse_repo_info(repo_info)
|
||||
|
||||
# 处理目录路径,确保格式正确
|
||||
if directory_path and not directory_path.endswith("/") and recursive:
|
||||
@@ -480,7 +481,7 @@ class RepoFileManager:
|
||||
target_dir: Path | None = None,
|
||||
) -> FileDownloadResult:
|
||||
"""
|
||||
下载多个文件
|
||||
下载单个文件
|
||||
|
||||
参数:
|
||||
repo_url: 仓库URL
|
||||
@@ -633,3 +634,38 @@ class RepoFileManager:
|
||||
result.success = False
|
||||
result.error_message = str(e)
|
||||
return result
|
||||
|
||||
async def get_file_last_commit_date(
|
||||
self, repo_url: str, file_path: str
|
||||
) -> str | None:
|
||||
"""
|
||||
获取 GitHub 仓库中指定文件的最新提交日期。
|
||||
|
||||
参数:
|
||||
repo_url: 仓库的URL。
|
||||
file_path: 文件在仓库中的路径。
|
||||
|
||||
返回:
|
||||
str | None: "YYYY-MM-DD" 格式的日期字符串,如果失败则返回 None。
|
||||
"""
|
||||
try:
|
||||
repo_info = GithubUtils.parse_github_url(repo_url)
|
||||
api_url = GIT_API_COMMIT_LIST_FORMAT.format(
|
||||
owner=repo_info.owner, repo=repo_info.repo
|
||||
)
|
||||
params = {
|
||||
"sha": repo_info.branch,
|
||||
"path": file_path,
|
||||
"page": 1,
|
||||
"per_page": 1,
|
||||
}
|
||||
|
||||
data = await AsyncHttpx.get_json(api_url, params=params)
|
||||
if data and isinstance(data, list) and data[0]:
|
||||
date_str = data[0]["commit"]["committer"]["date"]
|
||||
return date_str.split("T")[0]
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"获取 {repo_url} 中 {file_path} 的 commit 日期失败", LOG_COMMAND, e=e
|
||||
)
|
||||
return None
|
||||
|
||||
@@ -320,6 +320,12 @@ class GithubManager(BaseRepoManager):
|
||||
logger.error("获取提交信息失败", LOG_COMMAND, e=e)
|
||||
return None
|
||||
|
||||
async def get_latest_commit(self, repo_url: str, branch: str = "main") -> str:
|
||||
"""获取GitHub仓库指定分支的最新提交哈希值。"""
|
||||
repo_info = GithubUtils.parse_github_url(repo_url)
|
||||
repo_name = repo_info.repo.replace(".git", "")
|
||||
return await self._get_newest_commit(repo_info.owner, repo_name, branch)
|
||||
|
||||
async def _get_newest_commit(self, owner: str, repo: str, branch: str) -> str:
|
||||
"""
|
||||
获取仓库最新提交ID
|
||||
|
||||
@@ -7,7 +7,6 @@ import base64
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shutil
|
||||
import tempfile
|
||||
|
||||
from zhenxun.services.log import logger
|
||||
|
||||
@@ -146,85 +145,80 @@ async def sparse_checkout_clone(
|
||||
target_dir: Path,
|
||||
) -> None:
|
||||
"""
|
||||
使用 git 稀疏检出克隆指定路径到目标目录(在临时目录中操作)。
|
||||
使用 git 稀疏检出克隆指定路径到目标目录(完全独立于主项目 git)。
|
||||
|
||||
关键保障:
|
||||
- 在临时目录中执行所有 git 操作,避免影响 target_dir 中的现有内容
|
||||
- 只操作 target_dir/sparse_path 路径,不影响 target_dir 其他内容
|
||||
- 在 target_dir 下检测/初始化 .git,所有 git 操作均以 cwd=target_dir 执行
|
||||
- 强制拉取与工作区覆盖: fetch --force、checkout -B、reset --hard、clean -xdf
|
||||
- 反复设置 sparse-checkout 路径,确保路径更新生效
|
||||
"""
|
||||
target_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if not await check_git():
|
||||
raise GitUnavailableError()
|
||||
|
||||
# 在临时目录中进行 git 操作
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
temp_path = Path(temp_dir)
|
||||
|
||||
# 初始化临时目录为 git 仓库
|
||||
success, out, err = await run_git_command("init", temp_path)
|
||||
git_dir = target_dir / ".git"
|
||||
if not git_dir.exists():
|
||||
success, out, err = await run_git_command("init", target_dir)
|
||||
if not success:
|
||||
raise RuntimeError(f"git init 失败: {err or out}")
|
||||
success, out, err = await run_git_command(
|
||||
f"remote add origin {repo_url}", temp_path
|
||||
f"remote add origin {repo_url}", target_dir
|
||||
)
|
||||
if not success:
|
||||
raise RuntimeError(f"添加远程失败: {err or out}")
|
||||
|
||||
# 启用稀疏检出(使用 --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 模式,直接指定要检出的具体路径
|
||||
else:
|
||||
success, out, err = await run_git_command(
|
||||
f"sparse-checkout set {sparse_path}/", temp_path
|
||||
f"remote set-url origin {repo_url}", target_dir
|
||||
)
|
||||
if not success:
|
||||
raise RuntimeError(f"配置稀疏路径失败: {err or out}")
|
||||
# 兜底尝试添加
|
||||
await run_git_command(f"remote add origin {repo_url}", 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}")
|
||||
# 启用稀疏检出(使用 --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"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)}")
|
||||
# 设置需要检出的路径(每次都覆盖配置)
|
||||
if not sparse_path:
|
||||
raise RuntimeError("sparse-checkout 路径不能为空")
|
||||
|
||||
# 强制对齐工作区
|
||||
await run_git_command(f"reset --hard origin/{branch}", temp_path)
|
||||
await run_git_command("clean -xdf", temp_path)
|
||||
# 使用 --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}")
|
||||
|
||||
# 将检出的文件移动到目标位置
|
||||
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"fetch --force --depth 1 origin {branch}", target_dir
|
||||
)
|
||||
if not success:
|
||||
raise RuntimeError(f"fetch 失败: {err or out}")
|
||||
|
||||
# 如果目标路径已存在,先清理
|
||||
if target_path.exists():
|
||||
if target_path.is_dir():
|
||||
shutil.rmtree(target_path)
|
||||
else:
|
||||
target_path.unlink()
|
||||
# 使用远端强制更新本地分支并覆盖工作区
|
||||
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)}")
|
||||
|
||||
# 移动整个目录结构到目标位置
|
||||
shutil.move(str(source_path), str(target_path))
|
||||
# 强制对齐工作区
|
||||
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)
|
||||
|
||||
|
||||
def prepare_aliyun_url(repo_url: str) -> str:
|
||||
|
||||
Reference in New Issue
Block a user