调试

data_channel/piaoquan.py

@@ -5,7 +5,7 @@ import json
 
 import cffi
 import requests
-from urllib.parse import urlencode
+from urllib.parse import urlencode, urlparse
 
 from common import Common, AliyunLogger, Feishu
 from common.sql_help import sqlCollect
@@ -432,6 +432,37 @@ class PQ:
         except Exception:
             return None
 
+    """抖音"""
+    @classmethod
+    def download_dy_video(cls, video_url, video_path_url, video_id):
+        try:
+            for i in range(3):
+                payload = {}
+                headers = {
+                    'accept': '*/*',
+                    'accept-encoding': 'identity;q=1, *;q=0',
+                    'accept-language': 'zh-CN,zh;q=0.9',
+                    'cache-control': 'no-cache',
+                    'connection': 'keep-alive',
+                    'host': urlparse(video_url).netloc,
+                    'pragma': 'no-cache',
+                    'range': 'bytes=0-',
+                    'referer': video_url,
+                    'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36',
+                }
+                response = requests.request("GET", video_url, headers=headers, data=payload)
+                if response.status_code == 206:
+                    # 以二进制写入模式打开文件
+                    video = video_path_url + str(video_id) + '.mp4'
+                    with open(f"{video}", "wb") as file:
+                        # 将响应内容写入文件
+                        file.write(response.content)
+                    time.sleep(5)
+                    return video
+            return None
+        except Exception:
+            return None
+
     """
     票圈站内视频下载
     """

video_rewriting/video_processor.py

@@ -516,6 +516,11 @@ class VideoProcessor:
             if new_video_path == None:
                 return None
             Common.logger(mark).info(f"{channel_id}视频下载成功: {new_video_path}")
+        elif "抖音" in channel_id:
+            new_video_path = PQ.download_dy_video(video_url, video_path_url, v_id)
+            if new_video_path == None:
+                return None
+            Common.logger(mark).info(f"{channel_id}视频下载成功: {new_video_path}")
         elif channel_id == "视频号历史":
             new_video_path = Oss.download_sph_ls(video_url, video_path_url, v_id)
         else: