update

main.py

@@ -46,6 +46,7 @@ from toolbox.os.command import Command
 from toolbox.porter.manager import PorterManager
 from tabs.fs_tab import get_fs_tab
 from tabs.shell_tab import get_shell_tab
+from tabs.video_download import get_video_download_tab
 from tabs.youtube_player_tab import get_youtube_player_tab
 
 
@@ -136,6 +137,7 @@ def main():
         with gr.Tabs():
             _ = get_fs_tab()
             _ = get_shell_tab()
+            _ = get_video_download_tab()
             _ = get_youtube_player_tab()
 
     # http://127.0.0.1:7870/

tabs/fs_tab.py

@@ -6,12 +6,14 @@ from project_settings import project_path
 
 
 def get_fs_tab():
+    fs_file_dir_choices = ["data/tasks", "data/video"]
     with gr.TabItem("fs"):
         with gr.Row():
             with gr.Column(scale=3):
                 fs_filename = gr.Textbox(label="filename", max_lines=10)
                 fs_file = gr.File(label="file")
-                fs_file_dir = gr.Textbox(value="data/tasks", label="file_dir")
+                fs_file_dir = gr.Dropdown(choices=fs_file_dir_choices, value=fs_file_dir_choices[0],
+                                          allow_custom_value=True, label="file_dir")
                 fs_query = gr.Button("query", variant="primary")
             with gr.Column(scale=7):
                 fs_filelist_dataset_state = gr.State(value=[])

tabs/video_download.py

@@ -0,0 +1,60 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import os
+from pathlib import Path
+
+import gradio as gr
+
+from project_settings import project_path
+from toolbox.douyin.video.download import VideoDownload
+
+
+def get_client(platform: str):
+    platform = str(platform).lower()
+    if platform == "douyin":
+        client = VideoDownload()
+    else:
+        raise AssertionError(f"invalid platform: {platform}")
+    return client
+
+
+def download_douyin_video(text: str, platform: str, filename: str):
+    if os.path.isabs(filename):
+        filename = Path(filename)
+    else:
+        filename = project_path / filename
+    filename.parent.mkdir(parents=True, exist_ok=True)
+
+    client = get_client(platform)
+
+    client.download_video_by_share_text(text, filename)
+    return filename
+
+
+def get_video_download_tab():
+    download_platform_choices = ["Douyin"]
+
+    share_text = """6.66 g@B.TL 01/22 pDH:/ 骆驼祥子的大结局，是普通人无法逃脱的命运吗 # 老舍 # 骆驼祥子  https://v.douyin.com/Bocl1I_wcdg/ 复制此链接，打开Dou音搜索，直接观看视频！"""
+
+    with gr.TabItem("download"):
+        with gr.Row():
+            with gr.Column(scale=3):
+                download_text = gr.Textbox(value=share_text, label="text", lines=3, max_lines=20)
+                download_platform = gr.Dropdown(choices=download_platform_choices, value=download_platform_choices[0], label="platform")
+                download_filename = gr.Textbox(value="data/video/download/video.mp4", label="filename")
+
+                download_button = gr.Button(value="download", variant="primary")
+            with gr.Column(scale=3):
+                download_output_file = gr.Textbox(label="output_file")
+
+        download_button.click(
+            download_douyin_video,
+            inputs=[download_text, download_platform, download_filename],
+            outputs=[download_output_file],
+        )
+
+    return locals()
+
+
+if __name__ == "__main__":
+    pass

toolbox/douyin/video/download.py

@@ -6,6 +6,8 @@ from datetime import datetime
 import json
 import logging
 from pathlib import Path
+import re
+import requests
 from zoneinfo import ZoneInfo
 
 from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_after_attempt, wait_fixed
@@ -306,6 +308,75 @@ class VideoDownload(DouyinClient):
                 result.append(task)
         return result
 
+    @staticmethod
+    def get_share_url_by_share_text(text: str):
+        pattern = r"https://v\.douyin\.com/[A-Za-z0-9_\-]+/"
+
+        match = re.search(pattern, text, flags=re.IGNORECASE)
+        if match is None:
+            raise AssertionError(f"no share url found; text: {text}")
+        share_url = match.group(0)
+        return share_url
+
+    def get_video_download_url_by_share_url(self, share_url: str):
+        response = requests.request(
+            "GET",
+            url=share_url,
+            headers=self.headers
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"invalid share_url: {share_url}, status_code: {response.status_code}")
+        video_id = response.url.split("?")[0].strip("/").split("/")[-1]
+        video_url = f"https://www.iesdouyin.com/share/video/{video_id}"
+        headers = {
+            "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 17_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) EdgiOS/121.0.2277.107 Version/17.0 Mobile/15E148 Safari/604.1"
+        }
+        response = requests.request(
+            "GET",
+            url=video_url,
+            headers=headers
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; video_url: {video_url}")
+
+        # 使用正则表达式提取视频信息
+        pattern = re.compile(
+            pattern=r"window\._ROUTER_DATA\s*=\s*(.*?)</script>",
+            flags=re.DOTALL
+        )
+        match = pattern.search(response.text)
+        if match is None:
+            raise AssertionError(f"pattern parse failed; text: {response.text}")
+
+        js = json.loads(match.group(1).strip())
+        data = js["loaderData"]["video_(id)/page"]["videoInfoRes"]["item_list"][0]
+        video_download_url = data["video"]["play_addr"]["url_list"][0].replace("playwm", "play")
+        return video_download_url
+
+    def download_video_by_video_download_url(self, video_download_url: str, filename: str):
+        headers = {
+            "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 17_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) EdgiOS/121.0.2277.107 Version/17.0 Mobile/15E148 Safari/604.1"
+        }
+        response = requests.request(
+            "GET",
+            url=video_download_url,
+            headers=headers,
+            stream=True
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, url: {video_download_url}")
+
+        with open(filename, "wb") as f:
+            for chunk in response.iter_content(chunk_size=8192):
+                f.write(chunk)
+        return filename
+
+    def download_video_by_share_text(self, text: str, filename: str):
+        share_url = self.get_share_url_by_share_text(text)
+        video_download_url = self.get_video_download_url_by_share_url(share_url)
+        self.download_video_by_video_download_url(video_download_url, filename)
+        return filename
+
 
 def get_args():
     parser = argparse.ArgumentParser()
@@ -394,5 +465,19 @@ async def main2():
     return
 
 
+def main3():
+    client = VideoDownload()
+
+    text = """
+    6.66 g@B.TL 01/22 pDH:/ 骆驼祥子的大结局，是普通人无法逃脱的命运吗 # 老舍 # 骆驼祥子  https://v.douyin.com/Bocl1I_wcdg/ 复制此链接，打开Dou音搜索，直接观看视频！
+    """
+
+    filename = client.download_video_by_share_text(text, "test.mp4")
+    print(filename)
+
+    return
+
+
 if __name__ == "__main__":
-    asyncio.run(main2())
+    # asyncio.run(main2())
+    main3()