From 34a8de56c39854011e0134f28c1a0cd1aecb5517 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=99=B4=E7=8C=AB?= Date: Sun, 15 Mar 2026 10:06:59 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E9=9D=99=E6=80=81?= =?UTF-8?q?=E8=B7=AF=E5=BE=84=E5=87=86=E5=A4=87=E5=92=8C=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E6=9E=84=E5=BB=BA=E5=8A=9F=E8=83=BD=E7=9A=84=E5=8D=95=E5=85=83?= =?UTF-8?q?=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pytests/webui/test_app.py | 122 ++++++++++++++++ src/webui/app.py | 284 ++++++++++++++++++++++++++++++++++---- 2 files changed, 379 insertions(+), 27 deletions(-) create mode 100644 pytests/webui/test_app.py diff --git a/pytests/webui/test_app.py b/pytests/webui/test_app.py new file mode 100644 index 00000000..eb37305e --- /dev/null +++ b/pytests/webui/test_app.py @@ -0,0 +1,122 @@ +from unittest.mock import patch + +from src.webui import app as webui_app + + +def test_ensure_static_path_ready_uses_existing_static_path(tmp_path) -> None: + static_path = tmp_path / "dist" + static_path.mkdir() + (static_path / "index.html").write_text("", encoding="utf-8") + + with ( + patch.object(webui_app, "_resolve_static_path", return_value=static_path), + patch.object(webui_app, "_try_build_dashboard") as build_mock, + ): + result = webui_app._ensure_static_path_ready() + + assert result == static_path + build_mock.assert_not_called() + + +def test_ensure_static_path_ready_retries_after_auto_build(tmp_path) -> None: + static_path = tmp_path / "dist" + static_path.mkdir() + (static_path / "index.html").write_text("", encoding="utf-8") + + with ( + patch.object(webui_app, "_resolve_static_path", side_effect=[None, static_path]), + patch.object( + webui_app, + "_try_build_dashboard", + return_value=webui_app.DashboardAutoRecoveryResult(succeeded=True), + ) as build_mock, + ): + result = webui_app._ensure_static_path_ready() + + assert result == static_path + build_mock.assert_called_once_with() + + +def test_ensure_static_path_ready_logs_manual_hint_when_auto_build_fails() -> None: + with ( + patch.object(webui_app, "_resolve_static_path", return_value=None), + patch.object( + webui_app, + "_try_build_dashboard", + return_value=webui_app.DashboardAutoRecoveryResult( + succeeded=False, + manual_recovery_command=webui_app._MANUAL_BUILD_COMMAND, + ), + ), + patch.object(webui_app.logger, "warning") as warning_mock, + ): + result = webui_app._ensure_static_path_ready() + + assert result is None + warning_mock.assert_any_call(webui_app.t("startup.webui_auto_recovery_failed")) + warning_mock.assert_any_call( + webui_app.t("startup.webui_manual_build_hint", command=webui_app._MANUAL_BUILD_COMMAND) + ) + + +def test_setup_static_files_does_not_duplicate_warning_when_static_path_is_unavailable() -> None: + app = webui_app.FastAPI() + + with ( + patch.object(webui_app, "_ensure_static_path_ready", return_value=None), + patch.object(webui_app.logger, "warning") as warning_mock, + ): + webui_app._setup_static_files(app) + + warning_mock.assert_not_called() + + +def test_get_dashboard_build_command_defaults_to_npm(tmp_path) -> None: + (tmp_path / "package.json").write_text("{}", encoding="utf-8") + + with patch.object( + webui_app.shutil, + "which", + side_effect=lambda tool_name: "/usr/bin/npm" if tool_name == "npm" else None, + ): + command = webui_app._get_dashboard_build_command(tmp_path) + + assert command == ["npm", "run", "build"] + + +def test_try_build_dashboard_installs_missing_dependencies_before_build(monkeypatch, tmp_path) -> None: + (tmp_path / "package.json").write_text("{}", encoding="utf-8") + run_results = [ + webui_app.CompletedProcess(args=["npm", "install", "--no-package-lock"], returncode=0, stdout="", stderr=""), + webui_app.CompletedProcess(args=["npm", "run", "build"], returncode=0, stdout="", stderr=""), + ] + + monkeypatch.setattr(webui_app, "_get_dashboard_root", lambda: tmp_path) + monkeypatch.setattr(webui_app, "_should_auto_install_dashboard_dependencies", lambda dashboard_root: True) + + with ( + patch.object(webui_app, "_get_dashboard_build_command", return_value=["npm", "run", "build"]), + patch.object(webui_app, "run", side_effect=run_results) as run_mock, + ): + result = webui_app._try_build_dashboard() + + assert result.succeeded is True + assert run_mock.call_count == 2 + install_call = run_mock.call_args_list[0] + build_call = run_mock.call_args_list[1] + assert install_call.args[0] == ["npm", "install", "--no-package-lock"] + assert install_call.kwargs["cwd"] == tmp_path + assert build_call.args[0] == ["npm", "run", "build"] + assert build_call.kwargs["cwd"] == tmp_path + + +def test_resolve_static_path_uses_dashboard_dist(monkeypatch, tmp_path) -> None: + dashboard_dist = tmp_path / "dashboard" / "dist" + dashboard_dist.mkdir(parents=True) + + monkeypatch.setattr(webui_app, "_get_project_root", lambda: tmp_path) + + with patch.object(webui_app, "import_module", side_effect=ImportError): + resolved_path = webui_app._resolve_static_path() + + assert resolved_path == dashboard_dist diff --git a/src/webui/app.py b/src/webui/app.py index 30ed1199..934694c6 100644 --- a/src/webui/app.py +++ b/src/webui/app.py @@ -1,15 +1,44 @@ """FastAPI 应用工厂 - 创建和配置 WebUI 应用实例""" +from dataclasses import dataclass from importlib import import_module from pathlib import Path +from subprocess import CompletedProcess, TimeoutExpired, run import mimetypes +import shutil + from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse + +from src.common.i18n import t from src.common.logger import get_logger logger = get_logger("webui.app") +_DASHBOARD_BUILD_TIMEOUT_SECONDS = 300 +_DASHBOARD_INSTALL_TIMEOUT_SECONDS = 600 +_MANUAL_BUILD_COMMAND = "cd dashboard && npm install && npm run build" + +_DASHBOARD_BUILD_COMMANDS = { + "bun": ["bun", "run", "build"], + "npm": ["npm", "run", "build"], + "pnpm": ["pnpm", "build"], + "yarn": ["yarn", "build"], +} +_DASHBOARD_INSTALL_COMMANDS = { + "bun": ["bun", "install"], + "npm": ["npm", "install", "--no-package-lock"], + "pnpm": ["pnpm", "install"], + "yarn": ["yarn", "install"], +} + + +@dataclass(frozen=True) +class DashboardAutoRecoveryResult: + succeeded: bool + manual_recovery_command: str | None = None + def _resolve_safe_static_file_path(static_path: Path, full_path: str) -> Path | None: static_root = static_path.resolve() @@ -18,12 +47,215 @@ def _resolve_safe_static_file_path(static_path: Path, full_path: str) -> Path | candidate_path = (static_root / full_path).resolve() candidate_path.relative_to(static_root) except (OSError, RuntimeError, ValueError): - logger.warning(f"🚫 检测到疑似路径穿越请求: {full_path}") + logger.warning(t("startup.webui_path_traversal_detected", full_path=full_path)) return None return candidate_path +def _get_project_root() -> Path: + return Path(__file__).resolve().parents[2] + + +def _get_dashboard_root() -> Path: + return _get_project_root() / "dashboard" + + +def _format_dashboard_shell_commands(*commands: list[str]) -> str: + formatted_commands = " && ".join(" ".join(command) for command in commands) + return f"cd dashboard && {formatted_commands}" + + +def _validate_static_path(static_path: Path | None) -> tuple[str, dict[str, object]] | None: + if static_path is None: + return "startup.webui_static_dir_missing", {} + + if not static_path.exists(): + return "startup.webui_static_dir_missing_with_path", {"static_path": static_path} + + index_path = static_path / "index.html" + if not index_path.exists(): + return "startup.webui_index_missing", {"index_path": index_path} + + return None + + +def _summarize_command_output(command_result: CompletedProcess[str] | TimeoutExpired) -> str: + output_chunks: list[str] = [] + stdout = command_result.stdout + stderr = command_result.stderr + + if isinstance(stdout, str) and stdout.strip(): + output_chunks.append(stdout.strip()) + if isinstance(stderr, str) and stderr.strip(): + output_chunks.append(stderr.strip()) + + if not output_chunks: + return "" + + combined_output = "\n".join(output_chunks) + max_output_length = 2000 + if len(combined_output) <= max_output_length: + return combined_output + + return combined_output[-max_output_length:] + + +def _get_preferred_dashboard_package_manager(dashboard_root: Path) -> str: + if (dashboard_root / "pnpm-lock.yaml").exists(): + return "pnpm" + if (dashboard_root / "yarn.lock").exists(): + return "yarn" + if (dashboard_root / "bun.lock").exists() or (dashboard_root / "bun.lockb").exists(): + return "bun" + return "npm" + + +def _get_dashboard_build_command(dashboard_root: Path) -> list[str] | None: + if not (dashboard_root / "package.json").exists(): + return None + + preferred_package_manager = _get_preferred_dashboard_package_manager(dashboard_root) + package_managers = [ + preferred_package_manager, + *[package_manager for package_manager in _DASHBOARD_BUILD_COMMANDS if package_manager != preferred_package_manager], + ] + + for package_manager in package_managers: + if shutil.which(package_manager): + return _DASHBOARD_BUILD_COMMANDS[package_manager] + + return None + + +def _get_dashboard_manual_recovery_command(dashboard_root: Path, build_command: list[str] | None = None) -> str: + package_manager = build_command[0] if build_command is not None else _get_preferred_dashboard_package_manager(dashboard_root) + install_command = _DASHBOARD_INSTALL_COMMANDS.get(package_manager) + selected_build_command = _DASHBOARD_BUILD_COMMANDS.get(package_manager) + + if install_command is None or selected_build_command is None: + return _MANUAL_BUILD_COMMAND + + return _format_dashboard_shell_commands(install_command, selected_build_command) + + +def _should_auto_install_dashboard_dependencies(dashboard_root: Path) -> bool: + return not (dashboard_root / "node_modules").exists() + + +def _try_build_dashboard() -> DashboardAutoRecoveryResult: + dashboard_root = _get_dashboard_root() + if not dashboard_root.exists(): + logger.warning(t("startup.webui_dashboard_source_missing", dashboard_root=dashboard_root)) + return DashboardAutoRecoveryResult(succeeded=False) + + build_command = _get_dashboard_build_command(dashboard_root) + if build_command is None: + logger.warning(t("startup.webui_auto_build_tool_missing")) + manual_recovery_command = _get_dashboard_manual_recovery_command(dashboard_root) + return DashboardAutoRecoveryResult(succeeded=False, manual_recovery_command=manual_recovery_command) + + manual_recovery_command = _get_dashboard_manual_recovery_command(dashboard_root, build_command) + + if _should_auto_install_dashboard_dependencies(dashboard_root): + install_command = _DASHBOARD_INSTALL_COMMANDS[build_command[0]] + logger.info(t("startup.webui_auto_install_started", command=" ".join(install_command))) + + try: + install_result = run( + install_command, + capture_output=True, + check=False, + cwd=dashboard_root, + text=True, + timeout=_DASHBOARD_INSTALL_TIMEOUT_SECONDS, + ) + except TimeoutExpired as exc: + logger.warning( + t("startup.webui_auto_install_timeout", timeout_seconds=_DASHBOARD_INSTALL_TIMEOUT_SECONDS), + ) + install_output = _summarize_command_output(exc) + if install_output: + logger.warning(t("startup.webui_auto_install_failed_output", output=install_output)) + return DashboardAutoRecoveryResult(succeeded=False, manual_recovery_command=manual_recovery_command) + except OSError as exc: + logger.warning(t("startup.webui_auto_install_exec_failed", error=exc)) + return DashboardAutoRecoveryResult(succeeded=False, manual_recovery_command=manual_recovery_command) + + if install_result.returncode != 0: + logger.warning(t("startup.webui_auto_install_failed", return_code=install_result.returncode)) + install_output = _summarize_command_output(install_result) + if install_output: + logger.warning(t("startup.webui_auto_install_failed_output", output=install_output)) + return DashboardAutoRecoveryResult(succeeded=False, manual_recovery_command=manual_recovery_command) + + logger.info(t("startup.webui_auto_install_succeeded")) + + logger.info(t("startup.webui_auto_build_started", command=" ".join(build_command))) + + try: + build_result = run( + build_command, + capture_output=True, + check=False, + cwd=dashboard_root, + text=True, + timeout=_DASHBOARD_BUILD_TIMEOUT_SECONDS, + ) + except TimeoutExpired as exc: + logger.warning( + t("startup.webui_auto_build_timeout", timeout_seconds=_DASHBOARD_BUILD_TIMEOUT_SECONDS), + ) + build_output = _summarize_command_output(exc) + if build_output: + logger.warning(t("startup.webui_auto_build_failed_output", output=build_output)) + return DashboardAutoRecoveryResult(succeeded=False, manual_recovery_command=manual_recovery_command) + except OSError as exc: + logger.warning(t("startup.webui_auto_build_exec_failed", error=exc)) + return DashboardAutoRecoveryResult(succeeded=False, manual_recovery_command=manual_recovery_command) + + if build_result.returncode != 0: + logger.warning(t("startup.webui_auto_build_failed", return_code=build_result.returncode)) + build_output = _summarize_command_output(build_result) + if build_output: + logger.warning(t("startup.webui_auto_build_failed_output", output=build_output)) + return DashboardAutoRecoveryResult(succeeded=False, manual_recovery_command=manual_recovery_command) + + logger.info(t("startup.webui_auto_build_succeeded")) + return DashboardAutoRecoveryResult(succeeded=True, manual_recovery_command=manual_recovery_command) + + +def _ensure_static_path_ready() -> Path | None: + static_path = _resolve_static_path() + validation_error = _validate_static_path(static_path) + if validation_error is None: + return static_path + + logger.info(t("startup.webui_static_assets_try_auto_build")) + + auto_recovery_result = _try_build_dashboard() + if auto_recovery_result.succeeded: + static_path = _resolve_static_path() + validation_error = _validate_static_path(static_path) + if validation_error is None: + return static_path + logger.warning(t("startup.webui_auto_build_artifacts_invalid")) + error_key, error_kwargs = validation_error + logger.warning(t(error_key, **error_kwargs)) + logger.warning( + t( + "startup.webui_manual_build_hint", + command=auto_recovery_result.manual_recovery_command or _MANUAL_BUILD_COMMAND, + ) + ) + return None + + if auto_recovery_result.manual_recovery_command is not None: + logger.warning(t("startup.webui_auto_recovery_failed")) + logger.warning(t("startup.webui_manual_build_hint", command=auto_recovery_result.manual_recovery_command)) + return None + + def create_app( host: str = "0.0.0.0", port: int = 8001, @@ -71,7 +303,7 @@ def _setup_cors(app: FastAPI, port: int): ], expose_headers=["Content-Length", "Content-Type"], ) - logger.debug("✅ CORS 中间件已配置") + logger.debug(t("startup.webui_cors_configured")) def _setup_anti_crawler(app: FastAPI): @@ -83,15 +315,15 @@ def _setup_anti_crawler(app: FastAPI): app.add_middleware(AntiCrawlerMiddleware, mode=anti_crawler_mode) mode_descriptions = { - "false": "已禁用", - "strict": "严格模式", - "loose": "宽松模式", - "basic": "基础模式", + "false": t("startup.webui_anti_crawler_mode_disabled"), + "strict": t("startup.webui_anti_crawler_mode_strict"), + "loose": t("startup.webui_anti_crawler_mode_loose"), + "basic": t("startup.webui_anti_crawler_mode_basic"), } - mode_desc = mode_descriptions.get(anti_crawler_mode, "基础模式") - logger.info(f"🛡️ 防爬虫中间件已配置: {mode_desc}") + mode_desc = mode_descriptions.get(anti_crawler_mode, t("startup.webui_anti_crawler_mode_basic")) + logger.info(t("startup.webui_anti_crawler_configured", mode_desc=mode_desc)) except Exception as e: - logger.error(f"❌ 配置防爬虫中间件失败: {e}", exc_info=True) + logger.error(t("startup.webui_anti_crawler_config_failed", error=e), exc_info=True) def _setup_robots_txt(app: FastAPI): @@ -102,9 +334,9 @@ def _setup_robots_txt(app: FastAPI): async def robots_txt(): return create_robots_txt_response() - logger.debug("✅ robots.txt 路由已注册") + logger.debug(t("startup.webui_robots_route_registered")) except Exception as e: - logger.error(f"❌ 注册robots.txt路由失败: {e}", exc_info=True) + logger.error(t("startup.webui_robots_route_register_failed", error=e), exc_info=True) def _register_api_routes(app: FastAPI): @@ -114,9 +346,9 @@ def _register_api_routes(app: FastAPI): for router in get_all_routers(): app.include_router(router) - logger.info("✅ WebUI API 路由已注册") + logger.info(t("startup.webui_api_routes_registered")) except Exception as e: - logger.error(f"❌ 注册 WebUI API 路由失败: {e}", exc_info=True) + logger.error(t("startup.webui_api_routes_register_failed", error=e), exc_info=True) def _setup_static_files(app: FastAPI): @@ -126,20 +358,18 @@ def _setup_static_files(app: FastAPI): mimetypes.add_type("text/css", ".css") mimetypes.add_type("application/json", ".json") - static_path = _resolve_static_path() + static_path = _ensure_static_path_ready() if static_path is None: - logger.warning("❌ WebUI 静态文件目录不存在") - logger.warning("💡 请先构建前端: cd dashboard && npm run build") return if not static_path.exists(): - logger.warning(f"❌ WebUI 静态文件目录不存在: {static_path}") - logger.warning("💡 请先构建前端: cd dashboard && npm run build") + logger.warning(t("startup.webui_static_dir_missing_with_path", static_path=static_path)) + logger.warning(t("startup.webui_manual_build_hint", command=_MANUAL_BUILD_COMMAND)) return if not (static_path / "index.html").exists(): - logger.warning(f"❌ 未找到 index.html: {static_path / 'index.html'}") - logger.warning("💡 请确认前端已正确构建") + logger.warning(t("startup.webui_index_missing", index_path=static_path / "index.html")) + logger.warning(t("startup.webui_manual_build_hint", command=_MANUAL_BUILD_COMMAND)) return @app.get("/{full_path:path}", include_in_schema=False) @@ -151,7 +381,7 @@ def _setup_static_files(app: FastAPI): file_path = _resolve_safe_static_file_path(static_path, full_path) if file_path is None: - raise HTTPException(status_code=404, detail="Not Found") + raise HTTPException(status_code=404, detail=t("core.not_found")) if file_path.exists() and file_path.is_file(): media_type = mimetypes.guess_type(str(file_path))[0] @@ -164,7 +394,7 @@ def _setup_static_files(app: FastAPI): response.headers["X-Robots-Tag"] = "noindex, nofollow, noarchive" return response - logger.info(f"✅ WebUI 静态文件服务已配置: {static_path}") + logger.info(t("startup.webui_static_files_configured", static_path=static_path)) def _resolve_static_path() -> Path | None: @@ -178,8 +408,8 @@ def _resolve_static_path() -> Path | None: except Exception: pass - base_dir = Path(__file__).parent.parent.parent - static_path = base_dir / "webui" / "dist" + base_dir = _get_project_root() + static_path = base_dir / "dashboard" / "dist" if static_path.exists(): return static_path return None @@ -192,7 +422,7 @@ def show_access_token(): token_manager = get_token_manager() current_token = token_manager.get_token() - logger.info(f"🔑 WebUI Access Token: {current_token}") - logger.info("💡 请使用此 Token 登录 WebUI") + logger.info(t("startup.webui_access_token", token=current_token)) + logger.info(t("startup.webui_access_token_login_hint")) except Exception as e: - logger.error(f"❌ 获取 Access Token 失败: {e}") + logger.error(t("startup.webui_access_token_failed", error=e))