mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 20:34:48 +08:00
fc0eed08cc
* fix: retain plugin lifecycle ownership and recover installation state * fix: preserve directed messaging and channel interaction contracts * test: cover plugin rollback and uninstall lifecycle boundaries --------- Co-authored-by: jxxghp <jxxghp@gmail.com>
728 lines
26 KiB
Python
728 lines
26 KiB
Python
"""
|
||
系统升级、重启、备份恢复及通知渠道管理入口。
|
||
"""
|
||
|
||
import asyncio
|
||
import errno
|
||
import json
|
||
import re
|
||
import shutil
|
||
import threading
|
||
import uuid
|
||
from collections.abc import Callable, Coroutine, Iterable, Mapping
|
||
from pathlib import Path
|
||
from typing import TYPE_CHECKING, Any, Optional, Protocol, Union
|
||
|
||
from app.application.configuration import get_chain_runtime_config_snapshot
|
||
from app.chain.base import ChainBase
|
||
from app.runtime import version as runtime_version
|
||
from app.runtime.log import logger
|
||
from app.runtime.loop import main_loop_registry
|
||
from app.runtime.state import SystemHelper
|
||
from app.runtime.tasks import get_task_registry
|
||
from app.schemas.message import Message
|
||
from app.schemas.types import NotificationChannel
|
||
|
||
if TYPE_CHECKING:
|
||
from app.application.messaging.update import SystemUpdateInteractionHandler
|
||
|
||
|
||
class SystemResponsePort(Protocol):
|
||
"""
|
||
系统链查询发布版本所需的最小同步 HTTP 响应契约。
|
||
"""
|
||
|
||
def json(self) -> Any:
|
||
"""
|
||
返回响应 JSON 载荷。
|
||
|
||
:return: 解析后的 JSON 值，无效内容抛出解析异常
|
||
"""
|
||
...
|
||
|
||
def close(self) -> None:
|
||
"""
|
||
释放响应与连接资源。
|
||
"""
|
||
...
|
||
|
||
|
||
class SystemHttpPort(Protocol):
|
||
"""
|
||
系统链查询发布版本所需的同步 GET 端口。
|
||
"""
|
||
|
||
def get(
|
||
self,
|
||
url: str,
|
||
*,
|
||
proxies: Optional[dict[str, str]],
|
||
headers: Mapping[str, str],
|
||
) -> Optional[SystemResponsePort]:
|
||
"""
|
||
读取远端发布列表，并保留无响应与有响应两态。
|
||
|
||
:param url: 远程发布版本接口的完整地址
|
||
:param proxies: 协议到代理地址的映射，None 表示未指定代理
|
||
:param headers: 版本接口请求头，包含鉴权或内容协商信息
|
||
:return: 发布接口响应，未取得响应时为 None；调用方解析后负责关闭
|
||
"""
|
||
...
|
||
|
||
|
||
class SystemEnvironmentPort(Protocol):
|
||
"""
|
||
系统链备份插件时所需的最小运行环境端口。
|
||
"""
|
||
|
||
def is_docker(self) -> bool:
|
||
"""
|
||
返回当前进程是否运行在 Docker 容器中。
|
||
|
||
:return: 当前进程位于 Docker 容器中时为 True，否则为 False
|
||
"""
|
||
...
|
||
|
||
|
||
_system_port_lock = threading.RLock()
|
||
_system_http_port: Optional[SystemHttpPort] = None
|
||
_system_environment_port: Optional[SystemEnvironmentPort] = None
|
||
|
||
|
||
def configure_system_ports(
|
||
*,
|
||
http: SystemHttpPort,
|
||
environment: SystemEnvironmentPort,
|
||
) -> tuple[Optional[SystemHttpPort], Optional[SystemEnvironmentPort]]:
|
||
"""
|
||
由启动组合根装配系统链技术端口，并返回旧快照。
|
||
"""
|
||
global _system_http_port, _system_environment_port
|
||
with _system_port_lock:
|
||
previous = (_system_http_port, _system_environment_port)
|
||
_system_http_port = http
|
||
_system_environment_port = environment
|
||
return previous
|
||
|
||
|
||
def reset_system_ports(
|
||
http: Optional[SystemHttpPort] = None,
|
||
environment: Optional[SystemEnvironmentPort] = None,
|
||
) -> None:
|
||
"""
|
||
恢复指定系统链端口；省略参数时回到未装配状态。
|
||
"""
|
||
global _system_http_port, _system_environment_port
|
||
with _system_port_lock:
|
||
_system_http_port = http
|
||
_system_environment_port = environment
|
||
|
||
|
||
def _system_ports_snapshot() -> tuple[SystemHttpPort, SystemEnvironmentPort]:
|
||
"""
|
||
读取一致的系统链端口快照，未装配时稳定失败。
|
||
"""
|
||
with _system_port_lock:
|
||
http = _system_http_port
|
||
environment = _system_environment_port
|
||
if http is None or environment is None:
|
||
raise RuntimeError("系统链技术端口尚未由启动组合根装配")
|
||
return http, environment
|
||
|
||
|
||
def _close_system_response(response: SystemResponsePort) -> None:
|
||
"""
|
||
释放版本响应；关闭失败只记录诊断，不覆盖解析结果。
|
||
"""
|
||
try:
|
||
response.close()
|
||
except Exception as err:
|
||
logger.debug(f"释放版本响应失败：{str(err)}")
|
||
|
||
|
||
async def _run_update_operation(
|
||
function: Callable[..., Any],
|
||
/,
|
||
*args: Any,
|
||
**kwargs: Any,
|
||
) -> Any:
|
||
"""
|
||
通过任务登记器在线程池执行同步更新操作并等待真实终态。
|
||
"""
|
||
task = get_task_registry().create_sync(
|
||
function,
|
||
*args,
|
||
owner="chain.system.update_progress.operation",
|
||
**kwargs,
|
||
)
|
||
return await asyncio.shield(task)
|
||
|
||
|
||
class _SystemUpdateChain(ChainBase):
|
||
"""
|
||
提供通知渠道主程序升级交互的 Chain 入口。
|
||
"""
|
||
|
||
_update_restart_file = "__system_update_restart__"
|
||
|
||
def _update_interaction_handler(self) -> "SystemUpdateInteractionHandler":
|
||
"""
|
||
构造复用当前消息网关和系统应用服务的更新交互控制器。
|
||
"""
|
||
from app.application.messaging.update import SystemUpdateInteractionHandler
|
||
|
||
if self.system_service is None:
|
||
raise RuntimeError("系统更新服务尚未由启动组合根装配")
|
||
return SystemUpdateInteractionHandler(
|
||
messenger=self,
|
||
actions=self.system_service,
|
||
submit_monitor=self._submit_update_monitor,
|
||
run_sync=_run_update_operation,
|
||
mark_restart=self._mark_update_restart,
|
||
clear_restart_marker=self._clear_update_restart_marker,
|
||
)
|
||
|
||
@staticmethod
|
||
def _submit_update_monitor(monitor: Coroutine[Any, Any, None]) -> None:
|
||
"""
|
||
把进度监视协程跨线程登记到宿主事件循环。
|
||
"""
|
||
get_task_registry().submit_threadsafe(
|
||
monitor,
|
||
loop=main_loop_registry.require(),
|
||
owner="chain.system.update_progress",
|
||
)
|
||
|
||
def _mark_update_restart(
|
||
self,
|
||
channel: NotificationChannel,
|
||
userid: Union[int, str],
|
||
source: Optional[str],
|
||
) -> None:
|
||
"""
|
||
记录升级重启的回复目标，供服务恢复后发送完成通知。
|
||
"""
|
||
self.save_cache(
|
||
{"channel": channel.value, "userid": userid, "source": source},
|
||
self._update_restart_file,
|
||
)
|
||
|
||
def _clear_update_restart_marker(self) -> None:
|
||
"""
|
||
升级安装未能进入重启阶段时删除完成通知标记。
|
||
"""
|
||
self.remove_cache(self._update_restart_file)
|
||
|
||
def remote_update(
|
||
self,
|
||
arg_str: str = "",
|
||
channel: Optional[NotificationChannel] = None,
|
||
userid: Optional[Union[int, str]] = None,
|
||
source: Optional[str] = None,
|
||
) -> None:
|
||
"""
|
||
检查正式版本并启动通知渠道中的升级确认交互。
|
||
"""
|
||
self._update_interaction_handler().remote_update(
|
||
arg_str=arg_str,
|
||
channel=channel,
|
||
userid=userid,
|
||
source=source,
|
||
)
|
||
|
||
def handle_update_callback_interaction(
|
||
self,
|
||
callback_data: str,
|
||
channel: NotificationChannel,
|
||
source: Optional[str],
|
||
userid: Union[int, str],
|
||
username: Optional[str],
|
||
original_message_id: Optional[Union[str, int]] = None,
|
||
original_chat_id: Optional[str] = None,
|
||
) -> bool:
|
||
"""
|
||
处理更新交互按钮并把原消息定位参数交给进度编辑器。
|
||
"""
|
||
return self._update_interaction_handler().handle_callback_interaction(
|
||
callback_data=callback_data,
|
||
channel=channel,
|
||
source=source,
|
||
userid=userid,
|
||
username=username,
|
||
original_message_id=original_message_id,
|
||
original_chat_id=original_chat_id,
|
||
)
|
||
|
||
def handle_update_text_interaction(
|
||
self,
|
||
channel: NotificationChannel,
|
||
source: Optional[str],
|
||
userid: Union[int, str],
|
||
username: Optional[str],
|
||
text: str,
|
||
) -> bool:
|
||
"""
|
||
处理不支持按钮渠道中的升级或重启确认文本。
|
||
"""
|
||
return self._update_interaction_handler().handle_text_interaction(
|
||
channel=channel,
|
||
source=source,
|
||
userid=userid,
|
||
username=username,
|
||
text=text,
|
||
)
|
||
|
||
|
||
class SystemChain(_SystemUpdateChain):
|
||
"""
|
||
系统级处理链。
|
||
"""
|
||
|
||
_restart_file = "__system_restart__"
|
||
_plugin_restore_pending_file = "__plugin_restore_pending__"
|
||
|
||
def remote_clear_cache(self, channel: NotificationChannel, userid: Union[int, str], source: Optional[str] = None) -> None:
|
||
"""
|
||
清理系统缓存
|
||
"""
|
||
self.clear_cache()
|
||
self.post_message(Message(
|
||
channel=channel,
|
||
source=source,
|
||
title="缓存清理完成！",
|
||
userid=userid,
|
||
save_history=False))
|
||
|
||
def restart(self, channel: NotificationChannel, userid: Union[int, str], source: Optional[str] = None) -> None:
|
||
"""
|
||
重启系统
|
||
"""
|
||
if channel and userid:
|
||
self.post_message(Message(
|
||
channel=channel,
|
||
source=source,
|
||
title="系统正在重启，请耐心等候！",
|
||
userid=userid,
|
||
save_history=False))
|
||
self.save_cache({
|
||
"channel": channel.value,
|
||
"userid": userid,
|
||
"source": source,
|
||
}, self._restart_file)
|
||
self.backup_plugins()
|
||
SystemHelper.restart()
|
||
|
||
@staticmethod
|
||
def backup_plugins() -> None:
|
||
"""
|
||
备份插件到用户配置目录（仅docker环境）
|
||
"""
|
||
|
||
# 非docker环境不处理
|
||
_, environment = _system_ports_snapshot()
|
||
if not environment.is_docker():
|
||
return
|
||
|
||
try:
|
||
config = get_chain_runtime_config_snapshot()
|
||
plugins_dir = config.root_path / "app" / "plugins"
|
||
backup_dir = config.config_path / "plugins_backup"
|
||
|
||
if not plugins_dir.exists():
|
||
logger.info("插件目录不存在，跳过备份")
|
||
return
|
||
|
||
backup_dir.mkdir(parents=True, exist_ok=True)
|
||
pending_file = backup_dir / SystemChain._plugin_restore_pending_file
|
||
pending_items = (
|
||
SystemChain.__read_plugin_restore_pending(pending_file)
|
||
if pending_file.exists()
|
||
else None
|
||
)
|
||
exclude_items = {"__init__.py", "__pycache__", ".DS_Store"}
|
||
|
||
backup_failed = False
|
||
|
||
for item in plugins_dir.iterdir():
|
||
if item.name in exclude_items:
|
||
continue
|
||
# 失败项目的原快照是下一次恢复的唯一材料，关停备份不能覆盖它。
|
||
if pending_file.exists() and (
|
||
pending_items is None or item.name in pending_items
|
||
):
|
||
logger.debug(f"插件 {item.name} 有待重试恢复标记，保留原快照")
|
||
continue
|
||
target_path = backup_dir / item.name
|
||
|
||
try:
|
||
SystemChain.__replace_snapshot(
|
||
item,
|
||
target_path,
|
||
ignore=shutil.ignore_patterns(
|
||
"__pycache__", "*.pyc", ".DS_Store"
|
||
) if item.is_dir() else None,
|
||
)
|
||
logger.debug(f"已备份插件项目: {item.name}")
|
||
except Exception as e:
|
||
backup_failed = True
|
||
logger.error(f"备份插件 {item.name} 失败: {e}")
|
||
|
||
if backup_failed:
|
||
logger.warning(f"插件备份部分失败，保留可用旧快照: {backup_dir}")
|
||
else:
|
||
logger.info(f"插件备份完成，备份位置: {backup_dir}")
|
||
|
||
except Exception as e:
|
||
logger.error(f"插件备份失败: {str(e)}")
|
||
|
||
@staticmethod
|
||
def restore_plugins() -> None:
|
||
"""
|
||
从备份恢复插件到app/plugins目录，恢复完成后删除备份（仅docker环境）
|
||
"""
|
||
|
||
# 非docker环境不处理
|
||
_, environment = _system_ports_snapshot()
|
||
if not environment.is_docker():
|
||
return
|
||
|
||
config = get_chain_runtime_config_snapshot()
|
||
plugins_dir = config.root_path / "app" / "plugins"
|
||
backup_dir = config.config_path / "plugins_backup"
|
||
|
||
if not backup_dir.exists():
|
||
logger.info("插件备份目录不存在，跳过恢复")
|
||
return
|
||
|
||
pending_file = backup_dir / SystemChain._plugin_restore_pending_file
|
||
|
||
# 系统重置或上次恢复未完成时才消费备份。
|
||
system_reset = SystemHelper().is_system_reset()
|
||
should_restore = system_reset or pending_file.exists()
|
||
if not should_restore:
|
||
logger.info("当前不是系统重置，保留插件备份供后续重置使用")
|
||
return
|
||
|
||
plugins_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
restored_count = 0
|
||
restore_failed = False
|
||
failed_items: dict[str, bool] = {}
|
||
pending_items = (
|
||
SystemChain.__read_plugin_restore_pending(pending_file)
|
||
if pending_file.exists() and not system_reset
|
||
else None
|
||
)
|
||
for item in backup_dir.iterdir():
|
||
if (
|
||
item.name == SystemChain._plugin_restore_pending_file
|
||
or SystemChain.__is_snapshot_artifact(item.name)
|
||
):
|
||
continue
|
||
target_path = plugins_dir / item.name
|
||
if pending_items is not None:
|
||
if item.name not in pending_items:
|
||
continue
|
||
if not pending_items[item.name] and target_path.exists():
|
||
logger.info(f"插件 {item.name} 已在恢复失败后重新安装，跳过备份覆盖")
|
||
continue
|
||
target_existed = target_path.exists()
|
||
try:
|
||
if item.is_dir() or item.is_file():
|
||
SystemChain.__replace_snapshot(item, target_path)
|
||
logger.debug(f"已恢复插件文件: {item.name}")
|
||
restored_count += 1
|
||
except Exception as e:
|
||
restore_failed = True
|
||
failed_items[item.name] = target_existed
|
||
logger.error(f"恢复插件 {item.name} 时发生错误: {str(e)}")
|
||
continue
|
||
|
||
logger.info(f"插件恢复完成，共恢复 {restored_count} 个项目")
|
||
|
||
if restore_failed:
|
||
if SystemChain.__write_plugin_restore_pending(pending_file, failed_items):
|
||
logger.warning("插件恢复未完成，保留备份并标记为下次启动重试")
|
||
else:
|
||
logger.warning("插件恢复未完成，已保留备份，但无法写入下次启动重试标记")
|
||
return
|
||
|
||
# 源码恢复完成后即可消费备份；依赖由启动后的统一后台任务处理。
|
||
try:
|
||
shutil.rmtree(backup_dir)
|
||
logger.info(f"已删除插件备份目录: {backup_dir}")
|
||
except Exception as e:
|
||
logger.warning(f"删除备份目录失败: {str(e)}")
|
||
if backup_dir.exists():
|
||
SystemChain.__write_plugin_restore_pending(pending_file, {})
|
||
|
||
@staticmethod
|
||
def __is_snapshot_artifact(name: str) -> bool:
|
||
"""
|
||
识别快照替换过程中生成的临时或旧快照条目。
|
||
"""
|
||
return bool(re.fullmatch(r"\..+\.(?:tmp|old)-[0-9a-f]{32}", name))
|
||
|
||
@staticmethod
|
||
def __read_plugin_restore_pending(pending_file: Path) -> Optional[dict[str, bool]]:
|
||
"""
|
||
读取仍需恢复的插件项目；无效内容按全部项目重试。
|
||
"""
|
||
try:
|
||
payload = json.loads(pending_file.read_text(encoding="utf-8"))
|
||
failed_items = payload.get("failed_items")
|
||
if not isinstance(failed_items, dict):
|
||
return None
|
||
return {
|
||
str(name): target_existed
|
||
for name, target_existed in failed_items.items()
|
||
if isinstance(name, str) and isinstance(target_existed, bool)
|
||
}
|
||
except (OSError, ValueError, TypeError):
|
||
return None
|
||
|
||
@staticmethod
|
||
def __write_plugin_restore_pending(
|
||
pending_file: Path,
|
||
failed_items: dict[str, bool],
|
||
) -> bool:
|
||
"""
|
||
记录失败项目及其原目标状态，供普通重启继续未完成恢复。
|
||
"""
|
||
try:
|
||
pending_file.write_text(
|
||
json.dumps({"failed_items": failed_items}, ensure_ascii=False),
|
||
encoding="utf-8",
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"写入插件恢复重试标记失败: {e}")
|
||
return False
|
||
|
||
@staticmethod
|
||
def __replace_snapshot(
|
||
source: Path, target: Path, *,
|
||
ignore: Callable[[str, list[str]], Iterable[str]] | None = None,
|
||
) -> None:
|
||
"""
|
||
复制到同级临时路径后替换目标，避免失败时丢失旧快照。
|
||
"""
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
suffix = uuid.uuid4().hex
|
||
staging = target.with_name(f".{target.name}.tmp-{suffix}")
|
||
previous = target.with_name(f".{target.name}.old-{suffix}")
|
||
previous_available = False
|
||
published = False
|
||
try:
|
||
if source.is_dir():
|
||
shutil.copytree(source, staging, ignore=ignore)
|
||
else:
|
||
shutil.copy2(source, staging)
|
||
if target.exists():
|
||
try:
|
||
target.replace(previous)
|
||
except OSError as error:
|
||
if error.errno != errno.EXDEV:
|
||
raise
|
||
# overlayfs 可能拒绝把镜像层目录直接 rename 到可写层，
|
||
# 先复制旧目标保留恢复材料，再删除旧目录继续发布快照。
|
||
if target.is_dir():
|
||
shutil.copytree(target, previous, symlinks=True)
|
||
else:
|
||
shutil.copy2(target, previous, follow_symlinks=False)
|
||
previous_available = True
|
||
SystemChain.__remove_snapshot_path(target)
|
||
else:
|
||
previous_available = True
|
||
staging.replace(target)
|
||
published = True
|
||
except Exception:
|
||
if previous_available and not published:
|
||
try:
|
||
SystemChain.__remove_snapshot_path(target)
|
||
previous.replace(target)
|
||
previous_available = False
|
||
except Exception as rollback_error:
|
||
logger.error(
|
||
f"恢复旧快照失败，已保留恢复材料 {previous}: "
|
||
f"{rollback_error}"
|
||
)
|
||
raise
|
||
finally:
|
||
if staging.is_dir():
|
||
shutil.rmtree(staging, ignore_errors=True)
|
||
elif staging.exists():
|
||
staging.unlink(missing_ok=True)
|
||
if published and previous.exists():
|
||
if previous.is_dir():
|
||
shutil.rmtree(previous, ignore_errors=True)
|
||
else:
|
||
previous.unlink(missing_ok=True)
|
||
|
||
@staticmethod
|
||
def __remove_snapshot_path(path: Path) -> None:
|
||
"""
|
||
删除待替换目标，保留失败回滚所需的旧快照副本。
|
||
"""
|
||
if path.is_dir() and not path.is_symlink():
|
||
shutil.rmtree(path)
|
||
elif path.exists() or path.is_symlink():
|
||
path.unlink()
|
||
|
||
def __get_version_message(self) -> str:
|
||
"""
|
||
获取版本信息文本
|
||
"""
|
||
server_release_version = self.__get_server_release_version()
|
||
front_release_version = self.__get_front_release_version()
|
||
server_local_version = runtime_version.get_app_version()
|
||
front_local_version = runtime_version.get_frontend_version()
|
||
if server_release_version == server_local_version:
|
||
title = f"当前后端版本：{server_local_version}，已是最新版本\n"
|
||
else:
|
||
title = f"当前后端版本：{server_local_version}，远程版本：{server_release_version}\n"
|
||
if front_release_version == front_local_version:
|
||
title += f"当前前端版本：{front_local_version}，已是最新版本"
|
||
else:
|
||
title += f"当前前端版本：{front_local_version}，远程版本：{front_release_version}"
|
||
return title
|
||
|
||
def version(self, channel: NotificationChannel, userid: Union[int, str], source: Optional[str] = None) -> None:
|
||
"""
|
||
查看当前版本、远程版本
|
||
"""
|
||
self.post_message(Message(
|
||
channel=channel,
|
||
source=source,
|
||
title=self.__get_version_message(),
|
||
userid=userid,
|
||
save_history=False))
|
||
|
||
def restart_finish(self) -> None:
|
||
"""
|
||
如通过交互命令重启，
|
||
重启完发送msg
|
||
"""
|
||
restart_channel = self.load_cache(self._restart_file)
|
||
if restart_channel:
|
||
if not isinstance(restart_channel, dict):
|
||
restart_channel = json.loads(restart_channel)
|
||
channel = next(
|
||
(channel for channel in NotificationChannel.__members__.values() if
|
||
channel.value == restart_channel.get('channel')), None)
|
||
userid = restart_channel.get('userid')
|
||
|
||
title = self.__get_version_message()
|
||
self.post_message(Message(
|
||
channel=channel,
|
||
source=restart_channel.get("source"),
|
||
title=f"系统已重启完成！\n{title}",
|
||
userid=userid,
|
||
save_history=False))
|
||
self.remove_cache(self._restart_file)
|
||
|
||
update_restart_channel = self.load_cache(self._update_restart_file)
|
||
if update_restart_channel:
|
||
if not isinstance(update_restart_channel, dict):
|
||
update_restart_channel = json.loads(update_restart_channel)
|
||
channel = next(
|
||
(
|
||
candidate
|
||
for candidate in NotificationChannel.__members__.values()
|
||
if candidate.value == update_restart_channel.get("channel")
|
||
),
|
||
None,
|
||
)
|
||
userid = update_restart_channel.get("userid")
|
||
source = update_restart_channel.get("source")
|
||
self.post_message(
|
||
Message(
|
||
channel=channel,
|
||
source=source,
|
||
title=(
|
||
"MoviePilot 更新安装完成！\n"
|
||
f"当前后端版本：{runtime_version.get_app_version()}\n"
|
||
f"当前前端版本：{runtime_version.get_frontend_version()}"
|
||
),
|
||
userid=userid,
|
||
save_history=False,
|
||
)
|
||
)
|
||
self.remove_cache(self._update_restart_file)
|
||
|
||
@staticmethod
|
||
def __get_server_release_version() -> Optional[str]:
|
||
"""
|
||
获取后端V3最新版本
|
||
"""
|
||
try:
|
||
http, _ = _system_ports_snapshot()
|
||
response = http.get(
|
||
"https://api.github.com/repos/jxxghp/MoviePilot/releases",
|
||
proxies=get_chain_runtime_config_snapshot().proxy,
|
||
headers=get_chain_runtime_config_snapshot().github_headers,
|
||
)
|
||
if response:
|
||
try:
|
||
releases: list[str] = [release['tag_name'] for release in response.json()]
|
||
v3_releases = [tag for tag in releases if re.match(r"^v3\.", tag)]
|
||
if not v3_releases:
|
||
logger.warning("获取v3后端最新版本出错！")
|
||
else:
|
||
# 找到最新的v3版本
|
||
latest_v3 = sorted(v3_releases, key=lambda s: list(map(int, re.findall(r'\d+', s))))[-1]
|
||
logger.info(f"获取到后端最新版本：{latest_v3}")
|
||
return latest_v3
|
||
finally:
|
||
_close_system_response(response)
|
||
else:
|
||
logger.error("无法获取后端版本信息，请检查网络连接或GitHub API请求。")
|
||
except Exception as err:
|
||
logger.error(f"获取后端最新版本失败：{str(err)}")
|
||
return None
|
||
|
||
@staticmethod
|
||
def __get_front_release_version() -> Optional[str]:
|
||
"""
|
||
获取前端V3最新版本
|
||
"""
|
||
try:
|
||
http, _ = _system_ports_snapshot()
|
||
response = http.get(
|
||
"https://api.github.com/repos/jxxghp/MoviePilot-Frontend/releases",
|
||
proxies=get_chain_runtime_config_snapshot().proxy,
|
||
headers=get_chain_runtime_config_snapshot().github_headers,
|
||
)
|
||
if response:
|
||
try:
|
||
releases: list[str] = [release['tag_name'] for release in response.json()]
|
||
v3_releases = [tag for tag in releases if re.match(r"^v3\.", tag)]
|
||
if not v3_releases:
|
||
logger.warning("获取v3前端最新版本出错！")
|
||
else:
|
||
# 找到最新的v3版本
|
||
latest_v3 = sorted(v3_releases, key=lambda s: list(map(int, re.findall(r'\d+', s))))[-1]
|
||
logger.info(f"获取到前端最新版本：{latest_v3}")
|
||
return latest_v3
|
||
finally:
|
||
_close_system_response(response)
|
||
else:
|
||
logger.error("无法获取前端版本信息，请检查网络连接或GitHub API请求。")
|
||
except Exception as err:
|
||
logger.error(f"获取前端最新版本失败：{str(err)}")
|
||
return None
|
||
|
||
@staticmethod
|
||
def get_server_local_version() -> str:
|
||
"""
|
||
返回当前后端构建版本。
|
||
"""
|
||
return runtime_version.get_app_version()
|
||
|
||
@staticmethod
|
||
def get_frontend_version() -> str | None:
|
||
"""
|
||
返回当前部署的前端资源版本。
|
||
"""
|
||
return runtime_version.get_frontend_version()
|