add emby/jellyfin userlogin

app/chain/user.py

@@ -0,0 +1,15 @@
+from typing import Optional
+
+from app.chain import ChainBase
+
+
+class UserChain(ChainBase):
+
+    def user_authenticate(self, name, password) -> Optional[bool]:
+        """
+        辅助完成用户认证
+        :param name: 用户名
+        :param password: 密码
+        :return: bool
+        """
+        return self.run_module("user_authenticate", name=name, password=password)

app/modules/emby/__init__.py

@@ -10,7 +10,6 @@ from app.schemas.types import MediaType
 
 
 class EmbyModule(_ModuleBase):
-
     emby: Emby = None
 
     def init_module(self) -> None:
@@ -22,6 +21,15 @@ class EmbyModule(_ModuleBase):
     def init_setting(self) -> Tuple[str, Union[str, bool]]:
         return "MEDIASERVER", "emby"
 
+    def user_authenticate(self, name, password) -> Optional[bool]:
+        """
+        使用Emby用户辅助完成用户认证
+        :param name: 用户名
+        :param password: 密码
+        :return: bool
+        """
+        return self.emby.authenticate(name, password)
+
     def webhook_parser(self, body: Any, form: Any, args: Any) -> Optional[dict]:
         """
         解析Webhook报文体

app/modules/emby/emby.py

@@ -87,6 +87,33 @@ class Emby(metaclass=Singleton):
             logger.error(f"连接Users出错：" + str(e))
         return None
 
+    def authenticate(self, username: str, password: str) -> Optional[bool]:
+        """
+        用户认证
+        """
+        if not self._host or not self._apikey:
+            return None
+        req_url = "%semby/Users/AuthenticateByName" % self._host
+        try:
+            res = RequestUtils(content_type="application/json").post_res(
+                url=req_url,
+                data=json.dumps({
+                    "Username": username,
+                    "Pw": password,
+                    "KeepMeLoggedIn": False
+                })
+            )
+            if res:
+                auth_token = res.json().get("AccessToken")
+                if auth_token:
+                    logger.info(f"用户 {username} Emby认证成功")
+                    return True
+            else:
+                logger.error(f"Users/AuthenticateByName 未获取到返回数据")
+        except Exception as e:
+            logger.error(f"连接Users/AuthenticateByName出错：" + str(e))
+        return None
+
     def get_server_id(self) -> Optional[str]:
         """
         获得服务器信息

app/modules/jellyfin/__init__.py

@@ -22,6 +22,15 @@ class JellyfinModule(_ModuleBase):
     def init_setting(self) -> Tuple[str, Union[str, bool]]:
         return "MEDIASERVER", "jellyfin"
 
+    def user_authenticate(self, name, password) -> Optional[bool]:
+        """
+        使用Emby用户辅助完成用户认证
+        :param name: 用户名
+        :param password: 密码
+        :return: bool
+        """
+        return self.jellyfin.authenticate(name, password)
+
     def webhook_parser(self, body: Any, form: Any, args: Any) -> Optional[dict]:
         """
         解析Webhook报文体

app/modules/jellyfin/jellyfin.py

@@ -1,3 +1,4 @@
+import json
 import re
 from typing import List, Union, Optional, Dict
 
@@ -83,6 +84,32 @@ class Jellyfin(metaclass=Singleton):
             logger.error(f"连接Users出错：" + str(e))
         return None
 
+    def authenticate(self, username: str, password: str) -> Optional[bool]:
+        """
+        用户认证
+        """
+        if not self._host or not self._apikey:
+            return None
+        req_url = "%sUsers/authenticatebyname" % self._host
+        try:
+            res = RequestUtils(content_type="application/json").post_res(
+                url=req_url,
+                data=json.dumps({
+                    "Username": username,
+                    "Pw": password
+                })
+            )
+            if res:
+                auth_token = res.json().get("AccessToken")
+                if auth_token:
+                    logger.info(f"用户 {username} Jellyfin认证成功")
+                    return True
+            else:
+                logger.error(f"Users/AuthenticateByName 未获取到返回数据")
+        except Exception as e:
+            logger.error(f"连接Users/AuthenticateByName出错：" + str(e))
+        return None
+
     def get_server_id(self) -> Optional[str]:
         """
         获得服务器信息