GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-09-05 07:27:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: 完成绿联影视接入并补齐扫描模式与统计展示

Browse Source
This commit is contained in:
doumao
2026-02-28 21:58:35 +08:00
parent 383371dd6f
commit 296261da8a
10 changed files with 2548 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/manual/ugreen_media_cli.py
+299
View File
@@ -0,0 +1,299 @@
from __future__ import annotations
import argparse
import base64
import getpass
import json
import os
import sys
import uuid
from typing import Any, Mapping
from urllib.parse import urlsplit, urlunsplit
# 兼容直接运行脚本：避免 app/utils 被放在 sys.path 首位导致标准库模块被同名文件遮蔽
if __name__ == "__main__" and __package__ is None:
script_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.abspath(os.path.join(script_dir, "..", ".."))
if script_dir in sys.path:
sys.path.remove(script_dir)
if project_root not in sys.path:
sys.path.insert(0, project_root)
import requests
from app.utils.ugreen_crypto import UgreenCrypto
class UgreenLoginError(Exception):
pass
def _normalize_base_url(raw: str) -> str:
value = (raw or "").strip()
if not value:
raise UgreenLoginError("服务器地址不能为空")
if not value.startswith(("http://", "https://")):
value = f"http://{value}"
parsed = urlsplit(value)
if not parsed.netloc:
raise UgreenLoginError(f"无效服务器地址: {raw}")
return urlunsplit((parsed.scheme, parsed.netloc, "", "", "")).rstrip("/")
def _json_or_raise(resp: requests.Response, stage: str) -> dict[str, Any]:
try:
data = resp.json()
except Exception as exc: # pragma: no cover - 网络异常路径
raise UgreenLoginError(
f"{stage} 返回非 JSON，HTTP {resp.status_code}，响应片段: {resp.text[:200]}"
) from exc
if not isinstance(data, dict):
raise UgreenLoginError(f"{stage} 返回格式异常: {type(data).__name__}")
return data
def _decode_public_key(raw: str) -> str:
value = (raw or "").strip()
if not value:
raise UgreenLoginError("未获取到公钥")
if "BEGIN" in value:
return value
try:
return base64.b64decode(value).decode("utf-8")
except Exception as exc:
raise UgreenLoginError("公钥解码失败") from exc
def _raise_if_failed(payload: Mapping[str, Any], stage: str) -> None:
if payload.get("code") == 200:
return
raise UgreenLoginError(
f"{stage}失败: code={payload.get('code')} msg={payload.get('msg')}"
)
def _build_common_headers(
client_id: str, client_version: str, language: str
) -> dict[str, str]:
return {
"Accept": "application/json, text/plain, */*",
"Client-Id": client_id,
"Client-Version": client_version,
"UG-Agent": "PC/WEB",
"X-Specify-Language": language,
}
def _login_and_get_access(
session: requests.Session,
base_url: str,
username: str,
password: str,
keepalive: bool,
headers: Mapping[str, str],
timeout: float,
verify_ssl: bool,
) -> tuple[str, str]:
check_resp = session.post(
f"{base_url}/ugreen/v1/verify/check",
json={"username": username},
headers=dict(headers),
timeout=timeout,
verify=verify_ssl,
)
check_json = _json_or_raise(check_resp, "获取登录公钥")
_raise_if_failed(check_json, "获取登录公钥")
rsa_token = (
check_resp.headers.get("x-rsa-token")
or check_resp.headers.get("X-Rsa-Token")
or check_json.get("xRsaToken")
or check_json.get("x-rsa-token")
)
if not rsa_token:
data = check_json.get("data")
if isinstance(data, Mapping):
rsa_token = data.get("xRsaToken") or data.get("x-rsa-token")
if not rsa_token:
raise UgreenLoginError("登录公钥为空（x-rsa-token）")
login_public_key = _decode_public_key(str(rsa_token))
encrypted_password = UgreenCrypto(public_key=login_public_key).rsa_encrypt_long(
password
)
login_payload = {
"username": username,
"password": encrypted_password,
"keepalive": keepalive,
"otp": True,
"is_simple": True,
}
login_resp = session.post(
f"{base_url}/ugreen/v1/verify/login",
json=login_payload,
headers=dict(headers),
timeout=timeout,
verify=verify_ssl,
)
login_json = _json_or_raise(login_resp, "登录")
_raise_if_failed(login_json, "登录")
data = login_json.get("data")
if not isinstance(data, Mapping):
raise UgreenLoginError("登录成功但响应 data 为空")
token = str(data.get("token") or "").strip()
public_key = str(data.get("public_key") or "").strip()
if not token:
raise UgreenLoginError("登录成功但未拿到 token")
if not public_key:
raise UgreenLoginError("登录成功但未拿到 public_key")
return token, _decode_public_key(public_key)
def _fetch_media_lib(
session: requests.Session,
base_url: str,
token: str,
public_key: str,
client_id: str,
client_version: str,
language: str,
page: int,
page_size: int,
timeout: float,
verify_ssl: bool,
) -> Any:
crypto = UgreenCrypto(
public_key=public_key,
token=token,
client_id=client_id,
client_version=client_version,
ug_agent="PC/WEB",
language=language,
)
req = crypto.build_encrypted_request(
url=f"{base_url}/ugreen/v1/video/homepage/media_list",
method="GET",
params={"page": page, "page_size": page_size},
)
media_resp = session.get(
req.url,
headers=req.headers,
params=req.params,
timeout=timeout,
verify=verify_ssl,
)
media_json = _json_or_raise(media_resp, "获取媒体库")
return crypto.decrypt_response(media_json, req.aes_key)
def parse_args(argv: list[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="登录绿联 NAS 并调用媒体库接口（自动处理请求加密/响应解密）"
)
parser.add_argument("--host", help="服务器地址，例如: http://192.168.20.101:9999")
parser.add_argument("--username", help="用户名")
parser.add_argument("--password", help="密码（不传则交互输入）")
parser.add_argument("--client-id", help="可选，默认自动生成 UUID-WEB")
parser.add_argument("--client-version", default="76363", help="默认: 76363")
parser.add_argument("--language", default="zh-CN", help="默认: zh-CN")
parser.add_argument("--page", type=int, default=1, help="默认: 1")
parser.add_argument("--page-size", type=int, default=50, help="默认: 50")
parser.add_argument("--timeout", type=float, default=20.0, help="默认: 20 秒")
parser.add_argument("--insecure", action="store_true", help="忽略 HTTPS 证书校验")
parser.add_argument(
"--no-keepalive",
action="store_true",
help="关闭保持登录（默认保持登录）",
)
parser.add_argument("--pretty", action="store_true", help="美化输出 JSON")
parser.add_argument("--output", help="将解密后的结果写入文件")
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv or sys.argv[1:])
host = args.host or input("服务器地址: ").strip()
username = args.username or input("用户名: ").strip()
password = args.password or getpass.getpass("密码: ")
client_id = (args.client_id or f"{uuid.uuid4()}-WEB").strip()
keepalive = not args.no_keepalive
verify_ssl = not args.insecure
try:
base_url = _normalize_base_url(host)
if args.insecure:
requests.packages.urllib3.disable_warnings() # type: ignore[attr-defined]
session = requests.Session()
headers = _build_common_headers(
client_id=client_id,
client_version=args.client_version,
language=args.language,
)
token, public_key = _login_and_get_access(
session=session,
base_url=base_url,
username=username,
password=password,
keepalive=keepalive,
headers=headers,
timeout=args.timeout,
verify_ssl=verify_ssl,
)
decoded = _fetch_media_lib(
session=session,
base_url=base_url,
token=token,
public_key=public_key,
client_id=client_id,
client_version=args.client_version,
language=args.language,
page=args.page,
page_size=args.page_size,
timeout=args.timeout,
verify_ssl=verify_ssl,
)
if isinstance(decoded, Mapping):
if decoded.get("code") != 200:
raise UgreenLoginError(
f"媒体库接口失败: code={decoded.get('code')} msg={decoded.get('msg')}"
)
media_count = None
data = decoded.get("data")
if isinstance(data, Mapping) and isinstance(data.get("media_lib_info_list"), list):
media_count = len(data["media_lib_info_list"])
print(
f"调用成功: code={decoded.get('code')} msg={decoded.get('msg')} "
f"media_lib_info_list={media_count}"
)
text = json.dumps(
decoded,
ensure_ascii=False,
indent=2 if args.pretty else None,
separators=(",", ":") if not args.pretty else None,
)
if args.output:
with open(args.output, "w", encoding="utf-8") as f:
f.write(text)
f.write("\n")
print(f"解密结果已写入: {args.output}")
else:
print(text)
return 0
except UgreenLoginError as exc:
print(f"错误: {exc}", file=sys.stderr)
return 1
except requests.RequestException as exc:
print(f"网络错误: {exc}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())
tests/test_ugreen_mediaserver.py
+176
View File
@@ -0,0 +1,176 @@
import unittest
from unittest.mock import patch
import importlib.util
import sys
import types
from pathlib import Path
from app import schemas
try:
from app.api.endpoints import dashboard as dashboard_endpoint
except Exception:
dashboard_endpoint = None
def _load_ugreen_class():
"""
在测试中动态加载 Ugreen，避免受可选依赖（如 pyquery/sqlalchemy）影响。
"""
module_name = "_test_ugreen_module"
if module_name in sys.modules:
return sys.modules[module_name].Ugreen
# 轻量日志桩
if "app.log" not in sys.modules:
log_module = types.ModuleType("app.log")
class _Logger:
def info(self, *_args, **_kwargs):
pass
def warning(self, *_args, **_kwargs):
pass
def error(self, *_args, **_kwargs):
pass
def debug(self, *_args, **_kwargs):
pass
log_module.logger = _Logger()
sys.modules["app.log"] = log_module
# SystemConfigOper 桩
if "app.db.systemconfig_oper" not in sys.modules:
db_module = types.ModuleType("app.db.systemconfig_oper")
class _SystemConfigOper:
@staticmethod
def get(_key):
return {}
@staticmethod
def set(_key, _value):
return None
db_module.SystemConfigOper = _SystemConfigOper
sys.modules["app.db.systemconfig_oper"] = db_module
# app.modules / app.modules.ugreen / app.modules.ugreen.api 桩
if "app.modules" not in sys.modules:
pkg = types.ModuleType("app.modules")
pkg.__path__ = []
sys.modules["app.modules"] = pkg
if "app.modules.ugreen" not in sys.modules:
subpkg = types.ModuleType("app.modules.ugreen")
subpkg.__path__ = []
sys.modules["app.modules.ugreen"] = subpkg
if "app.modules.ugreen.api" not in sys.modules:
api_module = types.ModuleType("app.modules.ugreen.api")
class _Api:
host = ""
token = None
api_module.Api = _Api
sys.modules["app.modules.ugreen.api"] = api_module
ugreen_path = Path(__file__).resolve().parents[1] / "app" / "modules" / "ugreen" / "ugreen.py"
spec = importlib.util.spec_from_file_location(module_name, ugreen_path)
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
assert spec and spec.loader
spec.loader.exec_module(module)
return module.Ugreen
Ugreen = _load_ugreen_class()
class _FakeUgreenApi:
host = "http://127.0.0.1:9999"
token = "test-token"
@staticmethod
def video_all(classification: int, page: int = 1, page_size: int = 1):
if classification == -102:
return {"total_num": 12}
if classification == -103:
return {"total_num": 34}
return {"total_num": 0}
class UgreenScanModeTest(unittest.TestCase):
def test_resolve_scan_type(self):
resolve = Ugreen._Ugreen__resolve_scan_type
self.assertEqual(resolve(scan_mode="new_and_modified"), 1)
self.assertEqual(resolve(scan_mode="supplement_missing"), 2)
self.assertEqual(resolve(scan_mode="full_override"), 3)
self.assertEqual(resolve(scan_mode="1"), 1)
self.assertEqual(resolve(scan_mode="2"), 2)
self.assertEqual(resolve(scan_mode="3"), 3)
self.assertEqual(resolve(scan_type=1), 1)
self.assertEqual(resolve(scan_type=2), 2)
self.assertEqual(resolve(scan_type=3), 3)
self.assertEqual(resolve(scan_mode="unknown"), 2)
self.assertEqual(resolve(), 2)
class UgreenStatisticTest(unittest.TestCase):
def test_get_medias_count_episode_is_none(self):
ugreen = Ugreen.__new__(Ugreen)
ugreen._host = "http://127.0.0.1:9999"
ugreen._username = "tester"
ugreen._password = "secret"
ugreen._userinfo = {"name": "tester"}
ugreen._api = _FakeUgreenApi()
stat = ugreen.get_medias_count()
self.assertEqual(stat.movie_count, 12)
self.assertEqual(stat.tv_count, 34)
self.assertIsNone(stat.episode_count)
class DashboardStatisticTest(unittest.TestCase):
@unittest.skipIf(dashboard_endpoint is None, "dashboard endpoint dependencies are missing")
def test_statistic_all_episode_missing(self):
mocked_stats = [
schemas.Statistic(movie_count=10, tv_count=20, episode_count=None, user_count=2),
schemas.Statistic(movie_count=1, tv_count=2, episode_count=None, user_count=1),
]
with patch(
"app.api.endpoints.dashboard.DashboardChain.media_statistic",
return_value=mocked_stats,
):
ret = dashboard_endpoint.statistic(name="ugreen", _=None)
self.assertEqual(ret.movie_count, 11)
self.assertEqual(ret.tv_count, 22)
self.assertEqual(ret.user_count, 3)
self.assertIsNone(ret.episode_count)
@unittest.skipIf(dashboard_endpoint is None, "dashboard endpoint dependencies are missing")
def test_statistic_mixed_episode_count(self):
mocked_stats = [
schemas.Statistic(movie_count=10, tv_count=20, episode_count=None, user_count=2),
schemas.Statistic(movie_count=1, tv_count=2, episode_count=6, user_count=1),
]
with patch(
"app.api.endpoints.dashboard.DashboardChain.media_statistic",
return_value=mocked_stats,
):
ret = dashboard_endpoint.statistic(name="all", _=None)
self.assertEqual(ret.movie_count, 11)
self.assertEqual(ret.tv_count, 22)
self.assertEqual(ret.user_count, 3)
self.assertEqual(ret.episode_count, 6)
if __name__ == "__main__":
unittest.main()