本帖最后由 cattie 于 2026-9-22 14:44 编辑
import re
import json
from pprint import pprint
from DrissionPage import ChromiumOptions,ChromiumPage
import subprocess
import time
def download_m3u8(url: str, output: str, concurrent: int = 32):
cmd = [
"yt-dlp",
url,
"--concurrent-fragments", str(concurrent),
"-o", output,
# 你原来能跑通的其他参数照抄就行
# "--add-header", "Referer:xxx",
# "--cookies", "cookies.txt",
]
print("执行:", " ".join(cmd))
result = subprocess.run(
cmd,
capture_output=False, # 实时看进度,和终端一样
text=True,
)
if result.returncode != 0:
raise RuntimeError(f"yt-dlp 退出码 {result.returncode}")
def download():
url = 'https://appn1j7j0hh4155.pc.xiaoe-tech.com/p/t_pc/course_pc_detail/camp_pro/course_31MGlJyQ3IftMumeGaJoGDGCUBA'
dp = ChromiumPage()
dp.listen.start('resource_catalog_list.get/1.0.0')
dp.get(url)
charpter_ids = []
packets = []
start = time.time()
# 持续收集,直到超时或你觉得够了
while time.time() - start < 15:
try:
resp = dp.listen.wait(timeout=2)
if resp:
packets.append(resp)
print(f"捕获到第 {len(packets)} 个 1.0.0 请求")
except:
break
# 取最后一个
if packets:
last_resp = packets[-1]
print("最后一个 1.0.0 请求:")
print(last_resp.url)
print(last_resp.response.body)
body = last_resp.response.body
# 小鹅通常在这里
for item in body.get('data', {}).get('list', []):
cid = item.get('chapter_id')
if cid:
charpter_ids.append(cid)
else:
print("没有捕获到 1.0.0 请求")
print(charpter_ids)
v_ids = [x for x in charpter_ids if re.match(r'^v_', x)]
# print(v_ids)
url_s = []
qian = "https://appn1j7j0hh4155.pc.xiaoe-tech.com/p/t_pc/course_pc_detail/video/"
for lhou in v_ids:
quan = qian + lhou
if quan:
url_s.append(quan)
print('opening:', quan)
dp.listen.start('m3u8')
dp.get(quan)
try:
pkt = dp.listen.wait(timeout=10)
# header = pkt.response.headers
m3u8_url = pkt.url
print('m3u8 url =>', m3u8_url)
#使用控制台命令下载
download_m3u8(
m3u8_url,
lhou+".mp4",
concurrent=32,
)
except Exception as e:
print('timeout or error:', e)
dp.listen.stop()
if __name__ == '__main__':
download()
|