update

toolbox/bilibili/bilibili_client.py

@@ -14,17 +14,17 @@ import subprocess
 import time
 from urllib.parse import urlencode, urlparse
 
+logger = logging.getLogger("toolbox")
+
 import qrcode
 import requests
 import requests.utils
+from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_after_attempt, wait_fixed
 
 from project_settings import project_path
 from toolbox.design_patterns.singleton import ParamsSingleton
 
 
-logger = logging.getLogger("toolbox")
-
-
 class BilibiliUtils(object):
     app_key = "4409e2ce8ffd12b8"
     app_sec = "59b43e04ad6965f34319062b478f83dd"
@@ -167,6 +167,11 @@ class BilibiliClient(BilibiliUtils, ParamsSingleton):
                 logger.error(f"Check failed, please check the info; status_code: {response.status_code}, text: {response.text}")
                 return False
 
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
     def get_now(self):
         url = "https://api.bilibili.com/x/report/click/now"
 
@@ -179,6 +184,11 @@ class BilibiliClient(BilibiliUtils, ParamsSingleton):
         js = response.json()
         return js
 
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
     def get_version(self):
         url = "https://api.live.bilibili.com/xlive/app-blink/v1/liveVersionInfo/getHomePageLiveVersion"
 

toolbox/bilibili/live/live_manager.py

@@ -8,16 +8,10 @@ https://github.com/ChaceQC/bilibili_live_stream_code/blob/master/main/bilibili_l
 """
 import argparse
 import logging
-import math
-from pathlib import Path
-import re
-import json
 import urllib
 import urllib.parse
 import hashlib
 
-import requests
-import requests.utils
 from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_after_attempt, wait_fixed
 
 logger = logging.getLogger("toolbox")
@@ -26,22 +20,6 @@ from project_settings import project_path
 from toolbox.bilibili.bilibili_client import BilibiliClient
 
 
-def appsign(params, appkey, appsec):
-    """
-    为请求参数进行app签名
-    :param params: 原参数
-    :param appkey: key
-    :param appsec: key对应的secret
-    :return:
-    """
-    params.update({'appkey': appkey})
-    params = dict(sorted(params.items()))  # 按照 key 重排参数
-    query = urllib.parse.urlencode(params)  # 序列化参数
-    sign = hashlib.md5((query + appsec).encode()).hexdigest()  # 计算 api 签名
-    params.update({'sign': sign})
-    return params
-
-
 class RtmpPublishCMD(object):
     def __init__(self, cmd_list: list, cmd_str: str):
         self.cmd_list = cmd_list
@@ -52,6 +30,11 @@ class BilibiliLiveManager(BilibiliClient):
     def __init__(self):
         super().__init__()
 
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
     def set_live_title(self, title: str):
         """
         :return
@@ -95,6 +78,11 @@ class BilibiliLiveManager(BilibiliClient):
         # print(json.dumps(js, ensure_ascii=False, indent=4))
         return js
 
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
     def get_room_id_by_uid(self):
         dede_user_id = self.cookies.get("DedeUserID")
 
@@ -280,6 +268,32 @@ class BilibiliLiveManager(BilibiliClient):
         js = response.json()
         return js
 
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    def get_live_status(self):
+        js = self.get_room_id_by_uid()
+        room_id = js["data"]["room_id"]
+
+        url = f"https://api.live.bilibili.com/room/v1/Room/get_info?room_id={room_id}"
+
+        response = self.session.get(
+            url,
+            headers=self.headers,
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        return js
+
+    def is_living(self):
+        js = self.get_live_status()
+        live_status = js["data"]["live_status"]
+        flag = live_status == 1
+        return flag
+
 
 def get_args():
     parser = argparse.ArgumentParser()
@@ -311,8 +325,9 @@ def main():
 
     # result = client.get_room_id()
     # result = client.stop_live()
-    result = client.start_live_by_flv_file(args.room_id)
+    # result = client.start_live_by_flv_file(args.room_id)
     # result = client.set_live_title("设置直播标题")
+    result = client.is_living()
     print(result)
     return
 

toolbox/bilibili/video/video_manager.py

@@ -14,6 +14,8 @@ import re
 
 logger = logging.getLogger("toolbox")
 
+from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_after_attempt, wait_fixed
+
 from project_settings import project_path
 from toolbox.bilibili.bilibili_client import BilibiliClient
 
@@ -22,6 +24,11 @@ class BilibiliVideoUploader(BilibiliClient):
     def __init__(self):
         super().__init__()
 
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
     def preupload(self, filename, filesize):
         """
         The preupload process to get `upos_uri` and `auth` information.
@@ -62,6 +69,11 @@ class BilibiliVideoUploader(BilibiliClient):
             raise AssertionError(f"request failed;")
         return js
 
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
     def get_upload_video_id(self, *, upos_uri, auth):
         """
         Get the `upload_id` of video.
@@ -120,10 +132,14 @@ class BilibiliVideoUploader(BilibiliClient):
                 data=batchbytes,
                 headers={"X-Upos-Auth": auth}
             )
