166 lines
6.0 KiB
Python
166 lines
6.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
e2e_download.py —— 端到端验收：通过服务的 HTTP 接口真的下一次，并校验产物。
|
|
|
|
流程：playurl -> POST /api/download -> 订阅 SSE 收进度 -> 等完成 -> ffprobe 校验
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import urllib.request
|
|
|
|
BASE = "http://127.0.0.1:8799"
|
|
BVID = sys.argv[1] if len(sys.argv) > 1 else "BV1HVBAYLEAo"
|
|
|
|
ok_count = 0
|
|
fail_count = 0
|
|
|
|
|
|
def check(name, ok, detail=""):
|
|
global ok_count, fail_count
|
|
if ok:
|
|
ok_count += 1
|
|
print(f" [OK] {name}" + (f" — {detail}" if detail else ""))
|
|
else:
|
|
fail_count += 1
|
|
print(f" [XX] {name}" + (f" — {detail}" if detail else ""))
|
|
|
|
|
|
def get(path):
|
|
with urllib.request.urlopen(BASE + path, timeout=60) as r:
|
|
return json.loads(r.read().decode("utf-8"))
|
|
|
|
|
|
def post(path, payload):
|
|
req = urllib.request.Request(
|
|
BASE + path, data=json.dumps(payload).encode("utf-8"),
|
|
headers={"Content-Type": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=60) as r:
|
|
return json.loads(r.read().decode("utf-8"))
|
|
|
|
|
|
print("=" * 62)
|
|
print(f"端到端下载验收: {BVID}")
|
|
print("=" * 62)
|
|
|
|
video = get(f"/api/video?bvid={BVID}")
|
|
cid = video["pages"][0]["cid"]
|
|
print(f"\n视频: {video['title']}")
|
|
print(f"UP主: {video['owner']} 时长: {video['duration']}s")
|
|
|
|
play = get(f"/api/playurl?bvid={BVID}&cid={cid}")
|
|
check("拿到清晰度列表", len(play["qualities"]) > 0,
|
|
" / ".join(q["label"] for q in play["qualities"]))
|
|
quality = play["qualities"][0]["quality"]
|
|
print(f"选择清晰度: {quality} ({play['qualities'][0]['label']})")
|
|
|
|
# 先确认没有正在跑的任务
|
|
job0 = get("/api/job")
|
|
check("初始无进行中的任务", job0.get("state") != "running", str(job0.get("state")))
|
|
|
|
# 订阅 SSE
|
|
events = []
|
|
prog_samples = []
|
|
stop = threading.Event()
|
|
|
|
|
|
def listen():
|
|
try:
|
|
with urllib.request.urlopen(BASE + "/api/events", timeout=600) as r:
|
|
event = None
|
|
while not stop.is_set():
|
|
line = r.readline().decode("utf-8", "replace").strip()
|
|
if line.startswith("event:"):
|
|
event = line.split(":", 1)[1].strip()
|
|
elif line.startswith("data:"):
|
|
data = json.loads(line.split(":", 1)[1].strip())
|
|
events.append((event, data))
|
|
if event == "progress":
|
|
prog_samples.append(data)
|
|
except Exception as exc:
|
|
if not stop.is_set():
|
|
print(" SSE 读取结束:", exc)
|
|
|
|
|
|
t = threading.Thread(target=listen, daemon=True)
|
|
t.start()
|
|
time.sleep(0.8)
|
|
|
|
print("\n--- 开始下载 ---")
|
|
t0 = time.time()
|
|
job = post("/api/download", {"bvid": BVID, "cid": cid, "quality": quality,
|
|
"title": video["title"]})
|
|
check("已受理下载任务", job.get("state") == "running", str(job.get("state")))
|
|
|
|
deadline = time.time() + 300
|
|
last_state = None
|
|
while time.time() < deadline:
|
|
time.sleep(1)
|
|
cur = get("/api/job")
|
|
if cur.get("state") != last_state:
|
|
last_state = cur.get("state")
|
|
print(f" [{time.time() - t0:5.1f}s] state={last_state} stage={cur.get('stage')}")
|
|
if cur.get("state") in ("done", "error", "cancelled"):
|
|
break
|
|
|
|
elapsed = time.time() - t0
|
|
stop.set()
|
|
final = get("/api/job")
|
|
|
|
check("任务正常结束", final.get("state") == "done", str(final.get("state") or final.get("error")))
|
|
check("收到了进度事件", len(prog_samples) > 0, f"{len(prog_samples)} 条")
|
|
check("同时收到 job/stage 事件", len(events) > 0, f"{len(events)} 条")
|
|
|
|
out = final.get("outputPath") or ""
|
|
check("产物文件存在", bool(out) and os.path.exists(out), out)
|
|
if out and os.path.exists(out):
|
|
size = os.path.getsize(out)
|
|
check("产物大小合理", size > 100_000, f"{size / 1024 / 1024:.1f} MB")
|
|
ffprobe = None
|
|
import shutil
|
|
ffprobe = shutil.which("ffprobe") or r"C:\Users\Administrator\miniconda3\Library\bin\ffprobe.exe"
|
|
cp = subprocess.run(
|
|
[ffprobe, "-v", "error", "-show_entries",
|
|
"format=duration,size:stream=codec_type,codec_name,width,height",
|
|
"-of", "json", out],
|
|
capture_output=True, text=True)
|
|
if cp.returncode == 0:
|
|
info = json.loads(cp.stdout)
|
|
streams = info.get("streams", [])
|
|
kinds = {s["codec_type"]: s for s in streams}
|
|
check("产物含视频流", "video" in kinds,
|
|
f"{kinds.get('video', {}).get('codec_name')} "
|
|
f"{kinds.get('video', {}).get('width')}x{kinds.get('video', {}).get('height')}")
|
|
check("产物含音频流", "audio" in kinds, str(kinds.get("audio", {}).get("codec_name")))
|
|
dur = float(info["format"]["duration"])
|
|
check("时长与源基本一致", abs(dur - video["duration"]) < 5, f"{dur:.1f}s vs {video['duration']}s")
|
|
else:
|
|
check("ffprobe 可解析产物", False, cp.stderr.strip()[:120])
|
|
|
|
# 全片解码校验
|
|
ffmpeg = shutil.which("ffmpeg") or r"C:\Users\Administrator\miniconda3\Library\bin\ffmpeg.exe"
|
|
cp2 = subprocess.run([ffmpeg, "-v", "error", "-i", out, "-f", "null", "-"],
|
|
capture_output=True, text=True)
|
|
check("全片解码无错误", cp2.returncode == 0 and not cp2.stderr.strip(),
|
|
cp2.stderr.strip()[:160] or "clean")
|
|
|
|
# 临时文件应已清理
|
|
tmp_left = [f for f in os.listdir(os.path.dirname(out)) if f.endswith(".m4s")]
|
|
check("中间文件已清理", len(tmp_left) == 0, f"残留 {len(tmp_left)} 个" if tmp_left else "无残留")
|
|
|
|
# 并发保护：任务已结束，再来一次应能受理（说明状态机复位）
|
|
print(f"\n耗时 {elapsed:.1f}s")
|
|
print("\n进度采样（每 10 条取 1）:")
|
|
for s in prog_samples[::max(1, len(prog_samples) // 10)][:10]:
|
|
print(f" {s['label']} {s['percent']:5.1f}% {s['downloaded']/1024/1024:6.1f}MB"
|
|
f"/{s['total']/1024/1024:.1f}MB {s['speed']/1024/1024:.2f}MB/s {s['threads']}线程")
|
|
|
|
print("\n" + "=" * 62)
|
|
print(f"结果: {ok_count} 通过 / {fail_count} 失败")
|
|
print("=" * 62)
|
|
sys.exit(1 if fail_count else 0)
|