| """系统更新服务：用于 Admin 一键触发受控 Git 同步。""" |
|
|
| from __future__ import annotations |
|
|
| import logging |
| from datetime import UTC, datetime |
| from pathlib import Path |
| from typing import Any, Final, cast |
| from uuid import uuid4 |
|
|
| from django.core.cache import cache |
|
|
| from apps.core.exceptions import ExternalServiceError |
| from apps.core.infrastructure.subprocess_runner import SubprocessOutput, SubprocessRunner |
| from apps.core.tasking import TaskSubmissionService |
|
|
| logger = logging.getLogger(__name__) |
|
|
| _SYSTEM_UPDATE_STATE_CACHE_KEY: Final[str] = "core:system_update:state" |
| _SYSTEM_UPDATE_LOCK_CACHE_KEY: Final[str] = "core:system_update:lock" |
| _DEFAULT_LOCK_TIMEOUT_SECONDS: Final[int] = 30 * 60 |
| _DEFAULT_TASK_TIMEOUT_SECONDS: Final[int] = 15 * 60 |
| _DEFAULT_STEP_TIMEOUT_SECONDS: Final[int] = 60 |
|
|
|
|
| class SystemUpdateService: |
| """封装系统更新触发、执行与状态管理。""" |
|
|
| def __init__( |
| self, |
| *, |
| task_submission_service: TaskSubmissionService, |
| subprocess_runner: SubprocessRunner | None = None, |
| repo_root: Path | None = None, |
| default_branch: str = "main", |
| lock_timeout_seconds: int = _DEFAULT_LOCK_TIMEOUT_SECONDS, |
| task_timeout_seconds: int = _DEFAULT_TASK_TIMEOUT_SECONDS, |
| step_timeout_seconds: int = _DEFAULT_STEP_TIMEOUT_SECONDS, |
| ) -> None: |
| self._task_submission_service = task_submission_service |
| self._subprocess_runner = subprocess_runner or SubprocessRunner(allowed_programs={"git", "uv"}) |
| self._repo_root = repo_root or self._detect_repo_root() |
| self._backend_root = self._resolve_backend_root(self._repo_root) |
| self._default_branch = default_branch |
| self._lock_timeout_seconds = lock_timeout_seconds |
| self._task_timeout_seconds = task_timeout_seconds |
| self._step_timeout_seconds = step_timeout_seconds |
|
|
| def trigger_update(self, *, triggered_by: str, enable_post_update_setup: bool = False) -> dict[str, Any]: |
| """触发异步更新任务。""" |
| current_state = self.get_state() |
| if str(current_state.get("status")) in {"queued", "running"}: |
| return { |
| "accepted": False, |
| "message": "已有更新任务在执行，请稍后刷新查看状态。", |
| "state": current_state, |
| } |
|
|
| run_id = str(uuid4()) |
| acquired = bool(cache.add(_SYSTEM_UPDATE_LOCK_CACHE_KEY, run_id, timeout=self._lock_timeout_seconds)) |
| if not acquired: |
| return { |
| "accepted": False, |
| "message": "系统正在执行另一项更新任务，请稍后重试。", |
| "state": self.get_state(), |
| } |
|
|
| options = {"enable_post_update_setup": bool(enable_post_update_setup)} |
| base_state = self._build_state( |
| run_id=run_id, |
| status="queued", |
| message="更新任务已提交，等待队列执行。", |
| triggered_by=triggered_by, |
| started_at=self._now_iso(), |
| steps=[], |
| options=options, |
| ) |
| self._save_state(base_state) |
|
|
| try: |
| q_task_id = self._task_submission_service.submit( |
| "apps.core.services.system_update_service.run_system_update_task", |
| kwargs={ |
| "run_id": run_id, |
| "triggered_by": triggered_by, |
| "enable_post_update_setup": bool(enable_post_update_setup), |
| }, |
| task_name=f"system_update:{run_id}", |
| timeout=self._task_timeout_seconds, |
| ) |
| except Exception as exc: |
| self._release_lock(run_id) |
| failed_state = self._build_state( |
| run_id=run_id, |
| status="failed", |
| message=f"提交更新任务失败：{exc}", |
| triggered_by=triggered_by, |
| started_at=base_state["started_at"], |
| finished_at=self._now_iso(), |
| steps=base_state["steps"], |
| options=options, |
| ) |
| self._save_state(failed_state) |
| logger.exception("system_update_submit_failed") |
| return { |
| "accepted": False, |
| "message": "更新任务提交失败，请查看日志后重试。", |
| "state": failed_state, |
| } |
|
|
| queued_state = self._build_state( |
| run_id=run_id, |
| status="queued", |
| message="更新任务已进入队列。", |
| triggered_by=triggered_by, |
| q_task_id=q_task_id, |
| started_at=base_state["started_at"], |
| steps=base_state["steps"], |
| options=options, |
| ) |
| self._save_state(queued_state) |
| return { |
| "accepted": True, |
| "message": "更新任务已提交。", |
| "state": queued_state, |
| } |
|
|
| def run_update( |
| self, |
| *, |
| run_id: str, |
| triggered_by: str, |
| enable_post_update_setup: bool = False, |
| ) -> dict[str, Any]: |
| """在异步任务中执行 Git 同步。""" |
| lock_owner = cache.get(_SYSTEM_UPDATE_LOCK_CACHE_KEY) |
| if lock_owner not in (None, run_id): |
| blocked_state = self._build_state( |
| run_id=run_id, |
| status="failed", |
| message="检测到并发更新任务，当前任务已终止。", |
| triggered_by=triggered_by, |
| finished_at=self._now_iso(), |
| steps=[], |
| options={"enable_post_update_setup": bool(enable_post_update_setup)}, |
| ) |
| self._save_state(blocked_state) |
| return blocked_state |
|
|
| cache.set(_SYSTEM_UPDATE_LOCK_CACHE_KEY, run_id, timeout=self._lock_timeout_seconds) |
|
|
| current = self.get_state() |
| started_at = str(current.get("started_at") or self._now_iso()) |
| q_task_id = str(current.get("q_task_id") or "") |
| options_obj = current.get("options") |
| cached_enable_post_update_setup = False |
| if isinstance(options_obj, dict): |
| cached_enable_post_update_setup = bool(options_obj.get("enable_post_update_setup")) |
| effective_enable_post_update_setup = bool(enable_post_update_setup or cached_enable_post_update_setup) |
|
|
| steps: list[dict[str, Any]] = [] |
|
|
| running_state = self._build_state( |
| run_id=run_id, |
| status="running", |
| message="正在执行系统更新，请勿重复点击。", |
| triggered_by=triggered_by, |
| q_task_id=q_task_id, |
| started_at=started_at, |
| steps=steps, |
| options={"enable_post_update_setup": effective_enable_post_update_setup}, |
| ) |
| self._save_state(running_state) |
|
|
| try: |
| self._append_step(steps, name="git_fetch", output=self._run_git(["fetch", "origin"])) |
|
|
| branch_output = self._run_git(["rev-parse", "--abbrev-ref", "HEAD"]) |
| self._append_step(steps, name="detect_branch", output=branch_output) |
| current_branch = branch_output.stdout.strip() or self._default_branch |
| if current_branch == "HEAD": |
| current_branch = self._default_branch |
|
|
| target_branch = self._resolve_pull_branch(current_branch=current_branch, steps=steps) |
| pull_output = self._run_git(["pull", "--ff-only", "origin", target_branch]) |
| self._append_step(steps, name="git_pull", output=pull_output) |
|
|
| success_message = "系统代码同步完成。" |
| if effective_enable_post_update_setup: |
| self._append_step(steps, name="uv_sync", output=self._run_uv_sync()) |
| self._append_step(steps, name="db_migrate", output=self._run_migrate()) |
| success_message = "系统代码同步与依赖/数据库更新已完成。" |
|
|
| success_state = self._build_state( |
| run_id=run_id, |
| status="success", |
| message=success_message, |
| triggered_by=triggered_by, |
| q_task_id=q_task_id, |
| started_at=started_at, |
| finished_at=self._now_iso(), |
| steps=steps, |
| options={"enable_post_update_setup": effective_enable_post_update_setup}, |
| ) |
| self._save_state(success_state) |
| logger.info( |
| "system_update_success", |
| extra={ |
| "run_id": run_id, |
| "requested_branch": current_branch, |
| "pulled_branch": target_branch, |
| "used_branch_fallback": target_branch != current_branch, |
| "enable_post_update_setup": effective_enable_post_update_setup, |
| }, |
| ) |
| return success_state |
| except ExternalServiceError as exc: |
| error_state = self._build_state( |
| run_id=run_id, |
| status="failed", |
| message="系统更新失败，请检查执行日志。", |
| triggered_by=triggered_by, |
| q_task_id=q_task_id, |
| started_at=started_at, |
| finished_at=self._now_iso(), |
| steps=steps, |
| error={ |
| "code": exc.code, |
| "message": str(exc), |
| "details": exc.errors, |
| }, |
| options={"enable_post_update_setup": effective_enable_post_update_setup}, |
| ) |
| self._save_state(error_state) |
| logger.exception("system_update_failed", extra={"run_id": run_id}) |
| raise |
| finally: |
| self._release_lock(run_id) |
|
|
| def get_state(self) -> dict[str, Any]: |
| """读取最新更新状态。""" |
| state = cache.get(_SYSTEM_UPDATE_STATE_CACHE_KEY) |
| if isinstance(state, dict): |
| if "options" not in state: |
| state["options"] = {"enable_post_update_setup": False} |
| return state |
| return self._build_state( |
| run_id="", |
| status="idle", |
| message="暂无更新记录。", |
| triggered_by="", |
| steps=[], |
| options={"enable_post_update_setup": False}, |
| ) |
|
|
| def _run_git(self, git_args: list[str]) -> SubprocessOutput: |
| args = ["git", "-C", str(self._repo_root), *git_args] |
| return self._subprocess_runner.run( |
| args=args, |
| timeout_seconds=float(self._step_timeout_seconds), |
| check=True, |
| ) |
|
|
| def _resolve_pull_branch(self, *, current_branch: str, steps: list[dict[str, Any]]) -> str: |
| if current_branch == self._default_branch: |
| return current_branch |
| if self._remote_branch_exists(current_branch): |
| return current_branch |
|
|
| fallback_message = f"远端分支 origin/{current_branch} 不存在，已回退为 origin/{self._default_branch}。" |
| self._append_step( |
| steps, |
| name="branch_fallback", |
| output=SubprocessOutput(stdout=fallback_message, stderr="", returncode=0), |
| ) |
| return self._default_branch |
|
|
| def _remote_branch_exists(self, branch: str) -> bool: |
| output = self._run_git(["ls-remote", "--heads", "origin", branch]) |
| return bool(output.stdout.strip()) |
|
|
| def _run_uv_sync(self) -> SubprocessOutput: |
| args = ["uv", "sync", "--project", str(self._backend_root)] |
| return self._subprocess_runner.run( |
| args=args, |
| timeout_seconds=float(self._step_timeout_seconds), |
| check=True, |
| ) |
|
|
| def _run_migrate(self) -> SubprocessOutput: |
| args = [ |
| "uv", |
| "run", |
| "--project", |
| str(self._backend_root), |
| "python", |
| str(self._backend_root / "manage.py"), |
| "migrate", |
| "--noinput", |
| ] |
| return self._subprocess_runner.run( |
| args=args, |
| timeout_seconds=float(self._step_timeout_seconds), |
| check=True, |
| ) |
|
|
| def _append_step(self, steps: list[dict[str, Any]], *, name: str, output: SubprocessOutput) -> None: |
| steps.append( |
| { |
| "name": name, |
| "status": "success", |
| "at": self._now_iso(), |
| "returncode": int(output.returncode), |
| "stdout": output.stdout, |
| "stderr": output.stderr, |
| } |
| ) |
|
|
| def _save_state(self, state: dict[str, Any]) -> None: |
| state["updated_at"] = self._now_iso() |
| cache.set(_SYSTEM_UPDATE_STATE_CACHE_KEY, state, timeout=7 * 24 * 60 * 60) |
|
|
| def _release_lock(self, run_id: str) -> None: |
| current = cache.get(_SYSTEM_UPDATE_LOCK_CACHE_KEY) |
| if current == run_id: |
| cache.delete(_SYSTEM_UPDATE_LOCK_CACHE_KEY) |
|
|
| def _build_state( |
| self, |
| *, |
| run_id: str, |
| status: str, |
| message: str, |
| triggered_by: str, |
| steps: list[dict[str, Any]], |
| q_task_id: str = "", |
| started_at: str | None = None, |
| finished_at: str | None = None, |
| error: dict[str, Any] | None = None, |
| options: dict[str, Any] | None = None, |
| ) -> dict[str, Any]: |
| return { |
| "run_id": run_id, |
| "status": status, |
| "message": message, |
| "triggered_by": triggered_by, |
| "q_task_id": q_task_id, |
| "started_at": started_at, |
| "finished_at": finished_at, |
| "steps": steps, |
| "error": error or {}, |
| "options": options or {"enable_post_update_setup": False}, |
| "updated_at": self._now_iso(), |
| "repo_root": str(self._repo_root), |
| } |
|
|
| def _detect_repo_root(self) -> Path: |
| current = Path(__file__).resolve() |
| for parent in (current, *current.parents): |
| if (parent / ".git").exists(): |
| return parent |
| return current.parents[4] |
|
|
| @staticmethod |
| def _resolve_backend_root(repo_root: Path) -> Path: |
| backend_root = repo_root / "backend" |
| if backend_root.exists(): |
| return backend_root |
| return repo_root |
|
|
| @staticmethod |
| def _now_iso() -> str: |
| return datetime.now(tz=UTC).isoformat() |
|
|
|
|
| def run_system_update_task( |
| *, |
| run_id: str, |
| triggered_by: str, |
| enable_post_update_setup: bool = False, |
| ) -> dict[str, Any]: |
| """供 Django-Q 调用的入口函数。""" |
| from apps.core.dependencies import build_system_update_service |
|
|
| service = cast(SystemUpdateService, build_system_update_service()) |
| return service.run_update( |
| run_id=run_id, |
| triggered_by=triggered_by, |
| enable_post_update_setup=enable_post_update_setup, |
| ) |
|
|