-            assert response.status_code == 200
-            # logger.debug(f'Completed chunk{chunknum+1} uploading')
-            # print(res)
+            if response.status_code != 200:
+                raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
 
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
     def finish_upload(self, *, upos_uri, auth, filename, upload_id, biz_id, chunks):
         """
         Notify the all chunks have been uploaded.
@@ -162,6 +178,11 @@ class BilibiliVideoUploader(BilibiliClient):
             raise AssertionError(f"request failed;")
         return js
 
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
     def publish_video(self, bilibili_filename, metadata: dict):
         """
         publish the uploaded video

toolbox/douyin/homepage/follow.py

@@ -16,7 +16,7 @@ class FollowManager(DouyinClient):
     def __init__(self):
         super().__init__()
 
-    @async_cache_decorator(60)
+    @async_cache_decorator(600)
     async def get_living_list(self):
         url = "https://www.douyin.com/webcast/web/feed/follow/"
 

toolbox/douyin/live/live_recording.py

@@ -158,7 +158,7 @@ class LiveRecording(FollowManager):
                 break
         return result
 
-    @async_cache_decorator(60)
+    @async_cache_decorator(600)
     async def get_live_info_by_follow(self, room_id: str):
         result = None
         live_info_list: List[dict] = await self.get_living_list()

toolbox/douyin/video/download.py

@@ -12,6 +12,7 @@ from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_afte
 from project_settings import project_path
 from toolbox.douyin.douyin_client import DouyinClient
 from toolbox.exception import ExpectedError
+from toolbox.asyncio.cacheout import async_cache_decorator
 
 logger = logging.getLogger("toolbox")
 
@@ -48,11 +49,7 @@ class VideoDownload(DouyinClient):
         else:
             raise AssertionError(f"Got status code {response.status_code}")
 
-    # @retry(
-    #     wait=wait_fixed(10),
-    #     stop=stop_after_attempt(3),
-    #     before_sleep=before_sleep_log(logger, logging.ERROR),
-    # )
+    @async_cache_decorator(600)
     async def get_video_list_by_user_id(self, sec_user_id: str, max_cursor: int = 0, count: int = 18):
         url = "https://www.douyin.com/aweme/v1/web/aweme/post/"
 

toolbox/porter/manager.py

@@ -23,7 +23,7 @@ class PorterManager(object):
 
             if not enable:
                 continue
-            task_cls = BaseTask.by_name(task_type)
+            task_cls: BaseTask = BaseTask.by_name(task_type)
             task_obj = task_cls(**task)
 
             coro_task_set.add(task_obj.start())

toolbox/porter/tasks/douyin_live_to_bilibili_live_task.py

@@ -162,14 +162,17 @@ class DouyinLiveToBilibiliLiveTask(BaseTask):
             input_source=live_url,
             do_copy=True
         )
-        await asyncio.to_thread(
+        return_code = await asyncio.to_thread(
             self.blocking_stream, cmd
         )
-        # 每次开直播后须暂停一段时间，以避免频繁操作。
-        await asyncio.sleep(3600 * 1)
+        if return_code != 0:
+            interval = self.check_interval * 360
+            logger.info(f"{self.flag}推流进程异常结束，sleep {interval}s 以避免频繁调用。")
+            await asyncio.sleep(interval)
 
     def blocking_stream(self, cmd: RtmpPublishCMD):
-        _ = self.bilibili_client.stop_live()
+        if self.bilibili_client.is_living():
+            _ = self.bilibili_client.stop_live()
 
         logger.info(f"{self.flag}开始推流命令: \n{cmd.cmd_str}")
 
@@ -209,7 +212,7 @@ async def main():
 
     log.setup_size_rotating(log_directory=log_directory)
 
-    task = DouyinLiveToBilibiliTask(
+    task = DouyinLiveToBilibiliLiveTask(
         room_name="清源第一帅",
         room_id="139751520143",
         check_interval=10,