fix(agent): resolve copied skill runtime roots

This commit is contained in:
jxxghp
2026-10-01 08:40:53 +08:00
parent a7626a2f0f
commit 1cc9968865
10 changed files with 211 additions and 12 deletions

View File

@@ -24,6 +24,19 @@ This skill is the direct SQL boundary. It is implemented as a Python script and
is appropriate when the agent must inspect records, run data statistics, repair
stuck state, or perform an explicitly requested database update.
Run the bundled copy from the MoviePilot program directory with its project
runtime, for example:
```bash
cd <MOVIEPILOT_ROOT>
venv/bin/python skills/database-operation/scripts/mp-db.py tables
```
The runtime sets `MOVIEPILOT_ROOT` for copied skills. If you run a copied
script directly from `<CONFIG_PATH>/agent/skills/`, set that variable to the
program directory first; otherwise the helper reports how to relocate it.
Use the project runtime instead of a system `python3`.
Prefer safer product surfaces first:
| Request | Preferred skill |

View File

@@ -3,6 +3,7 @@
import argparse
import json
import os
import re
import sys
from dataclasses import dataclass
@@ -14,7 +15,23 @@ from sqlalchemy.engine import Engine
from sqlalchemy.exc import SQLAlchemyError
SCRIPT_PATH = Path(__file__).resolve()
PROJECT_ROOT = SCRIPT_PATH.parents[3]
def _find_project_root() -> Optional[Path]:
"""从运行环境或目录标记定位 MoviePilot 程序目录。"""
candidates = []
configured_root = os.environ.get("MOVIEPILOT_ROOT", "").strip()
if configured_root:
candidates.append(Path(configured_root).expanduser())
for base in (Path.cwd().resolve(), SCRIPT_PATH.parent):
candidates.extend((base, *base.parents))
for candidate in candidates:
if (candidate / "app" / "runtime" / "config.py").is_file():
return candidate
return None
PROJECT_ROOT = _find_project_root()
WRITE_STATEMENT_RE = re.compile(
r"^\s*(insert|update|delete|drop|alter|truncate|create|replace)\b",
re.IGNORECASE,
@@ -137,6 +154,11 @@ ACTIONS: dict[str, ActionSpec] = {
def _ensure_project_import() -> None:
"""确保脚本可以从任意工作目录导入 MoviePilot 项目模块。"""
if PROJECT_ROOT is None:
raise RuntimeError(
"无法定位 MoviePilot 程序目录;请在程序目录执行,或设置 MOVIEPILOT_ROOT="
"<程序目录> 后重试。"
)
project_path = str(PROJECT_ROOT)
if project_path not in sys.path:
sys.path.insert(0, project_path)

View File

@@ -16,6 +16,20 @@ Use `scripts/mp-downloader.py`. The helper reads MoviePilot's local downloader
configuration and credentials itself. Never request, print, or pass a host,
username, password, API key, Cookie, or arbitrary URL.
Run the bundled copy from the MoviePilot program directory with its project
runtime, for example:
```bash
cd <MOVIEPILOT_ROOT>
venv/bin/python skills/downloader-operation/scripts/mp-downloader.py instances
```
The runtime sets `MOVIEPILOT_ROOT` for copied skills. If you run a copied
script directly from `<CONFIG_PATH>/agent/skills/`, set that variable to the
program directory first; otherwise the helper returns a clear relocation
error. Do not use a system `python3` when it does not meet MoviePilot's
required Python version.
## Boundary
- Prefer `moviepilot-api` for ordinary MoviePilot acquisition workflows,

View File

@@ -6,6 +6,7 @@ from __future__ import annotations
import argparse
import importlib
import json
import os
import sys
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
@@ -13,7 +14,23 @@ from pathlib import Path
from typing import Any, Optional
SCRIPT_PATH = Path(__file__).resolve()
PROJECT_ROOT = SCRIPT_PATH.parents[3]
def _find_project_root() -> Optional[Path]:
"""从运行环境或目录标记定位 MoviePilot 程序目录。"""
candidates = []
configured_root = os.environ.get("MOVIEPILOT_ROOT", "").strip()
if configured_root:
candidates.append(Path(configured_root).expanduser())
for base in (Path.cwd().resolve(), SCRIPT_PATH.parent):
candidates.extend((base, *base.parents))
for candidate in candidates:
if (candidate / "app" / "runtime" / "config.py").is_file():
return candidate
return None
PROJECT_ROOT = _find_project_root()
DEFAULT_LIMIT = 50
MAX_LIMIT = 200
ALL_PROVIDERS = ("qbittorrent", "transmission", "rtorrent")
@@ -303,6 +320,11 @@ ACTIONS: dict[str, ActionSpec] = {
def _ensure_project_import() -> None:
"""确保脚本从任意工作目录都可导入 MoviePilot。"""
if PROJECT_ROOT is None:
raise RuntimeError(
"无法定位 MoviePilot 程序目录;请在程序目录执行,或设置 MOVIEPILOT_ROOT="
"<程序目录> 后重试。"
)
project_path = str(PROJECT_ROOT)
if project_path not in sys.path:
sys.path.insert(0, project_path)
@@ -857,7 +879,11 @@ def main() -> int:
payload = {
"success": False,
"error_type": type(error).__name__,
"message": str(error) if isinstance(error, (ValueError, OperationError)) else "下载器调用失败",
"message": (
str(error)
if isinstance(error, (ValueError, OperationError, ImportError, SyntaxError, RuntimeError))
else "下载器调用失败"
),
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
return 1

View File

@@ -17,12 +17,19 @@ from urllib.parse import quote, urlparse
def _find_repo_root() -> Path:
"""从当前工作目录和脚本路径向上查找 MoviePilot 仓库根目录。"""
script_path = Path(__file__).resolve()
candidates = [Path.cwd().resolve(), *Path.cwd().resolve().parents]
candidates = []
configured_root = os.environ.get("MOVIEPILOT_ROOT", "").strip()
if configured_root:
candidates.append(Path(configured_root).expanduser())
candidates.extend([Path.cwd().resolve(), *Path.cwd().resolve().parents])
candidates.extend([script_path.parent, *script_path.parents])
for candidate in candidates:
if (candidate / "app" / "core" / "config.py").is_file():
if (candidate / "app" / "runtime" / "config.py").is_file():
return candidate
return script_path.parents[3]
raise RuntimeError(
"无法定位 MoviePilot 程序目录;请在程序目录执行,或设置 MOVIEPILOT_ROOT="
"<程序目录> 后重试。"
)
REPO_ROOT = _find_repo_root()

View File

@@ -17,6 +17,20 @@ Use `scripts/mp-mediaserver.py`. The helper reads MoviePilot's local server
configuration and credentials itself. Never request, print, or pass a host,
username, password, API key, token, Cookie, or arbitrary URL.
Run the bundled copy from the MoviePilot program directory with its project
runtime, for example:
```bash
cd <MOVIEPILOT_ROOT>
venv/bin/python skills/mediaserver-operation/scripts/mp-mediaserver.py instances
```
The runtime sets `MOVIEPILOT_ROOT` for copied skills. If you run a copied
script directly from `<CONFIG_PATH>/agent/skills/`, set that variable to the
program directory first; otherwise the helper returns a clear relocation
error. Do not use a system `python3` when it does not meet MoviePilot's
required Python version.
## Boundary
- Keep `library.exists` in `moviepilot-api` for canonical duplicate checks. It

View File

@@ -7,6 +7,7 @@ import argparse
import importlib
import inspect
import json
import os
import sys
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
@@ -14,7 +15,23 @@ from pathlib import Path
from typing import Any, Optional
SCRIPT_PATH = Path(__file__).resolve()
PROJECT_ROOT = SCRIPT_PATH.parents[3]
def _find_project_root() -> Optional[Path]:
"""从运行环境或目录标记定位 MoviePilot 程序目录。"""
candidates = []
configured_root = os.environ.get("MOVIEPILOT_ROOT", "").strip()
if configured_root:
candidates.append(Path(configured_root).expanduser())
for base in (Path.cwd().resolve(), SCRIPT_PATH.parent):
candidates.extend((base, *base.parents))
for candidate in candidates:
if (candidate / "app" / "runtime" / "config.py").is_file():
return candidate
return None
PROJECT_ROOT = _find_project_root()
DEFAULT_LIMIT = 50
MAX_LIMIT = 200
ALL_PROVIDERS = (
@@ -238,6 +255,11 @@ ACTIONS: dict[str, ActionSpec] = {
def _ensure_project_import() -> None:
"""确保脚本从任意工作目录都可导入 MoviePilot。"""
if PROJECT_ROOT is None:
raise RuntimeError(
"无法定位 MoviePilot 程序目录;请在程序目录执行,或设置 MOVIEPILOT_ROOT="
"<程序目录> 后重试。"
)
project_path = str(PROJECT_ROOT)
if project_path not in sys.path:
sys.path.insert(0, project_path)
@@ -710,7 +732,11 @@ def main() -> int:
payload = {
"success": False,
"error_type": type(error).__name__,
"message": str(error) if isinstance(error, (ValueError, OperationError)) else "媒体服务器调用失败",
"message": (
str(error)
if isinstance(error, (ValueError, OperationError, ImportError, SyntaxError, RuntimeError))
else "媒体服务器调用失败"
),
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
return 1

View File

@@ -23,6 +23,24 @@ DEFAULT_TIMEOUT = 60
GITHUB_API_BASE = "https://api.github.com"
USER_AGENT = "MoviePilot-Plugin-Publisher"
COMMANDS_REQUIRE_LOCAL_PLUGIN = {"preview", "push", "pull"}
def _find_project_root() -> Path:
"""从运行环境或目录标记定位 MoviePilot 程序目录。"""
candidates = []
configured_root = os.environ.get("MOVIEPILOT_ROOT", "").strip()
if configured_root:
candidates.append(Path(configured_root).expanduser())
script_path = Path(__file__).resolve()
candidates.extend([Path.cwd().resolve(), *Path.cwd().resolve().parents])
candidates.extend([script_path.parent, *script_path.parents])
for candidate in candidates:
if (candidate / "app" / "runtime" / "config.py").is_file():
return candidate
raise RuntimeError(
"无法定位 MoviePilot 程序目录;请在程序目录执行,或设置 MOVIEPILOT_ROOT="
"<程序目录> 后重试。"
)
PACKAGE_BY_VERSION = {
"legacy": ("package.json", "plugins"),
"v1": ("package.json", "plugins"),
@@ -433,7 +451,7 @@ def load_request_utils_class() -> type:
:return: RequestUtils 类
"""
try:
root = Path(__file__).resolve().parents[3]
root = _find_project_root()
if str(root) not in sys.path:
sys.path.insert(0, str(root))
from app.adapters.network.http import RequestUtils # pylint: disable=import-outside-toplevel
@@ -581,7 +599,7 @@ def load_moviepilot_settings() -> Any:
:return: settings 对象或 None
"""
try:
root = Path(__file__).resolve().parents[3]
root = _find_project_root()
if str(root) not in sys.path:
sys.path.insert(0, str(root))
from app.runtime.config import settings # pylint: disable=import-outside-toplevel

View File

@@ -21,12 +21,19 @@ from typing import Any, Iterator, Optional
def _find_repo_root() -> Path:
"""从当前工作目录或 Skill 路径定位 MoviePilot 源码根目录。"""
script_path = Path(__file__).resolve()
candidates = [Path.cwd().resolve(), *Path.cwd().resolve().parents]
candidates = []
configured_root = os.environ.get("MOVIEPILOT_ROOT", "").strip()
if configured_root:
candidates.append(Path(configured_root).expanduser())
candidates.extend([Path.cwd().resolve(), *Path.cwd().resolve().parents])
candidates.extend([script_path.parent, *script_path.parents])
for candidate in candidates:
if (candidate / "app" / "runtime" / "config.py").is_file():
return candidate
return script_path.parents[3]
raise RuntimeError(
"无法定位 MoviePilot 程序目录;请在程序目录执行,或设置 MOVIEPILOT_ROOT="
"<程序目录> 后重试。"
)
SCRIPT_ROOT = _find_repo_root()

View File

@@ -2,6 +2,7 @@
import importlib.util
import json
import os
import subprocess
import sys
from pathlib import Path
@@ -157,6 +158,57 @@ def test_service_operation_script_loads_configs_without_lifespan(
assert payload[instances_key] == []
@pytest.mark.parametrize(
"relative_path",
[
"skills/downloader-operation/scripts/mp-downloader.py",
"skills/mediaserver-operation/scripts/mp-mediaserver.py",
],
)
def test_copied_service_skill_uses_runtime_root(
relative_path: str, tmp_path: Path
) -> None:
"""配置目录中的技能副本应通过运行时根目录导入 MoviePilot。"""
copied_script = tmp_path / Path(relative_path).name
copied_script.write_bytes((PROJECT_ROOT / relative_path).read_bytes())
result = subprocess.run(
[sys.executable, str(copied_script), "instances"],
cwd=tmp_path,
env={**os.environ, "MOVIEPILOT_ROOT": str(PROJECT_ROOT)},
capture_output=True,
text=True,
check=False,
timeout=30,
)
assert result.returncode == 0, result.stdout or result.stderr
payload = json.loads(result.stdout[result.stdout.index("{") :])
assert payload["success"] is True
def test_copied_downloader_reports_missing_runtime_root(tmp_path: Path) -> None:
"""缺少程序根目录时应返回可执行的定位提示和真实错误类型。"""
copied_script = tmp_path / "mp-downloader.py"
copied_script.write_bytes(
(PROJECT_ROOT / "skills/downloader-operation/scripts/mp-downloader.py").read_bytes()
)
env = {key: value for key, value in os.environ.items() if key != "MOVIEPILOT_ROOT"}
result = subprocess.run(
[sys.executable, str(copied_script), "instances"],
cwd=tmp_path,
env=env,
capture_output=True,
text=True,
check=False,
timeout=30,
)
assert result.returncode == 1
payload = json.loads(result.stdout)
assert payload["error_type"] == "RuntimeError"
assert "MOVIEPILOT_ROOT" in payload["message"]
def test_downloader_instances_and_capabilities_do_not_expose_credentials(
downloader_module: ModuleType,
monkeypatch,