feat(api): 更新直播相关接口

Author: JoeanAmier

src/application/TikTokDownloader.py

@@ -40,23 +40,10 @@
 
 # from typing import Type
 # from webbrowser import open
-# from flask import abort
-# from flask import request
-
-# from .main_server import Server
-# from .main_web_UI import WebUI
 
 __all__ = ["TikTokDownloader"]
 
 
-def server_tip(function):
-    async def inner(self, *args, **kwargs):
-        self.console.print(_("正在启动服务器，如需关闭服务器，请按下 Ctrl + C"))
-        return await function(self, *args, **kwargs)
-
-    return inner
-
-
 class TikTokDownloader:
     VERSION_MAJOR = VERSION_MAJOR
     VERSION_MINOR = VERSION_MINOR
@@ -150,7 +137,6 @@ async def disable_function(
             "该功能正在重构，未来开发完成重新开放！",
         )
 
-    @server_tip
     async def server(self):
         try:
             self.console.print(
@@ -169,12 +155,6 @@ async def server(self):
         except KeyboardInterrupt:
             self.running = False
 
-    # def __web_ui_object(self):
-    #     self.server(WebUI, token=False)
-
-    # def __server_object(self):
-    #     self.server(Server)
-
     async def __modify_record(self):
         await self.change_config("Record")
 

src/application/main_server.py

@@ -30,6 +30,8 @@
     Reply,
     Mix,
     MixTikTok,
+    Live,
+    LiveTikTok,
 )
 from ..translation import _
 from .main_terminal import TikTok
@@ -45,7 +47,7 @@ def token_dependency(token: str = Header(None)):
     if not is_valid_token(token):
         raise HTTPException(
             status_code=403,
-            detail=_("无效令牌！"),
+            detail=_("验证失败！"),
         )
 
 
@@ -248,6 +250,34 @@ async def handle_mix(extract: Mix, token: str = Depends(token_dependency)):
                 return self.success_response(extract, data)
             return self.failed_response(extract)
 
+        @self.server.post(
+            "/douyin/live",
+            summary=_("获取直播数据"),
+            description=dedent(
+                _("""
+                                        待更新
+                                                """)
+            ),
+            tags=[_("抖音")],
+            response_model=DataResponse,
+        )
+        async def handle_live(extract: Live, token: str = Depends(token_dependency)):
+            if self.check_live_params(
+                extract.web_rid,
+                extract.room_id,
+                extract.sec_user_id,
+            ):
+                if data := await self.handle_live(
+                    extract,
+                ):
+                    return self.success_response(extract, data[0])
+                return self.failed_response(extract)
+            return DataResponse(
+                message=_("参数错误！"),
+                data=None,
+                params=extract.model_dump(),
+            )
+
         @self.server.post(
             "/douyin/comment",
             summary=_("获取作品评论数据"),
@@ -314,7 +344,7 @@ async def handle_reply(extract: Reply, token: str = Depends(token_dependency)):
             tags=[_("抖音")],
             response_model=DataResponse,
         )
