fix(emby): fall back to configured user (#6222)

app/modules/emby/emby.py

@@ -191,7 +191,12 @@ class Emby:
 
     def get_user(self, user_name: Optional[str] = None) -> Optional[Union[str, int]]:
         """
-        获得管理员用户
+        获取用于查询用户范围数据的用户ID
+
+        优先匹配指定用户名，其次匹配媒体服务器配置用户名，最后回退管理员。
+
+        :param user_name: 优先匹配的用户名
+        :return: 匹配到的用户ID，未找到可用用户时返回None
         """
         if not self._host or not self._apikey:
             return None
@@ -203,15 +208,18 @@ class Emby:
             res = RequestUtils().get_res(url, params)
             if res:
                 users = res.json()
-                # 先查询是否有与当前用户名称匹配的
-                if user_name:
-                    for user in users:
-                        if user.get("Name") == user_name:
-                            return user.get("Id")
+                candidate_usernames = []
+                for candidate_username in (user_name, self._username):
+                    if candidate_username and candidate_username not in candidate_usernames:
+                        candidate_usernames.append(candidate_username)
+                for candidate_username in candidate_usernames:
+                    for emby_user in users:
+                        if emby_user.get("Name") == candidate_username:
+                            return emby_user.get("Id")
                 # 查询管理员
-                for user in users:
-                    if user.get("Policy", {}).get("IsAdministrator"):
-                        return user.get("Id")
+                for emby_user in users:
+                    if emby_user.get("Policy", {}).get("IsAdministrator"):
+                        return emby_user.get("Id")
             else:
                 logger.error(f"Users 未获取到返回数据")
         except Exception as e:

tests/test_emby_user_resolution.py

@@ -0,0 +1,59 @@
+from unittest.mock import Mock, patch
+
+from app.modules.emby.emby import Emby
+
+
+def _resolve_user(users: list[dict], requested_username: str, configured_username: str):
+    emby = Emby.__new__(Emby)
+    emby._host = "http://emby.local/"
+    emby._apikey = "test-api-key"
+    emby._username = configured_username
+
+    response = Mock()
+    response.json.return_value = users
+    with patch("app.modules.emby.emby.RequestUtils") as request_utils:
+        request_utils.return_value.get_res.return_value = response
+        return emby.get_user(requested_username)
+
+
+def test_get_user_prefers_requested_username():
+    """指定用户名存在时应优先使用该用户。"""
+    result = _resolve_user(
+        users=[
+            {"Id": "requested-id", "Name": "mp-user", "Policy": {}},
+            {"Id": "configured-id", "Name": "configured-user", "Policy": {}},
+            {"Id": "admin-id", "Name": "admin", "Policy": {"IsAdministrator": True}},
+        ],
+        requested_username="mp-user",
+        configured_username="configured-user",
+    )
+
+    assert result == "requested-id"
+
+
+def test_get_user_falls_back_to_configured_username():
+    """指定用户名不存在时应回退媒体服务器配置用户。"""
+    result = _resolve_user(
+        users=[
+            {"Id": "configured-id", "Name": "configured-user", "Policy": {}},
+            {"Id": "admin-id", "Name": "admin", "Policy": {"IsAdministrator": True}},
+        ],
+        requested_username="missing-mp-user",
+        configured_username="configured-user",
+    )
+
+    assert result == "configured-id"
+
+
+def test_get_user_falls_back_to_administrator():
+    """指定用户和配置用户均不存在时应回退管理员。"""
+    result = _resolve_user(
+        users=[
+            {"Id": "regular-id", "Name": "regular", "Policy": {}},
+            {"Id": "admin-id", "Name": "admin", "Policy": {"IsAdministrator": True}},
+        ],
+        requested_username="missing-mp-user",
+        configured_username="missing-configured-user",
+    )
+
+    assert result == "admin-id"