mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2025-12-15 14:22:55 +08:00
Compare commits
3 Commits
b56f162db5
...
a28963a7ee
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a28963a7ee | ||
|
|
fb0a9813e1 | ||
|
|
e3d49c7105 |
@ -84,13 +84,16 @@ async def _(
|
|||||||
):
|
):
|
||||||
result = ""
|
result = ""
|
||||||
await MessageUtils.build_message("正在进行检查更新...").send(reply_to=True)
|
await MessageUtils.build_message("正在进行检查更新...").send(reply_to=True)
|
||||||
|
|
||||||
|
if not ver_type.available:
|
||||||
|
result += await UpdateManager.check_version()
|
||||||
|
logger.info("查看当前版本...", "检查更新", session=session)
|
||||||
|
await MessageUtils.build_message(result).finish()
|
||||||
|
return
|
||||||
|
|
||||||
ver_type_str = ver_type.result
|
ver_type_str = ver_type.result
|
||||||
source_str = source.result
|
source_str = source.result
|
||||||
if ver_type_str in {"main", "release"}:
|
if ver_type_str in {"main", "release"}:
|
||||||
if not ver_type.available:
|
|
||||||
result += await UpdateManager.check_version()
|
|
||||||
logger.info("查看当前版本...", "检查更新", session=session)
|
|
||||||
await MessageUtils.build_message(result).finish()
|
|
||||||
try:
|
try:
|
||||||
result += await UpdateManager.update_zhenxun(
|
result += await UpdateManager.update_zhenxun(
|
||||||
bot,
|
bot,
|
||||||
|
|||||||
@ -1,37 +1,135 @@
|
|||||||
|
import asyncio
|
||||||
from typing import Literal
|
from typing import Literal
|
||||||
|
|
||||||
from nonebot.adapters import Bot
|
from nonebot.adapters import Bot
|
||||||
|
from packaging.specifiers import SpecifierSet
|
||||||
|
from packaging.version import InvalidVersion, Version
|
||||||
|
|
||||||
from zhenxun.services.log import logger
|
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.virtual_env_package_manager import VirtualEnvPackageManager
|
||||||
from zhenxun.utils.manager.zhenxun_repo_manager import (
|
from zhenxun.utils.manager.zhenxun_repo_manager import (
|
||||||
ZhenxunRepoConfig,
|
ZhenxunRepoConfig,
|
||||||
ZhenxunRepoManager,
|
ZhenxunRepoManager,
|
||||||
)
|
)
|
||||||
from zhenxun.utils.platform import PlatformUtils
|
from zhenxun.utils.platform import PlatformUtils
|
||||||
|
from zhenxun.utils.repo_utils import RepoFileManager
|
||||||
|
|
||||||
LOG_COMMAND = "AutoUpdate"
|
LOG_COMMAND = "AutoUpdate"
|
||||||
|
|
||||||
|
|
||||||
class UpdateManager:
|
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
|
@classmethod
|
||||||
async def check_version(cls) -> str:
|
async def check_version(cls) -> str:
|
||||||
"""检查更新版本
|
"""检查真寻和资源的版本"""
|
||||||
|
bot_cur_version = cls.__get_version()
|
||||||
|
|
||||||
返回:
|
release_task = ZhenxunRepoManager.zhenxun_get_latest_releases_data()
|
||||||
str: 更新信息
|
dev_version_task = RepoFileManager.get_file_content(
|
||||||
"""
|
ZhenxunRepoConfig.ZHENXUN_BOT_GITHUB_URL, "__version__"
|
||||||
cur_version = cls.__get_version()
|
|
||||||
release_data = await ZhenxunRepoManager.zhenxun_get_latest_releases_data()
|
|
||||||
if not release_data:
|
|
||||||
return "检查更新获取版本失败..."
|
|
||||||
return (
|
|
||||||
"检测到当前版本更新\n"
|
|
||||||
f"当前版本:{cur_version}\n"
|
|
||||||
f"最新版本:{release_data.get('name')}\n"
|
|
||||||
f"创建日期:{release_data.get('created_at')}\n"
|
|
||||||
f"更新内容:\n{release_data.get('body')}"
|
|
||||||
)
|
)
|
||||||
|
bot_commit_date_task = cls._get_latest_commit_date(
|
||||||
|
"HibiKier", "zhenxun_bot", "__version__"
|
||||||
|
)
|
||||||
|
res_commit_date_task = cls._get_latest_commit_date(
|
||||||
|
"zhenxun-org", "zhenxun-bot-resources", "__version__"
|
||||||
|
)
|
||||||
|
|
||||||
|
(
|
||||||
|
release_data,
|
||||||
|
dev_version_text,
|
||||||
|
bot_commit_date,
|
||||||
|
res_commit_date,
|
||||||
|
) = await asyncio.gather(
|
||||||
|
release_task,
|
||||||
|
dev_version_task,
|
||||||
|
bot_commit_date_task,
|
||||||
|
res_commit_date_task,
|
||||||
|
return_exceptions=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
if isinstance(release_data, dict):
|
||||||
|
bot_release_version = release_data.get("name", "获取失败")
|
||||||
|
bot_release_date = release_data.get("created_at", "").split("T")[0]
|
||||||
|
else:
|
||||||
|
bot_release_version = "获取失败"
|
||||||
|
bot_release_date = "获取失败"
|
||||||
|
logger.warning(f"获取 Bot release 信息失败: {release_data}")
|
||||||
|
|
||||||
|
if isinstance(dev_version_text, str):
|
||||||
|
bot_dev_version = dev_version_text.split(":")[-1].strip()
|
||||||
|
else:
|
||||||
|
bot_dev_version = "获取失败"
|
||||||
|
bot_commit_date = "获取失败"
|
||||||
|
logger.warning(f"获取 Bot dev 版本信息失败: {dev_version_text}")
|
||||||
|
|
||||||
|
bot_update_hint = ""
|
||||||
|
try:
|
||||||
|
cur_base_v = bot_cur_version.split("-")[0].lstrip("v")
|
||||||
|
dev_base_v = bot_dev_version.split("-")[0].lstrip("v")
|
||||||
|
|
||||||
|
if Version(cur_base_v) < Version(dev_base_v):
|
||||||
|
bot_update_hint = "\n-> 发现新开发版本, 可用 `检查更新 main` 更新"
|
||||||
|
elif (
|
||||||
|
Version(cur_base_v) == Version(dev_base_v)
|
||||||
|
and bot_cur_version != bot_dev_version
|
||||||
|
):
|
||||||
|
bot_update_hint = "\n-> 发现新开发版本, 可用 `检查更新 main` 更新"
|
||||||
|
except (InvalidVersion, TypeError, IndexError):
|
||||||
|
if bot_cur_version != bot_dev_version and bot_dev_version != "获取失败":
|
||||||
|
bot_update_hint = "\n-> 发现新开发版本, 可用 `检查更新 main` 更新"
|
||||||
|
|
||||||
|
bot_update_info = (
|
||||||
|
f"当前版本: {bot_cur_version}\n"
|
||||||
|
f"最新开发版: {bot_dev_version} (更新于: {bot_commit_date})\n"
|
||||||
|
f"最新正式版: {bot_release_version} (发布于: {bot_release_date})"
|
||||||
|
f"{bot_update_hint}"
|
||||||
|
)
|
||||||
|
|
||||||
|
res_version_file = ZhenxunRepoConfig.RESOURCE_PATH / "__version__"
|
||||||
|
res_cur_version = "未找到"
|
||||||
|
if res_version_file.exists():
|
||||||
|
if text := res_version_file.open(encoding="utf8").readline():
|
||||||
|
res_cur_version = text.split(":")[-1].strip()
|
||||||
|
|
||||||
|
res_latest_version = "获取失败"
|
||||||
|
try:
|
||||||
|
res_latest_version_text = await RepoFileManager.get_file_content(
|
||||||
|
ZhenxunRepoConfig.RESOURCE_GITHUB_URL, "__version__"
|
||||||
|
)
|
||||||
|
res_latest_version = res_latest_version_text.split(":")[-1].strip()
|
||||||
|
except Exception as e:
|
||||||
|
res_commit_date = "获取失败"
|
||||||
|
logger.warning(f"获取资源版本信息失败: {e}")
|
||||||
|
|
||||||
|
res_update_hint = ""
|
||||||
|
try:
|
||||||
|
if Version(res_cur_version) < Version(res_latest_version):
|
||||||
|
res_update_hint = "\n-> 发现新资源版本, 可用 `检查更新 resource` 更新"
|
||||||
|
except (InvalidVersion, TypeError):
|
||||||
|
pass
|
||||||
|
|
||||||
|
res_update_info = (
|
||||||
|
f"当前版本: {res_cur_version}\n"
|
||||||
|
f"最新版本: {res_latest_version} (更新于: {res_commit_date})"
|
||||||
|
f"{res_update_hint}"
|
||||||
|
)
|
||||||
|
|
||||||
|
return f"『绪山真寻 Bot』\n{bot_update_info}\n\n『真寻资源』\n{res_update_info}"
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
async def update_webui(
|
async def update_webui(
|
||||||
@ -125,6 +223,7 @@ class UpdateManager:
|
|||||||
f"检测真寻已更新,当前版本:{cur_version}\n开始更新...",
|
f"检测真寻已更新,当前版本:{cur_version}\n开始更新...",
|
||||||
user_id,
|
user_id,
|
||||||
)
|
)
|
||||||
|
result_message = ""
|
||||||
if zip:
|
if zip:
|
||||||
new_version = await ZhenxunRepoManager.zhenxun_zip_update(version_type)
|
new_version = await ZhenxunRepoManager.zhenxun_zip_update(version_type)
|
||||||
await PlatformUtils.send_superuser(
|
await PlatformUtils.send_superuser(
|
||||||
@ -133,7 +232,7 @@ class UpdateManager:
|
|||||||
await VirtualEnvPackageManager.install_requirement(
|
await VirtualEnvPackageManager.install_requirement(
|
||||||
ZhenxunRepoConfig.REQUIREMENTS_FILE
|
ZhenxunRepoConfig.REQUIREMENTS_FILE
|
||||||
)
|
)
|
||||||
return (
|
result_message = (
|
||||||
f"版本更新完成!\n版本: {cur_version} -> {new_version}\n"
|
f"版本更新完成!\n版本: {cur_version} -> {new_version}\n"
|
||||||
"请重新启动真寻以完成更新!"
|
"请重新启动真寻以完成更新!"
|
||||||
)
|
)
|
||||||
@ -155,13 +254,54 @@ class UpdateManager:
|
|||||||
await VirtualEnvPackageManager.install_requirement(
|
await VirtualEnvPackageManager.install_requirement(
|
||||||
ZhenxunRepoConfig.REQUIREMENTS_FILE
|
ZhenxunRepoConfig.REQUIREMENTS_FILE
|
||||||
)
|
)
|
||||||
return (
|
result_message = (
|
||||||
f"版本更新完成!\n"
|
f"版本更新完成!\n"
|
||||||
f"版本: {cur_version} -> {result.new_version}\n"
|
f"版本: {cur_version} -> {result.new_version}\n"
|
||||||
f"变更文件个数: {len(result.changed_files)}"
|
f"变更文件个数: {len(result.changed_files)}"
|
||||||
f"{'' if source == 'git' else '(阿里云更新不支持查看变更文件)'}\n"
|
f"{'' if source == 'git' else '(阿里云更新不支持查看变更文件)'}\n"
|
||||||
"请重新启动真寻以完成更新!"
|
"请重新启动真寻以完成更新!"
|
||||||
)
|
)
|
||||||
|
resource_warning = ""
|
||||||
|
if version_type == "main":
|
||||||
|
try:
|
||||||
|
spec_content = await RepoFileManager.get_file_content(
|
||||||
|
ZhenxunRepoConfig.ZHENXUN_BOT_GITHUB_URL, "resources.spec"
|
||||||
|
)
|
||||||
|
required_spec_str = None
|
||||||
|
for line in spec_content.splitlines():
|
||||||
|
if line.startswith("require_resources_version:"):
|
||||||
|
required_spec_str = line.split(":", 1)[1].strip().strip("\"'")
|
||||||
|
break
|
||||||
|
if required_spec_str:
|
||||||
|
res_version_file = ZhenxunRepoConfig.RESOURCE_PATH / "__version__"
|
||||||
|
local_res_version_str = "0.0.0"
|
||||||
|
if res_version_file.exists():
|
||||||
|
if text := res_version_file.open(encoding="utf8").readline():
|
||||||
|
local_res_version_str = text.split(":")[-1].strip()
|
||||||
|
|
||||||
|
spec = SpecifierSet(required_spec_str)
|
||||||
|
local_ver = Version(local_res_version_str)
|
||||||
|
if not spec.contains(local_ver):
|
||||||
|
warning_header = (
|
||||||
|
f"⚠️ **资源版本不兼容!**\n"
|
||||||
|
f"当前代码需要资源版本: `{required_spec_str}`\n"
|
||||||
|
f"您当前的资源版本是: `{local_res_version_str}`\n"
|
||||||
|
"**将自动为您更新资源文件...**"
|
||||||
|
)
|
||||||
|
await PlatformUtils.send_superuser(bot, warning_header, user_id)
|
||||||
|
resource_update_source = None if zip else source
|
||||||
|
resource_update_result = await cls.update_resources(
|
||||||
|
source=resource_update_source, force=force
|
||||||
|
)
|
||||||
|
resource_warning = (
|
||||||
|
f"\n\n{warning_header}\n{resource_update_result}"
|
||||||
|
)
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(f"检查资源版本兼容性时出错: {e}", LOG_COMMAND, e=e)
|
||||||
|
resource_warning = (
|
||||||
|
"\n\n⚠️ 检查资源版本兼容性时出错,建议手动运行 `检查更新 resource`"
|
||||||
|
)
|
||||||
|
return result_message + resource_warning
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def __get_version(cls) -> str:
|
def __get_version(cls) -> str:
|
||||||
|
|||||||
@ -132,7 +132,7 @@ async def gold_rank(session: Uninfo, group_id: str | None, num: int) -> bytes |
|
|||||||
else TextCell(content=""),
|
else TextCell(content=""),
|
||||||
TextCell(content=uid2name.get(user[0]) or user[0]),
|
TextCell(content=uid2name.get(user[0]) or user[0]),
|
||||||
TextCell(content=str(user[1]), bold=True),
|
TextCell(content=str(user[1]), bold=True),
|
||||||
ImageCell(src=platform_path.resolve().as_uri())
|
ImageCell(src=platform_path)
|
||||||
if (platform_path := PLATFORM_PATH.get(platform))
|
if (platform_path := PLATFORM_PATH.get(platform))
|
||||||
else TextCell(content=""),
|
else TextCell(content=""),
|
||||||
]
|
]
|
||||||
@ -532,15 +532,15 @@ class ShopManage:
|
|||||||
icon = ""
|
icon = ""
|
||||||
if prop.icon:
|
if prop.icon:
|
||||||
icon_path = ICON_PATH / prop.icon
|
icon_path = ICON_PATH / prop.icon
|
||||||
icon = (icon_path, 33, 33) if icon_path.exists() else ""
|
icon = icon_path if icon_path.exists() else ""
|
||||||
|
|
||||||
table_rows.append(
|
table_rows.append(
|
||||||
[
|
[
|
||||||
icon,
|
ImageCell(src=icon, height=33, width=33),
|
||||||
i,
|
TextCell(content=i),
|
||||||
prop.goods_name,
|
TextCell(content=prop.goods_name),
|
||||||
user.props[prop_uuid],
|
TextCell(content=user.props[prop_uuid]),
|
||||||
prop.goods_description,
|
TextCell(content=prop.goods_description),
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@ -91,7 +91,7 @@ class SignManage:
|
|||||||
TextCell(content=uid2name.get(user[0]) or user[0]),
|
TextCell(content=uid2name.get(user[0]) or user[0]),
|
||||||
TextCell(content=str(user[1]), bold=True),
|
TextCell(content=str(user[1]), bold=True),
|
||||||
TextCell(content=str(user[2])),
|
TextCell(content=str(user[2])),
|
||||||
ImageCell(src=platform_path.resolve().as_uri())
|
ImageCell(src=platform_path)
|
||||||
if (platform_path := PLATFORM_PATH.get(platform))
|
if (platform_path := PLATFORM_PATH.get(platform))
|
||||||
else TextCell(content=""),
|
else TextCell(content=""),
|
||||||
]
|
]
|
||||||
|
|||||||
@ -1,6 +1,8 @@
|
|||||||
|
from pathlib import Path
|
||||||
from typing import Literal
|
from typing import Literal
|
||||||
|
|
||||||
from pydantic import BaseModel, Field
|
from nonebot.compat import field_validator
|
||||||
|
from pydantic import BaseModel
|
||||||
|
|
||||||
from ...models.components.progress_bar import ProgressBar
|
from ...models.components.progress_bar import ProgressBar
|
||||||
from .base import RenderableComponent
|
from .base import RenderableComponent
|
||||||
@ -28,7 +30,7 @@ class TextCell(BaseCell):
|
|||||||
"""文本单元格"""
|
"""文本单元格"""
|
||||||
|
|
||||||
type: Literal["text"] = "text" # type: ignore
|
type: Literal["text"] = "text" # type: ignore
|
||||||
content: str
|
content: str | float
|
||||||
bold: bool = False
|
bold: bool = False
|
||||||
color: str | None = None
|
color: str | None = None
|
||||||
|
|
||||||
@ -37,12 +39,18 @@ class ImageCell(BaseCell):
|
|||||||
"""图片单元格"""
|
"""图片单元格"""
|
||||||
|
|
||||||
type: Literal["image"] = "image" # type: ignore
|
type: Literal["image"] = "image" # type: ignore
|
||||||
src: str
|
src: str | Path
|
||||||
width: int = 40
|
width: int = 40
|
||||||
height: int = 40
|
height: int = 40
|
||||||
shape: Literal["square", "circle"] = "square"
|
shape: Literal["square", "circle"] = "square"
|
||||||
alt: str = "image"
|
alt: str = "image"
|
||||||
|
|
||||||
|
@field_validator("src", mode="before")
|
||||||
|
def validate_src(cls, v: str) -> str:
|
||||||
|
if isinstance(v, Path):
|
||||||
|
v = v.resolve().as_uri()
|
||||||
|
return v
|
||||||
|
|
||||||
|
|
||||||
class StatusBadgeCell(BaseCell):
|
class StatusBadgeCell(BaseCell):
|
||||||
"""状态徽章单元格"""
|
"""状态徽章单元格"""
|
||||||
@ -62,9 +70,12 @@ class RichTextCell(BaseCell):
|
|||||||
"""富文本单元格,支持多个带样式的文本片段"""
|
"""富文本单元格,支持多个带样式的文本片段"""
|
||||||
|
|
||||||
type: Literal["rich_text"] = "rich_text" # type: ignore
|
type: Literal["rich_text"] = "rich_text" # type: ignore
|
||||||
spans: list[TextSpan] = Field(default_factory=list, description="文本片段列表")
|
spans: list[TextSpan] = []
|
||||||
direction: Literal["column", "row"] = Field("column", description="片段排列方向")
|
"""文本片段列表"""
|
||||||
gap: str = Field("4px", description="片段之间的间距")
|
direction: Literal["column", "row"] = "column"
|
||||||
|
"""片段排列方向"""
|
||||||
|
gap: str = "4px"
|
||||||
|
"""片段之间的间距"""
|
||||||
|
|
||||||
|
|
||||||
TableCell = (
|
TableCell = (
|
||||||
@ -84,16 +95,18 @@ class TableData(RenderableComponent):
|
|||||||
"""通用表格的数据模型"""
|
"""通用表格的数据模型"""
|
||||||
|
|
||||||
style_name: str | None = None
|
style_name: str | None = None
|
||||||
title: str = Field(..., description="表格主标题")
|
title: str
|
||||||
tip: str | None = Field(None, description="表格下方的提示信息")
|
"""表格主标题"""
|
||||||
headers: list[str] = Field(default_factory=list, description="表头列表")
|
tip: str | None = None
|
||||||
rows: list[list[TableCell]] = Field(default_factory=list, description="数据行列表")
|
"""表格下方的提示信息"""
|
||||||
column_alignments: list[Literal["left", "center", "right"]] | None = Field(
|
headers: list[str] = [] # noqa: RUF012
|
||||||
default=None, description="每列的对齐方式"
|
"""表头列表"""
|
||||||
)
|
rows: list[list[TableCell]] = [] # noqa: RUF012
|
||||||
column_widths: list[str | int] | None = Field(
|
"""数据行列表"""
|
||||||
default=None, description="每列的宽度 (e.g., ['50px', 'auto', 100])"
|
column_alignments: list[Literal["left", "center", "right"]] | None = None
|
||||||
)
|
"""每列的对齐方式"""
|
||||||
|
column_widths: list[str | int] | None = None
|
||||||
|
"""每列的宽度 (e.g., ['50px', 'auto', 100])"""
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def template_name(self) -> str:
|
def template_name(self) -> str:
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user