920 lines
34 KiB
Python
920 lines
34 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import json
|
|
import os
|
|
import random
|
|
import re
|
|
import ssl
|
|
import subprocess
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
import threading
|
|
import time
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Callable, Iterable
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urljoin, urlparse
|
|
from urllib.request import HTTPSHandler, Request, build_opener
|
|
|
|
|
|
DEFAULT_USER_AGENTS = (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:126.0) Gecko/20100101 Firefox/126.0",
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.2478.67",
|
|
)
|
|
TEXT_ENCODINGS = ("utf-8", "utf-8-sig", "gb18030", "gbk", "latin-1")
|
|
VIDEO_SEGMENT_EXTENSIONS = (
|
|
".ts",
|
|
".m4s",
|
|
".mp4",
|
|
".cmfv",
|
|
".aac",
|
|
".mp3",
|
|
".vtt",
|
|
)
|
|
VIDEO_FILE_EXTENSIONS = (".mp4", ".mkv", ".avi", ".mov", ".flv", ".wmv", ".m4v")
|
|
|
|
|
|
def natural_sort_key(value: str) -> list[object]:
|
|
return [int(token) if token.isdigit() else token.lower() for token in re.split(r"(\d+)", value)]
|
|
|
|
|
|
def discover_ffmpeg(preferred_path: str | Path | None = None, executable: str = "ffmpeg.exe") -> str:
|
|
if preferred_path:
|
|
candidate = Path(preferred_path)
|
|
if candidate.exists():
|
|
return str(candidate)
|
|
|
|
env_value = os.environ.get("FFMPEG_PATH")
|
|
if env_value and Path(env_value).exists():
|
|
return env_value
|
|
|
|
default_candidates = [
|
|
Path(r"C:\ffmpeg\bin") / executable,
|
|
Path.cwd() / executable,
|
|
]
|
|
for candidate in default_candidates:
|
|
if candidate.exists():
|
|
return str(candidate)
|
|
|
|
return executable
|
|
|
|
|
|
def discover_ffprobe(preferred_path: str | Path | None = None) -> str:
|
|
return discover_ffmpeg(preferred_path=preferred_path, executable="ffprobe.exe")
|
|
|
|
|
|
def slugify_filename(name: str) -> str:
|
|
cleaned = re.sub(r"[\\/:*?\"<>|]+", "_", name.strip())
|
|
cleaned = re.sub(r"\s+", " ", cleaned).strip()
|
|
return cleaned or "untitled"
|
|
|
|
|
|
def read_text_file(path: str | Path) -> list[str]:
|
|
file_path = Path(path)
|
|
for encoding in TEXT_ENCODINGS:
|
|
try:
|
|
with file_path.open("r", encoding=encoding) as handle:
|
|
return handle.readlines()
|
|
except UnicodeDecodeError:
|
|
continue
|
|
with file_path.open("r", encoding="utf-8", errors="replace") as handle:
|
|
return handle.readlines()
|
|
|
|
|
|
def ensure_directory(path: str | Path) -> Path:
|
|
directory = Path(path)
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
return directory
|
|
|
|
|
|
def subprocess_no_window_kwargs() -> dict[str, object]:
|
|
if os.name != "nt":
|
|
return {}
|
|
|
|
startupinfo = subprocess.STARTUPINFO()
|
|
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
|
startupinfo.wShowWindow = getattr(subprocess, "SW_HIDE", 0)
|
|
|
|
kwargs: dict[str, object] = {
|
|
"startupinfo": startupinfo,
|
|
}
|
|
creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
|
if creationflags:
|
|
kwargs["creationflags"] = creationflags
|
|
return kwargs
|
|
|
|
|
|
class OperationLogger:
|
|
def __init__(self, log_file: str | Path, callback: Callable[[str, str, str], None] | None = None) -> None:
|
|
self.log_file = Path(log_file)
|
|
self.callback = callback
|
|
self.lock = threading.Lock()
|
|
ensure_directory(self.log_file.parent)
|
|
|
|
def log(self, level: str, message: str) -> None:
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
line = f"[{timestamp}] [{level.upper()}] {message}"
|
|
with self.lock:
|
|
with self.log_file.open("a", encoding="utf-8") as handle:
|
|
handle.write(line + "\n")
|
|
if self.callback:
|
|
self.callback(level.upper(), message, line)
|
|
|
|
def info(self, message: str) -> None:
|
|
self.log("INFO", message)
|
|
|
|
def warning(self, message: str) -> None:
|
|
self.log("WARNING", message)
|
|
|
|
def error(self, message: str) -> None:
|
|
self.log("ERROR", message)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ProbeConfig:
|
|
input_file: str = "test.txt"
|
|
output_file: str = "pass_test.txt"
|
|
report_dir: str = "reports"
|
|
log_dir: str = "logs"
|
|
ffmpeg_path: str = field(default_factory=discover_ffmpeg)
|
|
max_workers: int = field(default_factory=lambda: min(96, max(24, (os.cpu_count() or 8) * 2)))
|
|
per_host_limit: int = 3
|
|
request_timeout: float = 15.0
|
|
ffmpeg_timeout: float = 18.0
|
|
ffmpeg_probe_seconds: float = 5.0
|
|
sample_bytes: int = 2048 * 1024
|
|
sample_segments: int = 3
|
|
min_successful_segments: int = 2
|
|
min_total_bytes: int = 1024 * 1024
|
|
min_speed_kbps: float = 400.0
|
|
verify_tls: bool = False
|
|
write_header: bool = True
|
|
ffmpeg_fallback: bool = True
|
|
randomize_user_agent: bool = True
|
|
user_agents: tuple[str, ...] = DEFAULT_USER_AGENTS
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ProbeTarget:
|
|
index: int
|
|
channel: str
|
|
url: str
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class SegmentSample:
|
|
url: str
|
|
ok: bool
|
|
bytes_read: int
|
|
elapsed_seconds: float
|
|
ttfb_ms: float
|
|
speed_kbps: float
|
|
status_code: int | None
|
|
error: str = ""
|
|
transfer_speed_kbps: float = 0.0
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ProbeResult:
|
|
index: int
|
|
channel: str
|
|
url: str
|
|
ok: bool
|
|
reason: str
|
|
source_type: str
|
|
final_playlist_url: str = ""
|
|
playlist_ms: float = 0.0
|
|
segment_count: int = 0
|
|
sample_count: int = 0
|
|
successful_samples: int = 0
|
|
total_bytes: int = 0
|
|
avg_speed_kbps: float = 0.0
|
|
peak_speed_kbps: float = 0.0
|
|
avg_ttfb_ms: float = 0.0
|
|
elapsed_seconds: float = 0.0
|
|
http_statuses: str = ""
|
|
error_detail: str = ""
|
|
avg_transfer_speed_kbps: float = 0.0
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ProbeSummary:
|
|
total: int
|
|
passed: int
|
|
failed: int
|
|
output_file: Path
|
|
csv_report: Path
|
|
json_report: Path
|
|
log_file: Path
|
|
elapsed_seconds: float
|
|
results: list[ProbeResult]
|
|
|
|
|
|
def parse_channel_file(path: str | Path) -> list[ProbeTarget]:
|
|
lines = read_text_file(path)
|
|
targets: list[ProbeTarget] = []
|
|
for index, raw_line in enumerate(lines):
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("#") or "," not in line:
|
|
continue
|
|
channel, url = map(str.strip, line.split(",", 1))
|
|
if channel and url:
|
|
targets.append(ProbeTarget(index=len(targets), channel=channel, url=url))
|
|
return targets
|
|
|
|
|
|
def build_ssl_context(verify_tls: bool) -> ssl.SSLContext:
|
|
if verify_tls:
|
|
return ssl.create_default_context()
|
|
return ssl._create_unverified_context()
|
|
|
|
|
|
class M3UProber:
|
|
def __init__(
|
|
self,
|
|
config: ProbeConfig,
|
|
logger: OperationLogger | None = None,
|
|
event_callback: Callable[[dict], None] | None = None,
|
|
stop_event: threading.Event | None = None,
|
|
) -> None:
|
|
self.config = config
|
|
self.logger = logger
|
|
self.event_callback = event_callback
|
|
self.stop_event = stop_event or threading.Event()
|
|
self.ssl_context = build_ssl_context(config.verify_tls)
|
|
self.host_lock = threading.Lock()
|
|
self.host_limits: dict[str, threading.Semaphore] = {}
|
|
|
|
def emit(self, payload: dict) -> None:
|
|
if self.event_callback:
|
|
self.event_callback(payload)
|
|
|
|
def log(self, level: str, message: str) -> None:
|
|
if self.logger:
|
|
getattr(self.logger, level.lower(), self.logger.info)(message)
|
|
self.emit({"kind": "log", "level": level.upper(), "message": message})
|
|
|
|
def get_host_limiter(self, url: str) -> threading.Semaphore:
|
|
hostname = urlparse(url).netloc.lower() or "unknown"
|
|
with self.host_lock:
|
|
if hostname not in self.host_limits:
|
|
self.host_limits[hostname] = threading.Semaphore(self.config.per_host_limit)
|
|
return self.host_limits[hostname]
|
|
|
|
def make_headers(self) -> dict[str, str]:
|
|
if self.config.randomize_user_agent:
|
|
user_agent = random.choice(self.config.user_agents)
|
|
else:
|
|
user_agent = self.config.user_agents[0]
|
|
return {
|
|
"User-Agent": user_agent,
|
|
"Accept": "*/*",
|
|
}
|
|
|
|
def open_url(self, url: str, timeout: float, headers: dict[str, str] | None = None):
|
|
request = Request(url, headers=headers or self.make_headers())
|
|
opener = build_opener(HTTPSHandler(context=self.ssl_context))
|
|
return opener.open(request, timeout=timeout)
|
|
|
|
def fetch_text(self, url: str) -> tuple[str, float]:
|
|
started = time.perf_counter()
|
|
with self.open_url(url, timeout=self.config.request_timeout) as response:
|
|
payload = response.read(512 * 1024)
|
|
elapsed_ms = (time.perf_counter() - started) * 1000
|
|
text = payload.decode("utf-8", errors="replace")
|
|
return text, elapsed_ms
|
|
|
|
def fetch_playlist(self, url: str) -> tuple[str, str, float]:
|
|
text, elapsed_ms = self.fetch_text(url)
|
|
if "#EXTM3U" not in text and "#EXTINF" not in text and ".m3u8" not in url.lower():
|
|
raise ValueError("响应内容不是有效的 m3u8 播放列表")
|
|
return url, text, elapsed_ms
|
|
|
|
def parse_variant_playlists(self, playlist_url: str, playlist_text: str) -> list[tuple[int, str]]:
|
|
lines = [line.strip() for line in playlist_text.splitlines()]
|
|
variants: list[tuple[int, str]] = []
|
|
for index, line in enumerate(lines):
|
|
if not line.startswith("#EXT-X-STREAM-INF"):
|
|
continue
|
|
bandwidth_match = re.search(r"BANDWIDTH=(\d+)", line)
|
|
bandwidth = int(bandwidth_match.group(1)) if bandwidth_match else 0
|
|
next_index = index + 1
|
|
while next_index < len(lines):
|
|
candidate = lines[next_index]
|
|
if candidate and not candidate.startswith("#"):
|
|
variants.append((bandwidth, urljoin(playlist_url, candidate)))
|
|
break
|
|
next_index += 1
|
|
variants.sort(key=lambda item: item[0], reverse=True)
|
|
return variants
|
|
|
|
def parse_media_segments(self, playlist_url: str, playlist_text: str) -> tuple[list[str], bool]:
|
|
segments: list[str] = []
|
|
is_live = "#EXT-X-ENDLIST" not in playlist_text
|
|
for line in playlist_text.splitlines():
|
|
candidate = line.strip()
|
|
if not candidate or candidate.startswith("#"):
|
|
continue
|
|
if candidate.lower().endswith(".m3u8"):
|
|
continue
|
|
if "://" in candidate or candidate.startswith("/"):
|
|
segments.append(urljoin(playlist_url, candidate))
|
|
continue
|
|
if any(ext in candidate.lower() for ext in VIDEO_SEGMENT_EXTENSIONS):
|
|
segments.append(urljoin(playlist_url, candidate))
|
|
continue
|
|
segments.append(urljoin(playlist_url, candidate))
|
|
return segments, is_live
|
|
|
|
def select_sample_urls(self, segments: list[str], is_live: bool) -> list[str]:
|
|
if not segments:
|
|
return []
|
|
|
|
requested = max(1, self.config.sample_segments)
|
|
if len(segments) <= requested:
|
|
return segments
|
|
|
|
if is_live:
|
|
return segments[-requested:]
|
|
|
|
indices = {
|
|
round(position * (len(segments) - 1) / max(requested - 1, 1))
|
|
for position in range(requested)
|
|
}
|
|
return [segments[index] for index in sorted(indices)]
|
|
|
|
def probe_segment(self, url: str) -> SegmentSample:
|
|
limiter = self.get_host_limiter(url)
|
|
with limiter:
|
|
headers = self.make_headers()
|
|
headers["Range"] = f"bytes=0-{self.config.sample_bytes - 1}"
|
|
request = Request(url, headers=headers)
|
|
opener = build_opener(HTTPSHandler(context=self.ssl_context))
|
|
started = time.perf_counter()
|
|
try:
|
|
with opener.open(request, timeout=self.config.request_timeout) as response:
|
|
ttfb_ms = (time.perf_counter() - started) * 1000
|
|
status_code = getattr(response, "status", None)
|
|
transfer_started = time.perf_counter()
|
|
bytes_read = 0
|
|
while bytes_read < self.config.sample_bytes:
|
|
chunk = response.read(min(64 * 1024, self.config.sample_bytes - bytes_read))
|
|
if not chunk:
|
|
break
|
|
bytes_read += len(chunk)
|
|
elapsed = max(time.perf_counter() - started, 0.001)
|
|
transfer_elapsed = max(time.perf_counter() - transfer_started, 0.001)
|
|
speed_kbps = bytes_read / 1024 / elapsed
|
|
transfer_speed_kbps = bytes_read / 1024 / transfer_elapsed
|
|
return SegmentSample(
|
|
url=url,
|
|
ok=bytes_read > 0,
|
|
bytes_read=bytes_read,
|
|
elapsed_seconds=elapsed,
|
|
ttfb_ms=ttfb_ms,
|
|
speed_kbps=speed_kbps,
|
|
transfer_speed_kbps=transfer_speed_kbps,
|
|
status_code=status_code,
|
|
)
|
|
except HTTPError as error:
|
|
return SegmentSample(
|
|
url=url,
|
|
ok=False,
|
|
bytes_read=0,
|
|
elapsed_seconds=max(time.perf_counter() - started, 0.001),
|
|
ttfb_ms=0.0,
|
|
speed_kbps=0.0,
|
|
transfer_speed_kbps=0.0,
|
|
status_code=error.code,
|
|
error=f"HTTP {error.code}",
|
|
)
|
|
except URLError as error:
|
|
return SegmentSample(
|
|
url=url,
|
|
ok=False,
|
|
bytes_read=0,
|
|
elapsed_seconds=max(time.perf_counter() - started, 0.001),
|
|
ttfb_ms=0.0,
|
|
speed_kbps=0.0,
|
|
transfer_speed_kbps=0.0,
|
|
status_code=None,
|
|
error=f"网络错误: {error.reason}",
|
|
)
|
|
except Exception as error: # pragma: no cover - network edge cases
|
|
return SegmentSample(
|
|
url=url,
|
|
ok=False,
|
|
bytes_read=0,
|
|
elapsed_seconds=max(time.perf_counter() - started, 0.001),
|
|
ttfb_ms=0.0,
|
|
speed_kbps=0.0,
|
|
transfer_speed_kbps=0.0,
|
|
status_code=None,
|
|
error=str(error),
|
|
)
|
|
|
|
def ffmpeg_probe(self, target: ProbeTarget) -> ProbeResult:
|
|
started = time.perf_counter()
|
|
command = [
|
|
self.config.ffmpeg_path,
|
|
"-hide_banner",
|
|
"-loglevel",
|
|
"error",
|
|
"-rw_timeout",
|
|
str(int(self.config.request_timeout * 1_000_000)),
|
|
"-user_agent",
|
|
random.choice(self.config.user_agents),
|
|
"-i",
|
|
target.url,
|
|
"-t",
|
|
str(self.config.ffmpeg_probe_seconds),
|
|
"-map",
|
|
"0:v:0?",
|
|
"-map",
|
|
"0:a:0?",
|
|
"-c",
|
|
"copy",
|
|
"-f",
|
|
"mpegts",
|
|
"pipe:1",
|
|
]
|
|
|
|
try:
|
|
process = subprocess.Popen(
|
|
command,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
stdin=subprocess.DEVNULL,
|
|
**subprocess_no_window_kwargs(),
|
|
)
|
|
stdout, stderr = process.communicate(timeout=self.config.ffmpeg_timeout)
|
|
elapsed = max(time.perf_counter() - started, 0.001)
|
|
total_bytes = len(stdout)
|
|
speed_kbps = total_bytes / 1024 / elapsed
|
|
|
|
if process.returncode == 0 and total_bytes >= self.config.min_total_bytes:
|
|
return ProbeResult(
|
|
index=target.index,
|
|
channel=target.channel,
|
|
url=target.url,
|
|
ok=speed_kbps >= self.config.min_speed_kbps,
|
|
reason=(
|
|
f"FFmpeg回退成功 {speed_kbps:.1f} KB/s"
|
|
if speed_kbps >= self.config.min_speed_kbps
|
|
else f"FFmpeg回退速度不足 {speed_kbps:.1f} KB/s"
|
|
),
|
|
source_type="ffmpeg",
|
|
total_bytes=total_bytes,
|
|
avg_speed_kbps=speed_kbps,
|
|
peak_speed_kbps=speed_kbps,
|
|
elapsed_seconds=elapsed,
|
|
error_detail=stderr.decode("utf-8", errors="replace").strip(),
|
|
)
|
|
|
|
reason = "FFmpeg回退不可用"
|
|
if total_bytes and speed_kbps < self.config.min_speed_kbps:
|
|
reason = f"FFmpeg回退速度不足 {speed_kbps:.1f} KB/s"
|
|
|
|
return ProbeResult(
|
|
index=target.index,
|
|
channel=target.channel,
|
|
url=target.url,
|
|
ok=False,
|
|
reason=reason,
|
|
source_type="ffmpeg",
|
|
total_bytes=total_bytes,
|
|
avg_speed_kbps=speed_kbps,
|
|
peak_speed_kbps=speed_kbps,
|
|
elapsed_seconds=elapsed,
|
|
error_detail=stderr.decode("utf-8", errors="replace").strip(),
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait()
|
|
return ProbeResult(
|
|
index=target.index,
|
|
channel=target.channel,
|
|
url=target.url,
|
|
ok=False,
|
|
reason="FFmpeg回退超时",
|
|
source_type="ffmpeg",
|
|
elapsed_seconds=time.perf_counter() - started,
|
|
)
|
|
except FileNotFoundError:
|
|
return ProbeResult(
|
|
index=target.index,
|
|
channel=target.channel,
|
|
url=target.url,
|
|
ok=False,
|
|
reason="FFmpeg 不存在",
|
|
source_type="ffmpeg",
|
|
)
|
|
except Exception as error: # pragma: no cover - subprocess edge cases
|
|
return ProbeResult(
|
|
index=target.index,
|
|
channel=target.channel,
|
|
url=target.url,
|
|
ok=False,
|
|
reason="FFmpeg回退异常",
|
|
source_type="ffmpeg",
|
|
error_detail=str(error),
|
|
)
|
|
|
|
def _fetch_playlist_throttled(self, url: str) -> tuple[str, str, float]:
|
|
limiter = self.get_host_limiter(url)
|
|
with limiter:
|
|
return self.fetch_playlist(url)
|
|
|
|
def probe_target(self, target: ProbeTarget) -> ProbeResult:
|
|
if self.stop_event.is_set():
|
|
return ProbeResult(
|
|
index=target.index,
|
|
channel=target.channel,
|
|
url=target.url,
|
|
ok=False,
|
|
reason="任务已停止",
|
|
source_type="stopped",
|
|
)
|
|
|
|
started = time.perf_counter()
|
|
|
|
try:
|
|
playlist_url, playlist_text, playlist_ms = self._fetch_playlist_throttled(target.url)
|
|
source_type = "media"
|
|
|
|
variants = self.parse_variant_playlists(playlist_url, playlist_text)
|
|
if variants:
|
|
source_type = "master"
|
|
variant_error: Exception | None = None
|
|
for _, variant_url in variants[:5]:
|
|
try:
|
|
playlist_url, playlist_text, variant_ms = self._fetch_playlist_throttled(variant_url)
|
|
playlist_ms += variant_ms
|
|
break
|
|
except Exception as error: # pragma: no cover - variant failover
|
|
variant_error = error
|
|
else:
|
|
raise variant_error or ValueError("没有可用的子播放列表")
|
|
|
|
segments, is_live = self.parse_media_segments(playlist_url, playlist_text)
|
|
if not segments:
|
|
raise ValueError("播放列表里没有可检测分片")
|
|
|
|
sample_urls = self.select_sample_urls(segments, is_live)
|
|
max_sample_workers = min(len(sample_urls), max(1, self.config.per_host_limit))
|
|
with ThreadPoolExecutor(max_workers=max_sample_workers) as sample_executor:
|
|
sample_futures = [sample_executor.submit(self.probe_segment, url) for url in sample_urls]
|
|
samples = [future.result() for future in as_completed(sample_futures)]
|
|
samples.sort(key=lambda s: sample_urls.index(s.url) if s.url in sample_urls else 999)
|
|
|
|
successful = [sample for sample in samples if sample.ok]
|
|
total_bytes = sum(sample.bytes_read for sample in successful)
|
|
elapsed_seconds = max(time.perf_counter() - started, 0.001)
|
|
|
|
avg_speed = sum(sample.speed_kbps for sample in successful) / max(len(successful), 1)
|
|
avg_transfer_speed = sum(sample.transfer_speed_kbps for sample in successful) / max(len(successful), 1)
|
|
peak_speed = max((sample.speed_kbps for sample in successful), default=0.0)
|
|
peak_transfer_speed = max((sample.transfer_speed_kbps for sample in successful), default=0.0)
|
|
avg_ttfb = sum(sample.ttfb_ms for sample in successful) / max(len(successful), 1)
|
|
statuses = ",".join(
|
|
str(sample.status_code) for sample in samples if sample.status_code is not None
|
|
)
|
|
|
|
def _fail(reason: str) -> ProbeResult:
|
|
return ProbeResult(
|
|
index=target.index,
|
|
channel=target.channel,
|
|
url=target.url,
|
|
ok=False,
|
|
reason=reason,
|
|
source_type=source_type,
|
|
final_playlist_url=playlist_url,
|
|
playlist_ms=playlist_ms,
|
|
segment_count=len(segments),
|
|
sample_count=len(samples),
|
|
successful_samples=len(successful),
|
|
total_bytes=total_bytes,
|
|
avg_speed_kbps=avg_speed,
|
|
peak_speed_kbps=peak_speed,
|
|
avg_transfer_speed_kbps=avg_transfer_speed,
|
|
avg_ttfb_ms=avg_ttfb,
|
|
elapsed_seconds=elapsed_seconds,
|
|
http_statuses=statuses,
|
|
error_detail=" | ".join(sample.error for sample in samples if sample.error),
|
|
)
|
|
|
|
def _pass(reason: str) -> ProbeResult:
|
|
return ProbeResult(
|
|
index=target.index,
|
|
channel=target.channel,
|
|
url=target.url,
|
|
ok=True,
|
|
reason=reason,
|
|
source_type=source_type,
|
|
final_playlist_url=playlist_url,
|
|
playlist_ms=playlist_ms,
|
|
segment_count=len(segments),
|
|
sample_count=len(samples),
|
|
successful_samples=len(successful),
|
|
total_bytes=total_bytes,
|
|
avg_speed_kbps=avg_speed,
|
|
peak_speed_kbps=peak_speed,
|
|
avg_transfer_speed_kbps=avg_transfer_speed,
|
|
avg_ttfb_ms=avg_ttfb,
|
|
elapsed_seconds=elapsed_seconds,
|
|
http_statuses=statuses,
|
|
error_detail=" | ".join(sample.error for sample in samples if sample.error),
|
|
)
|
|
|
|
if len(successful) < min(self.config.min_successful_segments, len(samples)):
|
|
return _fail(f"分片采样失败 {len(successful)}/{len(samples)}")
|
|
|
|
if total_bytes < self.config.min_total_bytes:
|
|
return _fail(f"有效数据不足 {total_bytes // 1024} KB")
|
|
|
|
min_speed = self.config.min_speed_kbps
|
|
# 判定顺序：传输速度优先 > 含连接速度 > 峰值速度(1.3倍冗余)
|
|
if avg_transfer_speed >= min_speed:
|
|
return _pass(f"可用 传输{avg_transfer_speed:.0f} KB/s")
|
|
if avg_speed >= min_speed:
|
|
return _pass(f"可用 {avg_speed:.0f} KB/s")
|
|
if peak_transfer_speed >= min_speed * 1.3:
|
|
return _pass(f"峰值通过 传输峰值{peak_transfer_speed:.0f} KB/s")
|
|
if peak_speed >= min_speed * 1.3:
|
|
return _pass(f"峰值通过 {peak_speed:.0f} KB/s")
|
|
|
|
return _fail(f"速度不足 传输{avg_transfer_speed:.0f} 总{avg_speed:.0f} KB/s")
|
|
except Exception as error:
|
|
if self.config.ffmpeg_fallback:
|
|
fallback = self.ffmpeg_probe(target)
|
|
if fallback.error_detail:
|
|
fallback.error_detail = f"{error} | {fallback.error_detail}"
|
|
else:
|
|
fallback.error_detail = str(error)
|
|
return fallback
|
|
|
|
return ProbeResult(
|
|
index=target.index,
|
|
channel=target.channel,
|
|
url=target.url,
|
|
ok=False,
|
|
reason="检测异常",
|
|
source_type="error",
|
|
elapsed_seconds=max(time.perf_counter() - started, 0.001),
|
|
error_detail=str(error),
|
|
)
|
|
|
|
|
|
def write_pass_list(results: Iterable[ProbeResult], output_file: str | Path, write_header: bool) -> Path:
|
|
output_path = Path(output_file)
|
|
ensure_directory(output_path.parent or Path("."))
|
|
with output_path.open("w", encoding="utf-8", newline="") as handle:
|
|
if write_header:
|
|
handle.write("# channel,url\n")
|
|
for result in results:
|
|
if result.ok:
|
|
handle.write(f"{result.channel},{result.url}\n")
|
|
return output_path
|
|
|
|
|
|
def write_probe_reports(
|
|
results: list[ProbeResult],
|
|
report_dir: str | Path,
|
|
elapsed_seconds: float,
|
|
log_file: str | Path,
|
|
output_file: str | Path,
|
|
) -> tuple[Path, Path]:
|
|
report_root = ensure_directory(report_dir)
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
csv_path = report_root / f"probe_report_{timestamp}.csv"
|
|
json_path = report_root / f"probe_report_{timestamp}.json"
|
|
|
|
ordered_results = sorted(results, key=lambda item: item.index)
|
|
with csv_path.open("w", encoding="utf-8-sig", newline="") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=list(asdict(ordered_results[0]).keys()) if ordered_results else [
|
|
"index",
|
|
"channel",
|
|
"url",
|
|
"ok",
|
|
"reason",
|
|
"source_type",
|
|
"final_playlist_url",
|
|
"playlist_ms",
|
|
"segment_count",
|
|
"sample_count",
|
|
"successful_samples",
|
|
"total_bytes",
|
|
"avg_speed_kbps",
|
|
"peak_speed_kbps",
|
|
"avg_transfer_speed_kbps",
|
|
"avg_ttfb_ms",
|
|
"elapsed_seconds",
|
|
"http_statuses",
|
|
"error_detail",
|
|
])
|
|
writer.writeheader()
|
|
for result in ordered_results:
|
|
writer.writerow(asdict(result))
|
|
|
|
payload = {
|
|
"generated_at": datetime.now().isoformat(timespec="seconds"),
|
|
"elapsed_seconds": elapsed_seconds,
|
|
"log_file": str(Path(log_file)),
|
|
"output_file": str(Path(output_file)),
|
|
"totals": {
|
|
"total": len(ordered_results),
|
|
"passed": sum(1 for item in ordered_results if item.ok),
|
|
"failed": sum(1 for item in ordered_results if not item.ok),
|
|
},
|
|
"results": [asdict(item) for item in ordered_results],
|
|
}
|
|
with json_path.open("w", encoding="utf-8") as handle:
|
|
json.dump(payload, handle, ensure_ascii=False, indent=2)
|
|
|
|
return csv_path, json_path
|
|
|
|
|
|
def run_probe_file(
|
|
config: ProbeConfig,
|
|
event_callback: Callable[[dict], None] | None = None,
|
|
stop_event: threading.Event | None = None,
|
|
) -> ProbeSummary:
|
|
ensure_directory(config.report_dir)
|
|
ensure_directory(config.log_dir)
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
log_file = Path(config.log_dir) / f"probe_{timestamp}.log"
|
|
logger = OperationLogger(log_file, callback=lambda level, message, line: event_callback({"kind": "log", "level": level, "message": message, "line": line}) if event_callback else None)
|
|
prober = M3UProber(config=config, logger=logger, event_callback=event_callback, stop_event=stop_event)
|
|
|
|
targets = parse_channel_file(config.input_file)
|
|
logger.info(
|
|
f"开始检测 {len(targets)} 条链接，线程 {config.max_workers}，每主机并发 {config.per_host_limit}，速度阈值 {config.min_speed_kbps:.0f} KB/s"
|
|
)
|
|
|
|
if not targets:
|
|
output_path = write_pass_list([], config.output_file, config.write_header)
|
|
csv_path, json_path = write_probe_reports([], config.report_dir, 0.0, log_file, output_path)
|
|
return ProbeSummary(
|
|
total=0,
|
|
passed=0,
|
|
failed=0,
|
|
output_file=output_path,
|
|
csv_report=csv_path,
|
|
json_report=json_path,
|
|
log_file=log_file,
|
|
elapsed_seconds=0.0,
|
|
results=[],
|
|
)
|
|
|
|
started = time.perf_counter()
|
|
results: list[ProbeResult] = []
|
|
finished = 0
|
|
|
|
with ThreadPoolExecutor(max_workers=config.max_workers) as executor:
|
|
futures = {executor.submit(prober.probe_target, target): target for target in targets}
|
|
for future in as_completed(futures):
|
|
result = future.result()
|
|
results.append(result)
|
|
finished += 1
|
|
state = "PASS" if result.ok else "FAIL"
|
|
logger.info(
|
|
f"[{state}] {finished}/{len(targets)} {result.channel} | {result.reason} | {result.source_type} | "
|
|
f"传输{result.avg_transfer_speed_kbps:.0f}/总{result.avg_speed_kbps:.0f} KB/s | {result.successful_samples}/{max(result.sample_count, 1)} 样本"
|
|
)
|
|
if event_callback:
|
|
event_callback(
|
|
{
|
|
"kind": "probe_result",
|
|
"finished": finished,
|
|
"total": len(targets),
|
|
"result": result,
|
|
}
|
|
)
|
|
|
|
if stop_event and stop_event.is_set():
|
|
logger.warning("收到停止信号，未完成任务会在当前请求结束后退出。")
|
|
|
|
ordered_results = sorted(results, key=lambda item: item.index)
|
|
output_path = write_pass_list(ordered_results, config.output_file, config.write_header)
|
|
elapsed_seconds = time.perf_counter() - started
|
|
csv_path, json_path = write_probe_reports(
|
|
ordered_results,
|
|
config.report_dir,
|
|
elapsed_seconds,
|
|
log_file,
|
|
output_path,
|
|
)
|
|
|
|
passed = sum(1 for item in ordered_results if item.ok)
|
|
failed = len(ordered_results) - passed
|
|
logger.info(f"检测完成：通过 {passed}/{len(ordered_results)}，耗时 {elapsed_seconds:.1f} 秒")
|
|
|
|
return ProbeSummary(
|
|
total=len(ordered_results),
|
|
passed=passed,
|
|
failed=failed,
|
|
output_file=output_path,
|
|
csv_report=csv_path,
|
|
json_report=json_path,
|
|
log_file=log_file,
|
|
elapsed_seconds=elapsed_seconds,
|
|
results=ordered_results,
|
|
)
|
|
|
|
|
|
def shuffle_text_lines(file_path: str | Path, keep_comment_header: bool = True) -> tuple[int, Path]:
|
|
path = Path(file_path)
|
|
lines = read_text_file(path)
|
|
header: list[str] = []
|
|
body = lines
|
|
|
|
if keep_comment_header and lines and lines[0].lstrip().startswith("#"):
|
|
header = [lines[0]]
|
|
body = lines[1:]
|
|
|
|
random.shuffle(body)
|
|
with path.open("w", encoding="utf-8") as handle:
|
|
handle.writelines(header + body)
|
|
return len(body), path
|
|
|
|
|
|
def collect_playlist_segments(lines: list[str]) -> list[tuple[int, int]]:
|
|
segments: list[tuple[int, int]] = []
|
|
for index, line in enumerate(lines):
|
|
if not line.strip().startswith("#EXTINF"):
|
|
continue
|
|
next_index = index + 1
|
|
while next_index < len(lines):
|
|
candidate = lines[next_index].strip()
|
|
if not candidate:
|
|
next_index += 1
|
|
continue
|
|
if candidate.startswith("#"):
|
|
break
|
|
segments.append((index, next_index))
|
|
break
|
|
return segments
|
|
|
|
|
|
def trim_m3u8_file(
|
|
file_path: str | Path,
|
|
remove_segments: int = 40,
|
|
min_segments: int = 120,
|
|
) -> tuple[bool, int, Path]:
|
|
path = Path(file_path)
|
|
lines = read_text_file(path)
|
|
segments = collect_playlist_segments(lines)
|
|
if len(segments) <= min_segments:
|
|
return False, len(segments), path
|
|
|
|
to_delete: set[int] = set()
|
|
for extinf_line, segment_line in segments[:remove_segments]:
|
|
to_delete.add(extinf_line)
|
|
to_delete.add(segment_line)
|
|
|
|
new_lines: list[str] = []
|
|
media_sequence_updated = False
|
|
for index, line in enumerate(lines):
|
|
if index in to_delete:
|
|
continue
|
|
if line.startswith("#EXT-X-MEDIA-SEQUENCE:"):
|
|
try:
|
|
current_sequence = int(line.split(":", 1)[1].strip())
|
|
except ValueError:
|
|
current_sequence = 0
|
|
new_lines.append(f"#EXT-X-MEDIA-SEQUENCE:{current_sequence + remove_segments}\n")
|
|
media_sequence_updated = True
|
|
continue
|
|
new_lines.append(line)
|
|
|
|
if not media_sequence_updated:
|
|
insertion_index = 0
|
|
for insertion_index, line in enumerate(new_lines):
|
|
if line.startswith("#EXTM3U"):
|
|
continue
|
|
break
|
|
new_lines.insert(insertion_index, f"#EXT-X-MEDIA-SEQUENCE:{remove_segments}\n")
|
|
|
|
with path.open("w", encoding="utf-8") as handle:
|
|
handle.writelines(new_lines)
|
|
return True, len(segments), path
|
|
|
|
|
|
def trim_m3u8_directory(
|
|
directory: str | Path,
|
|
remove_segments: int = 40,
|
|
min_segments: int = 120,
|
|
recursive: bool = True,
|
|
) -> list[tuple[bool, int, Path]]:
|
|
root = Path(directory)
|
|
pattern = "**/*.m3u8" if recursive else "*.m3u8"
|
|
results: list[tuple[bool, int, Path]] = []
|
|
for file_path in sorted(root.glob(pattern), key=lambda item: natural_sort_key(str(item))):
|
|
results.append(trim_m3u8_file(file_path, remove_segments=remove_segments, min_segments=min_segments))
|
|
return results
|