feat(trimemedia): 支持飞牛影视访问码

飞牛开启访问码后，需先请求 /c/{访问码} 获得会话凭证，
否则登录接口与图片等资源均返回 404。

- Api 增加 access_code 参数及 verify_access_code()，登录前先校验
- 创建 API 连接时校验访问码，失败则关闭会话
- 图片 Cookies 附带访问码校验凭证，避免代理图片 404

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

app/modules/trimemedia/api.py

@@ -5,6 +5,7 @@ import time
 from dataclasses import dataclass
 from enum import Enum
 from typing import List, Optional, Union
+from urllib.parse import quote
 
 from app.core.config import settings
 from app.log import logger
@@ -108,6 +109,7 @@ class Api:
         "_host",
         "_token",
         "_apikey",
+        "_access_code",
         "_api_path",
         "_request_utils",
         "_version",
@@ -130,18 +132,51 @@ class Api:
     def version(self) -> Optional[Version]:
         return self._version
 
-    def __init__(self, host: str, apikey: str):
+    @property
+    def cookies(self) -> dict:
+        """
+        当前会话的Cookies，开启访问码后包含访问码校验凭证
+        """
+        return self._session.cookies.get_dict()
+
+    def __init__(self, host: str, apikey: str, access_code: Optional[str] = None):
         """
         :param host: 飞牛服务端地址，如http://127.0.0.1:5666/v
+        :param access_code: 访问码，未开启时为空
         """
         self._api_path = "/api/v1"
         self._host = host.rstrip("/")
         self._apikey = apikey
+        self._access_code = access_code
         self._token: Optional[str] = None
         self._version: Optional[Version] = None
         self._session = requests.Session()
         self._request_utils = RequestUtils(session=self._session, timeout=10)
 
+    def verify_access_code(self) -> bool:
+        """
+        校验访问码，通过后会话获得访问凭证，否则无法访问登录页和各应用接口
+
+        :return: 未配置访问码或校验通过返回True
+        """
+        if not self._access_code:
+            return True
+        # 访问码校验地址位于设备根路径，不在/v下
+        root = self._host[: -len("/v")] if self._host.endswith("/v") else self._host
+        url = f"{root}/c/{quote(self._access_code, safe='')}"
+        res = self._request_utils.get_res(url, allow_redirects=True)
+        if res is None:
+            logger.error(f"校验飞牛访问码失败，无法访问 {url}")
+            return False
+        if res.status_code == 404:
+            # 访问码错误或校验失败时返回404
+            logger.error("飞牛访问码校验失败，请检查访问码是否正确")
+            return False
+        if not res.ok:
+            logger.error(f"飞牛访问码校验失败，状态码：{res.status_code}")
+            return False
+        return True
+
     def sys_version(self) -> Optional[Version]:
         """
         飞牛影视版本号
@@ -161,6 +196,9 @@ class Api:
 
         :return: 成功返回token 否则返回None
         """
+        # 开启访问码后需先通过访问码校验，否则无法访问登录接口
+        if not self.verify_access_code():
+            return None
         if (
             res := self.request(
                 "/login",

app/modules/trimemedia/trimemedia.py

@@ -12,6 +12,7 @@ from app.utils.url import UrlUtils
 class TrimeMedia:
     _username: Optional[str] = None
     _password: Optional[str] = None
+    _access_code: Optional[str] = None
 
     _userinfo: Optional[fnapi.User] = None
     _host: Optional[str] = None
@@ -28,6 +29,7 @@ class TrimeMedia:
         host: Optional[str] = None,
         username: Optional[str] = None,
         password: Optional[str] = None,
+        access_code: Optional[str] = None,
         play_host: Optional[str] = None,
         sync_libraries: Optional[list] = None,
         **kwargs,
@@ -37,13 +39,14 @@ class TrimeMedia:
             return
         self._username = username
         self._password = password
+        self._access_code = access_code
         self._host = host
         self._sync_libraries = sync_libraries or []
 
         if not self.reconnect():
             logger.error(f"请检查服务端地址 {host}")
             return
-        if result := self.__create_api(play_host):
+        if result := self.__create_api(play_host, access_code):
             self._playhost = result.api.host
             result.api.close()
         elif play_host:
@@ -69,11 +72,14 @@ class TrimeMedia:
         version: fnapi.Version
 
     @staticmethod
-    def __create_api(host: Optional[str]) -> Optional["TrimeMedia._ApiCreateResult"]:
+    def __create_api(
+        host: Optional[str], access_code: Optional[str] = None
+    ) -> Optional["TrimeMedia._ApiCreateResult"]:
         """
         创建一个飞牛API
 
         :param host:  服务端地址
+        :param access_code: 访问码，未开启时为空
         :return: 如果地址无效、不可访问则返回None
         """
 
@@ -85,16 +91,19 @@ class TrimeMedia:
         if not host.endswith("/v"):
             # 尝试补上结尾的/v 测试能否正常访问
             res = TrimeMedia._ApiCreateResult()
-            res.api = fnapi.Api(host + "/v", api_key)
+            res.api = fnapi.Api(host + "/v", api_key, access_code)
-            if fnver := res.api.sys_version():
+            # 开启访问码后，需先校验才能访问各应用接口
+            if res.api.verify_access_code() and (fnver := res.api.sys_version()):
                 res.version = fnver
                 return res
+            res.api.close()
         # 测试用户配置的地址
         res = TrimeMedia._ApiCreateResult()
-        res.api = fnapi.Api(host, api_key)
+        res.api = fnapi.Api(host, api_key, access_code)
-        if fnver := res.api.sys_version():
+        if res.api.verify_access_code() and (fnver := res.api.sys_version()):
             res.version = fnver
             return res
+        res.api.close()
         return None
 
     def close(self):
@@ -130,7 +139,7 @@ class TrimeMedia:
         if not self.is_configured():
             return False
         self.disconnect()
-        if result := self.__create_api(self._host):
+        if result := self.__create_api(self._host, self._access_code):
             self._api = result.api
             self._version = result.version
             # 版本号:0.8.53, 服务版本:0.8.23
@@ -683,4 +692,8 @@ class TrimeMedia:
             image_url, [self._api.host], strict=True
         ):
             return None
-        return {"Trim-MC-token": self._api.token}
+        cookies = {"Trim-MC-token": self._api.token}
+        if self._access_code:
+            # 开启访问码后，图片请求也需要携带访问码校验凭证
+            cookies.update(self._api.cookies)
+        return cookies