修复一些课程下载蓝牙音频信号失败的问题

README.md

@@ -16,9 +16,7 @@
 
 ### 网页
 
-双击运行`webui_interface.exe`文件打开网页服务器，按住`ctrl`点击命令行中的地址，或直接在浏览器中输入该地址打开。
-
-![image-20240529174653840](md/README/image-20240529174653840.png)
+双击运行`webui_interface.exe`文件打开网页服务器，会自动弹出浏览器网页。
 
 而后在打开的网页中新建任务即可。
 
@@ -124,7 +122,7 @@ pip install pyinstaller
 # 打包
 pyinstaller -F main.py
 pyinstaller -F gui.py
-pyinstakker -F webui_interface.py --add-data webui:webui
+pyinstaller -F webui_interface.py --add-data webui:webui
 pyinstaller -F gen_caption.py
 ```
 打包`gen_caption.py`时可能会失败，提示递归过深：

m3u8dl.py

@@ -12,7 +12,7 @@
 import signal
 from concurrent.futures import ThreadPoolExecutor
 
-from hashlib import md5
+import utils
 
 
 class ThreadPoolExecutorWithQueueSizeLimit(ThreadPoolExecutor):
@@ -32,9 +32,11 @@ def make_sum():
         yield ts_num
         ts_num += 1
 
+
 def dummy_func(downloaded, total, merge_status):
     return
 
+
 class M3u8Download:
     """
     :param url: 完整的m3u8文件链接 如"https://www.bilibili.com/example/index.m3u8"
@@ -45,7 +47,14 @@ class M3u8Download:
     """
 
     def __init__(
-        self, url, workDir, name, max_workers=32, num_retries=999, base64_key=None, progress_callback=dummy_func
+        self,
+        url,
+        workDir,
+        name,
+        max_workers=32,
+        num_retries=999,
+        base64_key=None,
+        progress_callback=dummy_func,
     ):
 
         self._url = url
@@ -72,12 +81,10 @@ def __init__(
             "Origin": "https://www.yanhekt.cn",
             "referer": "https://www.yanhekt.cn/",
         }
-        self.magic = "1tJrMwNq3h0yLgx86Rued2J1tFc"
-        self.updateSignature()
-
+        self.timestamp, self.signature = utils.getSignature()
         urllib3.disable_warnings()
 
-        self._url = self.encryptURL(self._url)
+        self._url = utils.encryptURL(self._url)
 
         self.get_m3u8_info(self._url, self._num_retries)
 
@@ -105,53 +112,21 @@ def signal_handler(sig, frame):
             print(f"Download successfully --> {self._name}")
             self._progress_callback(self._success_sum, self._ts_sum, 2)
 
-    def updateSignature(self):
-        self.timestamp = str(int(time.time()))
-        self.signature = md5(
-            (self.magic + "_v1_" + self.timestamp).encode()
-        ).hexdigest()
-
     def updateSignatureLoop(self):
         while self._success_sum != self._ts_sum:
-            self.updateSignature()
+            self.timestamp, self.signature = utils.getSignature()
             time.sleep(10)
 
-    def encryptURL(self, url):
-        url_list = url.split("/")
-        # "a97f12c055a10ee51d60e441e618bfef"
-        url_list.insert(-1, md5((self.magic + "_100").encode()).hexdigest())
-        return "/".join(url_list)
-
-    def getToken(self):
-        if self._token == None:
-            headers = {
-                "Xdomain-Client": "web_user",
-                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/93.0.4577.82 Safari/537.36 Edg/93.0.961.52",
-                "Origin": "https://www.yanhekt.cn",
-            }
-
-            req = requests.get(
-                "https://cbiz.yanhekt.cn/v1/auth/video/token?id=0", headers=headers
-            )
-            data = req.json()["data"]
-            self._token = data["token"]
-        return self._token
-
     def get_m3u8_info(self, m3u8_url, num_retries):
         """
         获取m3u8信息
         """
 
-        token = self.getToken()
-        url = (
-            m3u8_url
-            + "?Xvideo_Token="
-            + token
-            + "&Xclient_Timestamp="
-            + self.timestamp
-            + "&Xclient_Signature="
-            + self.signature
-            + "&Xclient_Version=v1&Platform=yhkt_user"
+        if not self._token:
+            self._token = utils.getToken()
+        token = self._token
+        url = utils.add_signature_for_url(
+            m3u8_url, token, self.timestamp, self.signature
         )
         try:
             with requests.get(
@@ -217,16 +192,11 @@ def download_ts(self, ts_url, name, num_retries):
         """
         下载 .ts 文件
         """
-        token = self.getToken()
-        ts_url = (
-            ts_url.split("\n")[0]
-            + "?Xvideo_Token="
-            + token
-            + "&Xclient_Timestamp="
-            + self.timestamp
-            + "&Xclient_Signature="
-            + self.signature
-            + "&Xclient_Version=v1&Platform=yhkt_user"
+        if not self._token:
+            self._token = utils.getToken()
+        token = self._token
+        ts_url = utils.add_signature_for_url(
+            ts_url.split("\n")[0], token, self.timestamp, self.signature
         )
         try:
             if not os.path.exists(name):

utils.py

@@ -2,14 +2,51 @@
 
 import requests
 import m3u8dl
+import time
+from hashlib import md5
 
 
 headers = {
     "Origin": "https://www.yanhekt.cn",
-    "Authorization": "Bearer 6277e60fa9e86fdcdd2411ce86a54f47",
     "xdomain-client": "web_user",
     "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36 Edg/107.0.1418.26",
 }
+magic = "1tJrMwNq3h0yLgx86Rued2J1tFc"
+
+
+def encryptURL(url):
+    url_list = url.split("/")
+    # "a97f12c055a10ee51d60e441e618bfef"
+    url_list.insert(-1, md5((magic + "_100").encode()).hexdigest())
+    return "/".join(url_list)
+
+
+def getSignature():
+    timestamp = str(int(time.time()))
+    signature = md5((magic + "_v1_" + timestamp).encode()).hexdigest()
+    return timestamp, signature
+
+
+def getToken():
+    req = requests.get(
+        "https://cbiz.yanhekt.cn/v1/auth/video/token?id=0", headers=headers
+    )
+    data = req.json()["data"]
+    return data["token"]
+
+
+def add_signature_for_url(url, token, timestamp, signature):
+    url = (
+        url
+        + "?Xvideo_Token="
+        + token
+        + "&Xclient_Timestamp="
+        + timestamp
+        + "&Xclient_Signature="
+        + signature
+        + "&Xclient_Version=v1&Platform=yhkt_user"
+    )
+    return url
 
 
 def get_course_info(courseID):
@@ -56,6 +93,9 @@ def get_audio_url(video_id):
 
 
 def download_audio(url, path, name):
+    token = getToken()
+    url = add_signature_for_url(url, token, *getSignature())
+    print(url)
     res = requests.get(url, headers=headers)
     with open(f"{path}/{name}.aac", "wb") as f:
         f.write(res.content)

webui_interface.py

@@ -13,6 +13,7 @@
 import threading
 import ctypes
 import multiprocessing
+import webbrowser
 
 app = Flask(__name__, static_folder="webui")
 
@@ -84,7 +85,6 @@ def execute_one_download_task_worker(task_dict, father_queue):
     m3u8dl.M3u8Download(url, output, name, progress_callback=executor_progress_callback)
     if task_dict["download_audio"]:
         audio_url = task_dict["audio_url"]
-        print(f"audio url: {audio_url}")
         if audio_url:
             print("Downloading audio...")
             utils.download_audio(audio_url, output, name)
@@ -225,4 +225,5 @@ def static_files(path):
     multiprocessing.freeze_support()
     t = threading.Thread(target=execute_tasks)
     t.start()
+    webbrowser.open("http://127.0.0.1:5001/")
     app.run(debug=False, host="0.0.0.0", use_reloader=False, port=5001)