fix: 支持淘宝直播完整 tbzb 链接解析 & cookies风控解决

main.py

@@ -31,6 +31,7 @@
 from src.proxy import ProxyDetector
 from src.utils import logger
 from src import utils
+from src.taobao_utils import is_taobao_live_url
 from msg_push import (
     dingtalk, xizhi, tg_bot, send_email, bark, ntfy, pushplus
 )
@@ -972,7 +973,7 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
                                 url=record_url, proxy_addr=proxy_address, cookies=youtube_cookie))
                             port_info = asyncio.run(stream.get_stream_url(json_data, record_quality, spec=True))
 
-                    elif record_url.find("tb.cn") > -1:
+                    elif is_taobao_live_url(record_url):
                         platform = '淘宝直播'
                         with semaphore:
                             json_data = asyncio.run(spider.get_taobao_stream_url(
@@ -2037,8 +2038,10 @@ def contains_url(string: str) -> bool:
                     "m.6.cn",
                     'www.lehaitv.com',
                     'h.catshow168.com',
+                    'm.tb.cn',
                     'e.tb.cn',
                     'huodong.m.taobao.com',
+                    'tbzb.taobao.com',
                     '3.cn',
                     'eco.m.jd.com',
                     'www.miguvideo.com',

src/spider.py

@@ -30,6 +30,7 @@
 from .room import get_sec_user_id, get_unique_id, UnsupportedUrlError
 from .http_clients.async_http import async_req
 from .ab_sign import ab_sign
+from .taobao_utils import build_taobao_sign, get_cookie_value, get_taobao_live_id, merge_cookie_header
 
 
 ssl_context = ssl.create_default_context()
@@ -3032,18 +3033,29 @@ async def get_taobao_stream_url(url: str, proxy_addr: OptionalStr = None, cookie
         'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/115.0',
         'Cookie': '',
     }
+    result = {'anchor_name': '', 'is_live': False}
 
     if cookies:
         headers['Cookie'] = cookies
 
     if '_m_h5_tk' not in headers['Cookie']:
         print('Error: Cookies is empty! please input correct cookies')
+        return result
+
+    live_id = get_taobao_live_id(url)
+    if not live_id:
+        redirect_url = await async_req(url, proxy_addr=proxy_addr, headers=headers, redirect_url=True)
+        live_id = get_taobao_live_id(redirect_url)
 
-    live_id = get_params(url, 'id')
     if not live_id:
         html_str = await async_req(url, proxy_addr=proxy_addr, headers=headers)
-        redirect_url = re.findall("var url = '(.*?)';", html_str)[0]
-        live_id = get_params(redirect_url, 'id')
+        redirect_urls = re.findall("var url = '(.*?)';", html_str)
+        if redirect_urls:
+            live_id = get_taobao_live_id(redirect_urls[0])
+
+    if not live_id:
+        print(f'Error: Unable to extract Taobao liveId from url: {url}')
+        return result
 
     params = {
         'jsv': '2.7.0',
@@ -3062,25 +3074,27 @@ async def get_taobao_stream_url(url: str, proxy_addr: OptionalStr = None, cookie
     }
 
     for i in range(2):
-        app_key = '12574478'
-        _m_h5_tk = re.findall('_m_h5_tk=(.*?);', headers['Cookie'])[0]
+        _m_h5_tk = get_cookie_value(headers['Cookie'], '_m_h5_tk')
+        if not _m_h5_tk:
+            print('Error: Unable to read _m_h5_tk from cookies')
+            return result
         t13 = int(time.time() * 1000)
-        pre_sign_str = f'{_m_h5_tk.split("_")[0]}&{t13}&{app_key}&' + params['data']
-        sign = execjs.compile(open(f'{JS_SCRIPT_PATH}/taobao-sign.js').read()).call('sign', pre_sign_str)
+        sign = build_taobao_sign(_m_h5_tk, t13, params['data'])
         params |= {'sign': sign, 't': t13}
         api = f'https://h5api.m.taobao.com/h5/mtop.mediaplatform.live.livedetail/4.0/?{urllib.parse.urlencode(params)}'
         jsonp_str, new_cookie = await async_req(url=api, proxy_addr=proxy_addr, headers=headers, timeout=20,
                                                 return_cookies=True, include_cookies=True)
         json_data = utils.jsonp_to_json(jsonp_str)
+        headers['Cookie'] = merge_cookie_header(headers['Cookie'], new_cookie)
 
         ret_msg = json_data['ret']
         if ret_msg == ['SUCCESS::调用成功']:
             anchor_name = json_data['data']['broadCaster']['accountName']
             result = {"anchor_name": anchor_name, "is_live": False}
             live_status = json_data['data']['streamStatus']
-            if live_status == '1':
+            play_url_list = json_data['data'].get('liveUrlList') or []
+            if str(live_status) in {'0', '1'} or play_url_list:
                 live_title = json_data['data']['title']
-                play_url_list = json_data['data']['liveUrlList']
 
                 def get_sort_key(item):
                     definition_priority = {
@@ -3094,13 +3108,8 @@ def get_sort_key(item):
                 result |= {"is_live": True, "title": live_title, "play_url_list": play_url_list, 'live_id': live_id}
 
             return result
-        else:
-            print(f'Error: Taobao live data fetch failed, {ret_msg[0]}')
 
-        if '_m_h5_tk' in new_cookie and '_m_h5_tk_enc' in new_cookie:
-            headers['Cookie'] = re.sub('_m_h5_tk=(.*?);', new_cookie['_m_h5_tk'], headers['Cookie'])
-            headers['Cookie'] = re.sub('_m_h5_tk_enc=(.*?);', new_cookie['_m_h5_tk_enc'], headers['Cookie'])
-        else:
+        if '_m_h5_tk' not in new_cookie or '_m_h5_tk_enc' not in new_cookie:
             print('Error: Try to update cookie failed, please update the cookies in the configuration file')
 
 

src/taobao_utils.py

@@ -0,0 +1,50 @@
+import hashlib
+import re
+import urllib.parse
+
+
+OptionalStr = str | None
+
+
+def get_params(url: str, params: str) -> OptionalStr:
+    parsed_url = urllib.parse.urlparse(url)
+    query_params = urllib.parse.parse_qs(parsed_url.query)
+
+    if params in query_params:
+        return query_params[params][0]
+
+    return None
+
+
+def is_taobao_live_url(url: str) -> bool:
+    return any(marker in url for marker in ("tb.cn", "tbzb.taobao.com/live"))
+
+
+def get_taobao_live_id(url: str) -> OptionalStr:
+    return get_params(url, "liveId") or get_params(url, "id")
+
+
+def get_cookie_value(cookie_str: str, key: str) -> OptionalStr:
+    match = re.search(rf"(?:^|;\s*){re.escape(key)}=([^;]+)", cookie_str)
+    return match.group(1) if match else None
+
+
+def merge_cookie_header(cookie_str: str, new_cookie: dict | None) -> str:
+    if not new_cookie:
+        return cookie_str
+
+    cookie_dict: dict[str, str] = {}
+    for item in cookie_str.split(";"):
+        item = item.strip()
+        if not item or "=" not in item:
+            continue
+        key, value = item.split("=", 1)
+        cookie_dict[key] = value
+
+    cookie_dict.update(new_cookie)
+    return "; ".join(f"{key}={value}" for key, value in cookie_dict.items())
+
+
+def build_taobao_sign(token: str, t13: int, data: str) -> str:
+    raw = f"{token.split('_', 1)[0]}&{t13}&12574478&{data}"
+    return hashlib.md5(raw.encode("utf-8")).hexdigest()

tests/test_taobao_helpers.py

@@ -0,0 +1,76 @@
+import importlib.util
+from pathlib import Path
+import unittest
+
+
+HELPERS_PATH = Path(__file__).resolve().parents[1] / "src" / "taobao_utils.py"
+SPEC = importlib.util.spec_from_file_location("taobao_utils", HELPERS_PATH)
+taobao_utils = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(taobao_utils)
+
+get_taobao_live_id = taobao_utils.get_taobao_live_id
+is_taobao_live_url = taobao_utils.is_taobao_live_url
+get_cookie_value = taobao_utils.get_cookie_value
+merge_cookie_header = taobao_utils.merge_cookie_header
+build_taobao_sign = taobao_utils.build_taobao_sign
+
+
+class TestTaobaoHelpers(unittest.TestCase):
+    def test_detects_supported_taobao_live_urls(self):
+        urls = (
+            "https://m.tb.cn/h.TWp0HTd",
+            "https://tbzb.taobao.com/live?liveId=557156521789",
+        )
+
+        for url in urls:
+            with self.subTest(url=url):
+                self.assertTrue(is_taobao_live_url(url))
+
+    def test_extracts_live_id_from_tbzb_live_url(self):
+        url = "https://tbzb.taobao.com/live?liveId=557156521789"
+        self.assertEqual(get_taobao_live_id(url), "557156521789")
+
+    def test_extracts_live_id_from_redirect_style_url(self):
+        url = "https://huodong.m.taobao.com/act/talent/live.html?id=557156521789"
+        self.assertEqual(get_taobao_live_id(url), "557156521789")
+
+    def test_merges_refreshed_taobao_cookie_values(self):
+        cookie = "_m_h5_tk=old_token; _m_h5_tk_enc=old_enc; cna=keep_me"
+        merged = merge_cookie_header(
+            cookie,
+            {"_m_h5_tk": "new_token", "_m_h5_tk_enc": "new_enc", "isg": "new_isg"},
+        )
+        self.assertEqual(get_cookie_value(merged, "_m_h5_tk"), "new_token")
+        self.assertEqual(get_cookie_value(merged, "_m_h5_tk_enc"), "new_enc")
+        self.assertEqual(get_cookie_value(merged, "cna"), "keep_me")
+        self.assertEqual(get_cookie_value(merged, "isg"), "new_isg")
+
+    def test_builds_taobao_sign_with_token_prefix(self):
+        token = "sampletoken123_1700000000000"
+        data = '{"liveId":"557156521789","creatorId":null}'
+        self.assertEqual(
+            build_taobao_sign(token, 1234567890123, data),
+            "2d678931295fc39ea82599c50e4a6f4e",
+        )
+
+    def test_parses_real_tbzb_url_and_cookie_sample(self):
+        url = "https://tbzb.taobao.com/live?liveId=557156521789"
+        cookies = "_m_h5_tk=sampletoken123_1700000000000;_m_h5_tk_enc=sample_enc_token"
+
+        live_id = get_taobao_live_id(url)
+        token = get_cookie_value(cookies, "_m_h5_tk")
+        token_enc = get_cookie_value(cookies, "_m_h5_tk_enc")
+        data = f'{{"liveId":"{live_id}","creatorId":null}}'
+
+        self.assertTrue(is_taobao_live_url(url))
+        self.assertEqual(live_id, "557156521789")
+        self.assertEqual(token, "sampletoken123_1700000000000")
+        self.assertEqual(token_enc, "sample_enc_token")
+        self.assertEqual(
+            build_taobao_sign(token, 1234567890123, data),
+            "2d678931295fc39ea82599c50e4a6f4e",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_taobao_stream.py

@@ -0,0 +1,45 @@
+import asyncio
+import os
+import unittest
+
+from src import spider, stream
+
+
+class TestTaobaoIntegration(unittest.TestCase):
+    def test_fetches_real_taobao_stream_urls(self):
+        url = "https://tbzb.taobao.com/live?liveId=557156521789"
+        cookies = "_m_h5_tk=2007b3c0ac705a948417b47ea0f92638_1773349778735;_m_h5_tk_enc=8b6eb1331f579522765e81bfe5289990"
+
+        if not url or not cookies:
+            self.skipTest(
+                "Set TAOBAO_TEST_URL and TAOBAO_TEST_COOKIES to run this integration test"
+            )
+
+        json_data = asyncio.run(spider.get_taobao_stream_url(url=url, cookies=cookies))
+        self.assertTrue(
+            json_data.get("anchor_name"), "Expected Taobao anchor_name to be resolved"
+        )
+
+        if not json_data.get("is_live"):
+            self.skipTest(
+                f"Taobao live room is offline: {json_data.get('anchor_name')}"
+            )
+
+        port_info = asyncio.run(
+            stream.get_stream_url(
+                json_data,
+                "OD",
+                url_type="all",
+                hls_extra_key="hlsUrl",
+                flv_extra_key="flvUrl",
+            )
+        )
+
+        print(f"m3u8_url={port_info.get('m3u8_url')}")
+        print(f"flv_url={port_info.get('flv_url')}")
+
+        self.assertTrue(port_info.get("m3u8_url") or port_info.get("flv_url"))
+
+
+if __name__ == "__main__":
+    unittest.main()