58 lines
2.3 KiB
Python
58 lines
2.3 KiB
Python
from __future__ import annotations
|
||
|
||
import argparse
|
||
import threading
|
||
|
||
from m3u_tools import ProbeConfig, run_probe_file
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(
|
||
description="M3U 链接可用性与速度检测工具",
|
||
)
|
||
parser.add_argument("-i", "--input", default="test.txt", help="输入文件，格式为 频道,URL")
|
||
parser.add_argument("-o", "--output", default="pass_test.txt", help="输出通过列表")
|
||
parser.add_argument("--workers", type=int, default=None, help="并发线程数")
|
||
parser.add_argument("--per-host", type=int, default=3, help="单主机最大并发")
|
||
parser.add_argument("--timeout", type=float, default=15.0, help="单次 HTTP 请求超时秒数")
|
||
parser.add_argument("--min-speed", type=float, default=400.0, help="最低平均速度，单位 KB/s")
|
||
parser.add_argument("--sample-bytes", type=int, default=2048, help="每个分片采样大小，单位 KB")
|
||
parser.add_argument("--sample-segments", type=int, default=3, help="每条链接抽测分片数量")
|
||
parser.add_argument("--disable-ffmpeg-fallback", action="store_true", help="禁用 FFmpeg 回退检测")
|
||
return parser
|
||
|
||
|
||
def main() -> int:
|
||
parser = build_parser()
|
||
args = parser.parse_args()
|
||
|
||
config = ProbeConfig(
|
||
input_file=args.input,
|
||
output_file=args.output,
|
||
max_workers=args.workers or ProbeConfig().max_workers,
|
||
per_host_limit=max(1, args.per_host),
|
||
request_timeout=max(1.0, args.timeout),
|
||
min_speed_kbps=max(1.0, args.min_speed),
|
||
sample_bytes=max(64, args.sample_bytes) * 1024,
|
||
sample_segments=max(1, args.sample_segments),
|
||
ffmpeg_fallback=not args.disable_ffmpeg_fallback,
|
||
)
|
||
|
||
def on_event(payload: dict) -> None:
|
||
kind = payload.get("kind")
|
||
if kind == "log":
|
||
line = payload.get("line")
|
||
if line:
|
||
print(line)
|
||
|
||
summary = run_probe_file(config, event_callback=on_event, stop_event=threading.Event())
|
||
print(
|
||
f"\n完成：通过 {summary.passed}/{summary.total}，失败 {summary.failed}，"
|
||
f"输出 {summary.output_file}，报告 {summary.csv_report}"
|
||
)
|
||
return 0 if summary.failed == 0 else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|