三步流程：
① GET [url=https://live.douyin.com/]https://live.douyin.com/[/url] → 拿到 ttwid 等基础 cookie
② GET /webcast/room/web/enter/?web_rid=xx → JSON 中 stream_url.flv_pull_url
③ 把地址的 http:// 换成 https:// → 直接拉流
[Asm] 纯文本查看复制代码
# -*- coding: utf-8 -*-
""" 抖音直播间取流器 —— 纯 HTTP 版（不启动浏览器、不做签名逆向）
==============================================================
实测结论（未登录状态）：
1. 普通 GET 请求即可拿到完整推流地址，服务端未强制校验 a_bogus/X-Bogus；
2. 地址中的 sign / expire 由抖音服务端签发，本地无需也无法计算；
【修改说明】
直播间ID直接写在下面 FIXED_WEB_RID
"""
from __future__ import annotations
import json
import re
import sys
import time
import urllib.parse
import urllib.request
import http.cookiejar
REFERER = "https://live.douyin.com/"
USER_AGENT = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36"
)
# 清晰度优先级（由高到低）
QUALITY_ORDER = ["FULL_HD1", "HD1", "SD1", "SD2"]
# ===================== 在这里填入你的直播间ID =====================
FIXED_WEB_RID = "355655334081"
# ===================================================================
def build_session():
"""建立带 cookie 的会话：先访问首页拿 ttwid"""
cj = http.cookiejar.CookieJar()
op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
op.addheaders = [
("User-Agent", USER_AGENT),
("Accept-Language", "zh-CN,zh;q=0.9"),
("Accept", "text/html,application/xhtml+xml,*/*;q=0.8"),
]
op.open(REFERER, timeout=15).read()
return op, cj
def normalize_rid(target: str) -> str:
s = target.strip()
if re.fullmatch(r"\d+", s):
return s
m = re.search(r"live\.douyin\.com/(\d+)", s)
return m.group(1) if m else s
def get_streams(web_rid: str, op=None):
"""调 enter 接口，返回 (房间信息, {清晰度: https 流地址})"""
if op is None:
op, _ = build_session()
params = {
"aid": "6383", "app_name": "douyin_web", "live_id": "1",
"device_platform": "web", "language": "zh-CN", "enter_from": "web_live",
"cookie_enabled": "true", "screen_width": "1920", "screen_height": "1080",
"browser_language": "zh-CN", "browser_platform": "Win32",
"browser_name": "Chrome", "browser_version": "140.0.0.0",
"web_rid": web_rid,
}
url = "https://live.douyin.com/webcast/room/web/enter/?" + urllib.parse.urlencode(params)
req = urllib.request.Request(url, headers={
"Referer": f"{REFERER}{web_rid}",
"Accept": "application/json, text/plain, */*",
})
body = op.open(req, timeout=20).read().decode("utf-8", "ignore")
j = json.loads(body)
if j.get("status_code") != 0:
raise RuntimeError(f"接口返回异常 status_code={j.get('status_code')}")
rooms = (j.get("data") or {}).get("data") or []
if not rooms:
raise RuntimeError("未取到房间数据：可能房间号错误或未开播")
r0 = rooms[0]
su = r0.get("stream_url") or {}
flv = su.get("flv_pull_url") or {}
# 关键：http -> https
flv = {k: v.replace("http://", "https://", 1) for k, v in flv.items()}
info = {
"room_id": web_rid,
"title": r0.get("title", ""),
"status": r0.get("status_str", ""), # "2" = 直播中
"user_count": r0.get("user_count_str", ""),
"nickname": ((r0.get("owner") or {}).get("nickname") or ""),
"hls": su.get("hls_pull_url_map") or {},
}
return info, flv
def probe(url: str, seconds: int = 3):
"""带 Referer 试拉一小段，确认地址真实可用"""
req = urllib.request.Request(url, headers={"Referer": REFERER, "User-Agent": USER_AGENT})
t0, total, ctype = time.time(), 0, ""
try:
with urllib.request.urlopen(req, timeout=12) as r:
ctype = r.headers.get("Content-Type", "")
while time.time() - t0 < seconds:
b = r.read(65536)
if not b:
break
total += len(b)
except Exception as e:
return False, f"{type(e).__name__}: {e}"
if total == 0:
return False, f"{ctype} 但 0 字节（通常是该档位未推流）"
return True, f"{ctype} | {seconds}s {total/1024:.0f} KB (≈{total*8/seconds/1e6:.2f} Mbps)"
def main() -> int:
# 直接使用代码内写死的直播间ID
target = FIXED_WEB_RID
rid = normalize_rid(target)
op, cj = build_session()
info, flv = get_streams(rid, op)
# ========= 这里你可以自行开关两个选项 =========
OUTPUT_JSON = False # True=输出json，False=普通文本输出
NO_VERIFY = False # True=跳过流地址校验
SELECT_QUALITY = "FULL_HD1"
# ==============================================
if not OUTPUT_JSON:
print(f"房间：{info['title']}")
print(f"主播：{info['nickname']} 在線：{info['user_count']} 狀態：{'直播中' if info['status'] == '2' else info['status']}")
print(f"cookie：{[c.name for c in cj]}")
print(f"\n可用清晰度：{list(flv.keys())}")
result = {"info": info, "streams": flv, "verify": {}}
for q in QUALITY_ORDER:
if q not in flv:
continue
if OUTPUT_JSON:
result["verify"][q] = probe(flv[q]) if not NO_VERIFY else None
continue
ok, msg = (True, "已跳过校验") if NO_VERIFY else probe(flv[q])
print(f"\n[{q}] {'OK ' if ok else 'ERR'} {msg}")
print(f" {flv[q]}")
if OUTPUT_JSON:
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
sys.exit(main())
发表于 2026-9-27 04:40