-        async def handle_general(
+        async def handle_search_general(
             extract: GeneralSearch, token: str = Depends(token_dependency)
         ):
             return await self.handle_search(extract)
@@ -330,7 +360,7 @@ async def handle_general(
             tags=[_("抖音")],
             response_model=DataResponse,
         )
-        async def handle_video(
+        async def handle_search_video(
             extract: VideoSearch, token: str = Depends(token_dependency)
         ):
             return await self.handle_search(extract)
@@ -346,7 +376,7 @@ async def handle_video(
             tags=[_("抖音")],
             response_model=DataResponse,
         )
-        async def handle_user(
+        async def handle_search_user(
             extract: UserSearch, token: str = Depends(token_dependency)
         ):
             return await self.handle_search(extract)
@@ -362,7 +392,7 @@ async def handle_user(
             tags=[_("抖音")],
             response_model=DataResponse,
         )
-        async def handle_live(
+        async def handle_search_live(
             extract: LiveSearch, token: str = Depends(token_dependency)
         ):
             return await self.handle_search(extract)
@@ -452,6 +482,27 @@ async def handle_mix_tiktok(
                 return self.success_response(extract, data)
             return self.failed_response(extract)
 
+        @self.server.post(
+            "/tiktok/live",
+            summary=_("获取直播数据"),
+            description=dedent(
+                _("""
+                                                        待更新
+                                                                """)
+            ),
+            tags=["TikTok"],
+            response_model=DataResponse,
+        )
+        async def handle_live_tiktok(
+            extract: Live, token: str = Depends(token_dependency)
+        ):
+            if data := await self.handle_live(
+                extract,
+                True,
+            ):
+                return self.success_response(extract, data[0])
+            return self.failed_response(extract)
+
     async def handle_search(self, extract):
         if data := await self.deal_search_data(
             extract,
@@ -523,3 +574,35 @@ def generate_mix_params(mix_id: str = None, detail_id: str = None):
         if mix_id:
             return True, mix_id
         return (False, detail_id) if detail_id else (None, None)
+
+    @staticmethod
+    def check_live_params(
+        web_rid: str = None,
+        room_id: str = None,
+        sec_user_id: str = None,
+    ) -> bool:
+        return bool(web_rid or room_id and sec_user_id)
+
+    async def handle_live(self, extract: Live | LiveTikTok, tiktok=False):
+        if tiktok:
+            data = await self.get_live_data_tiktok(
+                extract.room_id,
+                extract.cookie,
+                extract.proxy,
+            )
+        else:
+            data = await self.get_live_data(
+                extract.web_rid,
+                extract.room_id,
+                extract.sec_user_id,
+                extract.cookie,
+                extract.proxy,
+            )
+        if extract.source:
+            return data
+        return await self.extractor.run(
+            [data],
+            None,
+            "live",
+            tiktok=tiktok,
+        )

src/application/main_terminal.py

@@ -1108,10 +1108,33 @@ def _choice_live_quality(
             return None
         return list(flv_items.values())[i], list(m3u8_items.values())[i]
 
-    async def live_interactive(
+    async def get_live_data(
         self,
+        web_rid: str = None,
+        room_id: str = None,
+        sec_user_id: str = None,
         cookie: str = None,
         proxy: str = None,
+    ):
+        return await Live(
+            self.parameter,
+            cookie,
+            proxy,
+            web_rid,
+            room_id,
+            sec_user_id,
+        ).run()
+
+    async def get_live_data_tiktok(
+        self,
+        room_id: str = None,
+        cookie: str = None,
+        proxy: str = None,
+    ):
+        return await LiveTikTok(self.parameter, cookie, proxy, room_id).run()
+
+    async def live_interactive(
+        self,
         *args,
     ):
         while url := self._inquire_input(_("直播")):
@@ -1121,9 +1144,7 @@ async def live_interactive(
             if not params:
                 self.logger.warning(_("{} 提取直播 ID 失败").format(url=url))
                 continue
-            live_data = [
-                await Live(self.parameter, cookie, proxy, **i).run() for i in params
-            ]
+            live_data = [await self.get_live_data(**i) for i in params]
             if not [i for i in live_data if i]:
                 self.logger.warning(_("获取直播数据失败"))
                 continue
@@ -1134,18 +1155,14 @@ async def live_interactive(
 
     async def live_interactive_tiktok(
         self,
-        cookie: str = None,
-        proxy: str = None,
         *args,
     ):
         while url := self._inquire_input(_("直播")):
             __, ids = await self.links_tiktok.run(url, type_="live")
             if not ids:
                 self.logger.warning(_("{} 提取直播 ID 失败").format(url=url))
                 continue
-            live_data = [
-                await LiveTikTok(self.parameter, cookie, proxy, i).run() for i in ids
-            ]
+            live_data = [await self.get_live_data_tiktok(i) for i in ids]
             if not [i for i in live_data if i]:
                 self.logger.warning(_("获取直播数据失败"))
                 continue