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