feat: add backend health check for unmanaged MoviePilot processes

app/__init__.py

@@ -0,0 +1,15 @@
+import warnings
+
+
+def _filter_third_party_startup_warnings() -> None:
+    """
+    过滤第三方库在新版 Python 下产生的已知无害启动警告。
+    """
+    warnings.filterwarnings(
+        "ignore",
+        message=r"invalid escape sequence '\\&'",
+        category=SyntaxWarning,
+    )
+
+
+_filter_third_party_startup_warnings()

app/doctor/checks.py

@@ -11,6 +11,9 @@ import sys
 from collections import deque
 from pathlib import Path
 from typing import Any, Callable, Optional
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
 
 import psutil
 
@@ -33,6 +36,9 @@ CORE_DEPENDENCIES = (
     "uvicorn",
 )
 LOCAL_HOSTS = {"", "0.0.0.0", "::", "::1", "localhost"}
+BACKEND_HEALTH_PATH = "/api/v1/system/global"
+BACKEND_HEALTH_TOKEN = "moviepilot"
+BACKEND_HEALTH_TIMEOUT = 0.5
 LOG_ERROR_PATTERNS = (
     re.compile(r"\btraceback\b", re.IGNORECASE),
     re.compile(r"\b(error|critical|exception)\b", re.IGNORECASE),
@@ -220,6 +226,42 @@ def _can_connect(host: str, port: int, timeout: float = 1.0) -> tuple[bool, str]
         return False, str(err)
 
 
+def _is_moviepilot_backend_payload(payload: Any) -> bool:
+    """
+    判断本地健康接口响应是否来自 MoviePilot 后端。
+    """
+    if not isinstance(payload, dict) or payload.get("success") is False:
+        return False
+    data = payload.get("data")
+    if not isinstance(data, dict):
+        return False
+    return bool(data.get("BACKEND_VERSION"))
+
+
+def _backend_health_payload(port: int, timeout: float = BACKEND_HEALTH_TIMEOUT) -> Optional[dict[str, Any]]:
+    """
+    读取本机后端健康接口响应，用于识别非 CLI 管理的 MoviePilot 进程。
+    """
+    query = urlencode({"token": BACKEND_HEALTH_TOKEN})
+    url = f"http://{_client_host(settings.HOST)}:{port}{BACKEND_HEALTH_PATH}?{query}"
+    request = Request(url=url, headers={"Accept": "application/json"}, method="GET")
+    try:
+        with urlopen(request, timeout=timeout) as response:
+            if response.status != 200:
+                return None
+            raw = response.read().decode("utf-8", errors="ignore")
+    except (HTTPError, URLError):
+        return None
+    except OSError:
+        return None
+
+    try:
+        payload = json.loads(raw) if raw else None
+    except json.JSONDecodeError:
+        return None
+    return payload if _is_moviepilot_backend_payload(payload) else None
+
+
 def _tail_lines(path: Path, max_lines: int = 120, max_bytes: int = 256 * 1024) -> list[str]:
     try:
         size = path.stat().st_size
@@ -425,6 +467,24 @@ def _check_port(
         )
         return
 
+    if name == "backend":
+        health_payload = _backend_health_payload(port)
+        if health_payload:
+            runner.add(
+                finding_id=f"port.{name}_listening_unmanaged",
+                severity=DoctorSeverity.Info,
+                status=DoctorFindingStatus.Ok,
+                title=f"{name} 端口由 MoviePilot 后端监听",
+                detail=f"端口 {port} 健康接口响应正常；监听进程：{'; '.join(descriptions)}",
+                recommendation="Docker 或非 CLI 管理启动方式下，后端端口被当前服务监听属于正常状态。",
+                context={
+                    "port": port,
+                    "pids": [process.pid for process in occupants],
+                    "backend_version": health_payload.get("data", {}).get("BACKEND_VERSION"),
+                },
+            )
+            return
+
     runner.add(
         finding_id=f"port.{name}_occupied",
         severity=DoctorSeverity.Error,

tests/test_doctor.py

@@ -1,9 +1,12 @@
 from __future__ import annotations
 
+from types import SimpleNamespace
+
 from app.core.config import settings
-from app.doctor import run_doctor
+from app.doctor import checks, run_doctor
 from app.doctor.formatters import format_json_report, format_text_report
 from app.doctor.models import DoctorFinding, DoctorFindingStatus, DoctorSeverity
+from app.doctor.runner import DoctorRunner
 
 
 def test_doctor_report_has_stable_json_shape(tmp_path, monkeypatch):
@@ -60,3 +63,26 @@ def test_doctor_fix_removes_stale_runtime(tmp_path, monkeypatch):
     finding = report.find("runtime.backend_stale")
     assert finding is not None
     assert finding.fixed
+
+
+def test_doctor_accepts_healthy_unmanaged_backend_port(monkeypatch):
+    """doctor 在容器中应把健康的非 CLI 管理后端端口识别为正常。"""
+    occupant = SimpleNamespace(pid=12345)
+    monkeypatch.setattr(checks, "_port_occupants", lambda port: [occupant])
+    monkeypatch.setattr(checks, "_process_description", lambda process: f"PID {process.pid} (python)")
+    monkeypatch.setattr(checks, "_is_expected_port_process", lambda name, process: False)
+    monkeypatch.setattr(
+        checks,
+        "_backend_health_payload",
+        lambda port: {"success": True, "data": {"BACKEND_VERSION": "v2-test"}},
+    )
+
+    runner = DoctorRunner()
+    checks._check_port(runner, name="backend", port=3001, managed_process=None)
+
+    finding = runner.report.find("port.backend_listening_unmanaged")
+    assert finding is not None
+    assert finding.status == DoctorFindingStatus.Ok
+    assert finding.severity == DoctorSeverity.Info
+    assert finding.context["backend_version"] == "v2-test"
+    assert runner.report.find("port.backend_occupied") is None