From bb27963fc1f8f9a1ca15418a7f61d18293fc6574 Mon Sep 17 00:00:00 2001 From: zhouxiaofeng <2946471396@qq.com> Date: Fri, 12 Jun 2026 10:59:32 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8F=90=E4=BA=A4=E5=BF=85=E8=A6=81=E6=96=87?= =?UTF-8?q?=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 46 +++ app_runtime.py | 150 ++++++++ build_exe.bat | 11 + build_exe.py | 59 +++ convert.py | 53 +++ convertbb.py | 5 + delAllDirLine.py | 27 ++ dline.py | 28 ++ fix_gen_playlist.py | 98 +++++ m3u_tools.py | 919 ++++++++++++++++++++++++++++++++++++++++++++ randomLine.py | 20 + rename.bat | 14 + testm3u.py | 57 +++ video_pipeline.py | 775 +++++++++++++++++++++++++++++++++++++ videoconvert_gui.py | 732 +++++++++++++++++++++++++++++++++++ 15 files changed, 2994 insertions(+) create mode 100644 .gitignore create mode 100644 app_runtime.py create mode 100644 build_exe.bat create mode 100644 build_exe.py create mode 100644 convert.py create mode 100644 convertbb.py create mode 100644 delAllDirLine.py create mode 100644 dline.py create mode 100644 fix_gen_playlist.py create mode 100644 m3u_tools.py create mode 100644 randomLine.py create mode 100644 rename.bat create mode 100644 testm3u.py create mode 100644 video_pipeline.py create mode 100644 videoconvert_gui.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a64d774 --- /dev/null +++ b/.gitignore @@ -0,0 +1,46 @@ +# Python +__pycache__/ +*.py[cod] +*.pyo +*.egg-info/ +*.egg + +# Virtual environment +.venv/ +venv/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# Codex +.codex/ + +# PyInstaller build output +dist/ +build/ +*.spec + +# Runtime generated +logs/ +reports/ +pass_test.txt +test.txt +videoconvert_settings.json +*.bak + +# Batch/build helpers (keep if you want to share) +# build_exe.bat +# rename.bat + +# Downloaded/input/output data +input/ +output/ +download/ +delLine/ + +# OS +Thumbs.db +Desktop.ini diff --git a/app_runtime.py b/app_runtime.py new file mode 100644 index 0000000..0c9fe10 --- /dev/null +++ b/app_runtime.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import json +import sys +from dataclasses import dataclass, field +from pathlib import Path + + +DEFAULT_TEST_TXT = """# 频道,URL +# 示例: +# CCTV-1,http://example.com/live/index.m3u8 +""" + +DEFAULT_PASS_TEST_TXT = "# channel,url\n" + + +@dataclass(slots=True) +class RuntimePaths: + app_dir: Path + settings_file: Path + test_file: Path + pass_test_file: Path + input_dir: Path + output_dir: Path + logs_dir: Path + reports_dir: Path + del_line_dir: Path + download_dir: Path + + +def get_runtime_dir() -> Path: + if getattr(sys, "frozen", False): + return Path(sys.executable).resolve().parent + return Path(__file__).resolve().parent + + +def get_runtime_paths() -> RuntimePaths: + app_dir = get_runtime_dir() + return RuntimePaths( + app_dir=app_dir, + settings_file=app_dir / "videoconvert_settings.json", + test_file=app_dir / "test.txt", + pass_test_file=app_dir / "pass_test.txt", + input_dir=app_dir / "input", + output_dir=app_dir / "output", + logs_dir=app_dir / "logs", + reports_dir=app_dir / "reports", + del_line_dir=app_dir / "delLine", + download_dir=app_dir / "download", + ) + + +def build_default_settings(paths: RuntimePaths) -> dict: + cpu_count = 8 + try: + import os + + cpu_count = os.cpu_count() or 8 + except Exception: + pass + + return { + "probe_input_file": str(paths.test_file), + "probe_output_file": str(paths.pass_test_file), + "probe_workers": min(96, max(24, cpu_count * 2)), + "probe_host_limit": 3, + "probe_timeout": 8.0, + "probe_min_speed": 400.0, + "probe_sample_bytes": 384, + "probe_sample_segments": 3, + "probe_ffmpeg_fallback": True, + "convert_root_dir": str(paths.app_dir), + "convert_parallel_jobs": min(4, max(2, (cpu_count + 2) // 6)), + "convert_resolution": "1280:720", + "convert_video_bitrate": "0", + "convert_audio_bitrate": "96k", + "convert_hls_time": 10, + "convert_preset": "p3", + "convert_episode_mode": False, + "shuffle_file": str(paths.pass_test_file), + "shuffle_keep_header": True, + "trim_dir": str(paths.del_line_dir), + "trim_remove_segments": 40, + "trim_min_segments": 120, + "trim_recursive": True, + } + + +def _ensure_text_file(path: Path, content: str) -> None: + if not path.exists(): + path.write_text(content, encoding="utf-8") + + +def ensure_runtime_workspace() -> RuntimePaths: + paths = get_runtime_paths() + for directory in ( + paths.app_dir, + paths.input_dir, + paths.output_dir, + paths.logs_dir, + paths.reports_dir, + paths.del_line_dir, + paths.download_dir, + ): + directory.mkdir(parents=True, exist_ok=True) + + _ensure_text_file(paths.test_file, DEFAULT_TEST_TXT) + _ensure_text_file(paths.pass_test_file, DEFAULT_PASS_TEST_TXT) + + if not paths.settings_file.exists(): + defaults = build_default_settings(paths) + paths.settings_file.write_text(json.dumps(defaults, ensure_ascii=False, indent=2), encoding="utf-8") + + return paths + + +def initialize_runtime_workspace() -> dict[str, object]: + paths = ensure_runtime_workspace() + created = { + "app_dir": str(paths.app_dir), + "settings_file": str(paths.settings_file), + "test_file": str(paths.test_file), + "pass_test_file": str(paths.pass_test_file), + "input_dir": str(paths.input_dir), + "output_dir": str(paths.output_dir), + "logs_dir": str(paths.logs_dir), + "reports_dir": str(paths.reports_dir), + "del_line_dir": str(paths.del_line_dir), + "download_dir": str(paths.download_dir), + } + return created + + +def load_settings_with_defaults(paths: RuntimePaths | None = None) -> dict: + runtime_paths = paths or ensure_runtime_workspace() + defaults = build_default_settings(runtime_paths) + + if runtime_paths.settings_file.exists(): + try: + loaded = json.loads(runtime_paths.settings_file.read_text(encoding="utf-8")) + defaults.update(loaded) + except Exception: + pass + return defaults + + +def save_settings(paths: RuntimePaths, data: dict) -> None: + merged = build_default_settings(paths) + merged.update(data) + paths.settings_file.write_text(json.dumps(merged, ensure_ascii=False, indent=2), encoding="utf-8") diff --git a/build_exe.bat b/build_exe.bat new file mode 100644 index 0000000..32306c3 --- /dev/null +++ b/build_exe.bat @@ -0,0 +1,11 @@ +@echo off +setlocal +cd /d "%~dp0" + +if exist ".venv\Scripts\python.exe" ( + ".venv\Scripts\python.exe" build_exe.py +) else ( + py build_exe.py +) + +endlocal diff --git a/build_exe.py b/build_exe.py new file mode 100644 index 0000000..adaa2e7 --- /dev/null +++ b/build_exe.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +import shutil +import subprocess +import sys +from pathlib import Path + + +PROJECT_DIR = Path(__file__).resolve().parent +DIST_DIR = PROJECT_DIR / "dist" +BUILD_DIR = PROJECT_DIR / "build" +SPEC_FILE = PROJECT_DIR / "VideoConvert.spec" +ENTRY_FILE = PROJECT_DIR / "videoconvert_gui.py" +EXE_NAME = "VideoConvert" + + +def remove_if_exists(path: Path) -> None: + if path.is_dir(): + shutil.rmtree(path, ignore_errors=True) + elif path.exists(): + path.unlink() + + +def main() -> int: + remove_if_exists(DIST_DIR) + remove_if_exists(BUILD_DIR) + remove_if_exists(SPEC_FILE) + + command = [ + sys.executable, + "-m", + "PyInstaller", + "--noconfirm", + "--clean", + "--onefile", + "--windowed", + "--name", + EXE_NAME, + str(ENTRY_FILE), + ] + + print("开始打包 exe...") + print(" ".join(f'"{part}"' if " " in part else part for part in command)) + completed = subprocess.run(command, cwd=PROJECT_DIR) + if completed.returncode != 0: + return completed.returncode + + exe_path = DIST_DIR / f"{EXE_NAME}.exe" + if exe_path.exists(): + print(f"\n打包完成：{exe_path}") + print("首次启动会自动在 exe 同级目录创建 input/、output/、test.txt、pass_test.txt、logs、reports 等默认结构。") + else: + print("\nPyInstaller 已执行，但没有找到生成的 exe。") + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/convert.py b/convert.py new file mode 100644 index 0000000..b4bded6 --- /dev/null +++ b/convert.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +import argparse +import threading + +from video_pipeline import ConversionConfig, run_conversion_jobs + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="批量视频转 HLS 切片") + parser.add_argument("--root", default=".", help="任务根目录") + parser.add_argument("--jobs", type=int, default=None, help="并发转码任务数") + parser.add_argument("--resolution", default="1280:720", help="输出分辨率") + parser.add_argument("--video-bitrate", default="2000k", help="视频码率") + parser.add_argument("--audio-bitrate", default="128k", help="音频码率") + parser.add_argument("--hls-time", type=int, default=10, help="切片时长") + parser.add_argument("--preset", default="p4", help="NVENC preset") + parser.add_argument("--flat-output", action="store_true", help="关闭单集子目录模式") + return parser + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + + config = ConversionConfig( + root_dir=args.root, + parallel_jobs=args.jobs or ConversionConfig().parallel_jobs, + resolution=args.resolution, + video_bitrate=args.video_bitrate, + audio_bitrate=args.audio_bitrate, + hls_time=max(1, args.hls_time), + preset=args.preset, + episode_folder_mode=not args.flat_output, + ) + + def on_event(payload: dict) -> None: + kind = payload.get("kind") + if kind == "log": + line = payload.get("line") + if line: + print(line) + + summary = run_conversion_jobs(config, event_callback=on_event, stop_event=threading.Event()) + print( + f"\n完成：成功 {summary.succeeded}/{summary.total}，失败 {summary.failed}，" + f"日志 {summary.log_file}" + ) + return 0 if summary.failed == 0 else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/convertbb.py b/convertbb.py new file mode 100644 index 0000000..6d29803 --- /dev/null +++ b/convertbb.py @@ -0,0 +1,5 @@ +from convert import main + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/delAllDirLine.py b/delAllDirLine.py new file mode 100644 index 0000000..c918a9a --- /dev/null +++ b/delAllDirLine.py @@ -0,0 +1,27 @@ +from __future__ import annotations + +import argparse + +from m3u_tools import trim_m3u8_directory + + +def main() -> int: + parser = argparse.ArgumentParser(description="递归删除 m3u8 前置分片") + parser.add_argument("directory", nargs="?", default="delLine", help="目标目录") + parser.add_argument("--remove", type=int, default=40, help="删除前多少个分片") + parser.add_argument("--min-segments", type=int, default=120, help="最少分片阈值") + args = parser.parse_args() + + results = trim_m3u8_directory( + args.directory, + remove_segments=max(1, args.remove), + min_segments=max(1, args.min_segments), + recursive=True, + ) + processed = sum(1 for success, _, _ in results if success) + print(f"处理完成：成功 {processed} 个，跳过 {len(results) - processed} 个") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/dline.py b/dline.py new file mode 100644 index 0000000..2b7cb4a --- /dev/null +++ b/dline.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +import argparse + +from m3u_tools import trim_m3u8_directory + + +def main() -> int: + parser = argparse.ArgumentParser(description="批量删除 m3u8 前置分片") + parser.add_argument("directory", nargs="?", default="delLine", help="目标目录") + parser.add_argument("--remove", type=int, default=40, help="删除前多少个分片") + parser.add_argument("--min-segments", type=int, default=40, help="最少分片阈值，低于则跳过") + parser.add_argument("--no-recursive", action="store_true", help="只处理当前目录") + args = parser.parse_args() + + results = trim_m3u8_directory( + args.directory, + remove_segments=max(1, args.remove), + min_segments=max(1, args.min_segments), + recursive=not args.no_recursive, + ) + processed = sum(1 for success, _, _ in results if success) + print(f"处理完成：成功 {processed} 个，跳过 {len(results) - processed} 个") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/fix_gen_playlist.py b/fix_gen_playlist.py new file mode 100644 index 0000000..bf28a27 --- /dev/null +++ b/fix_gen_playlist.py @@ -0,0 +1,98 @@ +path = 'D:\\proj\\videoconvert\\video_pipeline.py' +content = open(path, 'r', encoding='utf-8').read() + +old_func = '''def generate_playlist_txt( + output_dir: str | Path, + url_prefix: str = \"http://192.168.10.32:2088\", +) -> list[Path]: + \"\"\"Scan output_dir for series folders, generate playlist txt files. + For each series, scans episode subdirs for m3u8 files and generates: + series_name + episode_name,url_prefix/series_folder/episode_folder/m3u8_file + One txt per series, placed in output_dir/.\"\"\" + output = Path(output_dir) + if not output.is_dir(): + return [] + + created: list[Path] = [] + for series_dir in sorted(output.iterdir(), key=lambda f: natural_sort_key(f.name)): + if not series_dir.is_dir(): + continue + lines: list[str] = [] + episode_dirs = sorted( + [d for d in series_dir.iterdir() if d.is_dir()], + key=lambda d: natural_sort_key(d.name), + ) + for ep_dir in episode_dirs: + m3u8_files = sorted(ep_dir.glob(\"*.m3u8\")) + if not m3u8_files: + continue + m3u8_name = m3u8_files[0].name + display_name = series_dir.name + ep_dir.name + url = f\"{url_prefix}/{series_dir.name}/{ep_dir.name}/{m3u8_name}\" + lines.append(f\"{display_name},{url}\\n\") + + if not lines: + continue + txt_path = output / f\"{series_dir.name}_playlist.txt\" + txt_path.write_text(\"\".join(lines), encoding=\"utf-8\") + created.append(txt_path) + + return created''' + +new_func = '''def generate_playlist_txt( + output_dir: str | Path, + url_prefix: str = \"http://192.168.10.32:2088\", +) -> list[Path]: + \"\"\"Scan output_dir for series folders, generate playlist txt files. + Supports both flat mode (m3u8 in series dir) and episode folder mode. + One txt per series, placed in output_dir/.\"\"\" + output = Path(output_dir) + if not output.is_dir(): + return [] + + created: list[Path] = [] + for series_dir in sorted(output.iterdir(), key=lambda f: natural_sort_key(f.name)): + if not series_dir.is_dir(): + continue + lines: list[str] = [] + + # Flat mode: m3u8 files directly in series dir + flat_m3u8 = sorted(series_dir.glob(\"*.m3u8\"), key=lambda f: natural_sort_key(f.stem)) + if flat_m3u8: + for m3u8_file in flat_m3u8: + ep_name = m3u8_file.stem + display_name = series_dir.name + ep_name + url = f\"{url_prefix}/{series_dir.name}/{m3u8_file.name}\" + lines.append(f\"{display_name},{url}\\n\") + else: + # Episode folder mode: m3u8 files in subdirectories + episode_dirs = sorted( + [d for d in series_dir.iterdir() if d.is_dir()], + key=lambda d: natural_sort_key(d.name), + ) + for ep_dir in episode_dirs: + m3u8_files = sorted(ep_dir.glob(\"*.m3u8\")) + if not m3u8_files: + continue + m3u8_name = m3u8_files[0].name + display_name = series_dir.name + ep_dir.name + url = f\"{url_prefix}/{series_dir.name}/{ep_dir.name}/{m3u8_name}\" + lines.append(f\"{display_name},{url}\\n\") + + if not lines: + continue + txt_path = output / f\"{series_dir.name}_playlist.txt\" + txt_path.write_text(\"\".join(lines), encoding=\"utf-8\") + created.append(txt_path) + + return created''' + +if old_func in content: + content = content.replace(old_func, new_func) + open(path, 'w', encoding='utf-8').write(content) + print('Replaced OK') +else: + print('Old function not found - checking around line 735') + lines = content.split('\\n') + for i in range(730, min(770, len(lines))): + print(f'{i+1}: {repr(lines[i][:120])}') diff --git a/m3u_tools.py b/m3u_tools.py new file mode 100644 index 0000000..f21f7c9 --- /dev/null +++ b/m3u_tools.py @@ -0,0 +1,919 @@ +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 diff --git a/randomLine.py b/randomLine.py new file mode 100644 index 0000000..b173bfb --- /dev/null +++ b/randomLine.py @@ -0,0 +1,20 @@ +from __future__ import annotations + +import argparse + +from m3u_tools import shuffle_text_lines + + +def main() -> int: + parser = argparse.ArgumentParser(description="随机打乱文本文件行顺序") + parser.add_argument("file", nargs="?", default="pass_test.txt", help="目标文件") + parser.add_argument("--no-header", action="store_true", help="不保留第一行注释头") + args = parser.parse_args() + + count, path = shuffle_text_lines(args.file, keep_comment_header=not args.no_header) + print(f"已打乱 {count} 行：{path}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/rename.bat b/rename.bat new file mode 100644 index 0000000..047091b --- /dev/null +++ b/rename.bat @@ -0,0 +1,14 @@ +@echo off +setlocal enabledelayedexpansion + +set count=1 + +for /f "delims=" %%i in ('dir /b /a-d') do ( + if not "%%i"=="%~nx0" ( + ren "%%i" "!count!%%~xi" + set /a count+=1 + ) +) + +echo 重命名完成！ +pause \ No newline at end of file diff --git a/testm3u.py b/testm3u.py new file mode 100644 index 0000000..43191b2 --- /dev/null +++ b/testm3u.py @@ -0,0 +1,57 @@ +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()) diff --git a/video_pipeline.py b/video_pipeline.py new file mode 100644 index 0000000..9796660 --- /dev/null +++ b/video_pipeline.py @@ -0,0 +1,775 @@ +from __future__ import annotations + +import os +import shutil +import subprocess +import threading +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from typing import Callable + +from m3u_tools import ( + OperationLogger, + VIDEO_FILE_EXTENSIONS, + discover_ffmpeg, + discover_ffprobe, + ensure_directory, + natural_sort_key, + slugify_filename, + subprocess_no_window_kwargs, +) + + +def _default_parallel_jobs() -> int: + cpu_count = os.cpu_count() or 8 + # 14600K has 20 threads, RTX 5060 can handle 3 parallel NVENC encodes + return min(4, max(1, (cpu_count + 4) // 8)) + + +def auto_rename_series(series_dir: str | Path) -> list[tuple[Path, str]]: + """Rename all video files in a directory to sequential numbers (1.mp4, 2.mp4...). + Skips empty dirs and non-video files. Returns list of (old_path, new_name).""" + dir_path = Path(series_dir) + if not dir_path.is_dir(): + return [] + + video_files = sorted( + [f for f in dir_path.iterdir() if f.is_file() and f.suffix.lower() in VIDEO_FILE_EXTENSIONS], + key=lambda f: natural_sort_key(f.stem), + ) + if not video_files: + return [] + + renamed: list[tuple[Path, str]] = [] + for idx, src in enumerate(video_files, start=1): + new_name = f"{idx}{src.suffix}" + dst = dir_path / new_name + if src != dst: + src.rename(dst) + renamed.append((src, new_name)) + return renamed + + +@dataclass(slots=True) +class ConversionConfig: + root_dir: str = "." + ffmpeg_path: str = field(default_factory=discover_ffmpeg) + ffprobe_path: str = field(default_factory=discover_ffprobe) + log_dir: str = "logs" + parallel_jobs: int = field(default_factory=_default_parallel_jobs) + resolution: str = "1280:720" + video_codec: str = "hevc_nvenc" + video_bitrate: str = "0" + video_maxrate: str = "1800k" + video_bufsize: str = "3600k" + video_cq: int = 30 + audio_codec: str = "aac" + audio_bitrate: str = "96k" + hls_time: int = 10 + preset: str = "p3" + hwaccel: str = "cuda" + max_retries: int = 2 + timeout_seconds: int = 3600 + input_dir_name: str = "input" + output_dir_name: str = "output" + episode_folder_mode: bool = False + auto_rename: bool = True + + +@dataclass(slots=True) +class ConversionJob: + name: str + input_dir: Path + output_dir: Path + mode: str + + +@dataclass(slots=True) +class ConversionTask: + job_name: str + input_file: Path + output_dir: Path + playlist_file: Path + segment_prefix: str + mode: str + + +@dataclass(slots=True) +class ConversionResult: + job_name: str + input_file: Path + output_dir: Path + success: bool + message: str + elapsed_seconds: float + + +@dataclass(slots=True) +class ConversionSummary: + total: int + succeeded: int + failed: int + log_file: Path + elapsed_seconds: float + results: list[ConversionResult] + + +def discover_series_jobs(root_dir: str | Path, config: ConversionConfig) -> list[ConversionJob]: + """Scan root_dir/input/ subdirs for video series. + Each subdir under input/ is treated as one series. + Output goes to root_dir/output/// + Falls back to legacy mode (numbered input dirs) for backward compat. + """ + root = Path(root_dir) + jobs: list[ConversionJob] = [] + + input_path = root / config.input_dir_name + output_path = root / config.output_dir_name + + if input_path.is_dir(): + for child in sorted(input_path.iterdir(), key=lambda item: natural_sort_key(item.name)): + if not child.is_dir(): + continue + video_count = sum( + 1 for f in child.iterdir() + if f.is_file() and f.suffix.lower() in VIDEO_FILE_EXTENSIONS + ) + if video_count == 0: + continue + jobs.append( + ConversionJob( + name=child.name, + input_dir=child, + output_dir=output_path / child.name, + mode="series", + ) + ) + + if jobs: + return jobs + + # Legacy: numbered input dirs directly under root + numbered_inputs = sorted( + [path for path in root.iterdir() if path.is_dir() and path.name[:-1] == "input"], + key=lambda item: natural_sort_key(item.name), + ) + for inp in numbered_inputs: + suffix = inp.name[5:] # "input" -> suffix after "input" + out = root / f"output{suffix}" + video_count = sum( + 1 for f in inp.iterdir() + if f.is_file() and f.suffix.lower() in VIDEO_FILE_EXTENSIONS + ) + if video_count == 0: + continue + jobs.append( + ConversionJob(name=inp.name, input_dir=inp, output_dir=out, mode="legacy") + ) + + if jobs: + return jobs + + # Single: video files directly in root + direct_videos = [ + path for path in root.iterdir() + if path.is_file() and path.suffix.lower() in VIDEO_FILE_EXTENSIONS + ] + if direct_videos: + jobs.append( + ConversionJob( + name=root.name, + input_dir=root, + output_dir=output_path, + mode="single", + ) + ) + return jobs + + +def discover_single_series(root_dir: str | Path, series_name: str, config: ConversionConfig) -> ConversionJob | None: + """Create a single ConversionJob for a specified series folder. + The series folder should be root_dir/input// or an absolute path. + """ + root = Path(root_dir) + series_path = Path(series_name) + if not series_path.is_absolute(): + series_path = root / config.input_dir_name / series_name + + if not series_path.is_dir(): + return None + + video_count = sum( + 1 for f in series_path.iterdir() + if f.is_file() and f.suffix.lower() in VIDEO_FILE_EXTENSIONS + ) + if video_count == 0: + return None + + output_path = root / config.output_dir_name / series_path.name + return ConversionJob( + name=series_path.name, + input_dir=series_path, + output_dir=output_path, + mode="series", + ) + + +def build_conversion_tasks(jobs: list[ConversionJob], config: ConversionConfig) -> list[ConversionTask]: + tasks: list[ConversionTask] = [] + for job in jobs: + audio_seen: set[str] = set() + video_files = sorted( + [item for item in job.input_dir.iterdir() if item.is_file() and item.suffix.lower() in VIDEO_FILE_EXTENSIONS], + key=lambda item: natural_sort_key(item.name), + ) + for input_file in video_files: + episode_name = slugify_filename(input_file.stem) + if config.episode_folder_mode: + output_dir = job.output_dir / episode_name + playlist_file = output_dir / "index.m3u8" + else: + output_dir = job.output_dir + playlist_file = output_dir / f"{episode_name}.m3u8" + + tasks.append( + ConversionTask( + job_name=job.name, + input_file=input_file, + output_dir=output_dir, + playlist_file=playlist_file, + segment_prefix=episode_name, + mode=job.mode, + ) + ) + return tasks + + +class VideoConverter: + def __init__( + self, + config: ConversionConfig, + logger: OperationLogger, + 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._active_processes: list[subprocess.Popen] = [] + self._process_lock = threading.Lock() + + def log(self, level: str, message: str) -> None: + getattr(self.logger, level, self.logger.info)(message) + + def register_process(self, process: subprocess.Popen) -> None: + with self._process_lock: + self._active_processes.append(process) + + def unregister_process(self, process: subprocess.Popen) -> None: + with self._process_lock: + if process in self._active_processes: + self._active_processes.remove(process) + + def terminate_active_processes(self) -> None: + with self._process_lock: + for process in self._active_processes: + try: + process.terminate() + except Exception: + pass + + def build_filter(self) -> str: + return f"scale={self.config.resolution}:flags=lanczos,setsar=1:1,format=yuv420p" + + def probe_duration(self, file_path: Path) -> float: + try: + cmd = [ + self.config.ffprobe_path, + "-v", + "quiet", + "-show_entries", + "format=duration", + "-of", + "default=noprint_wrappers=1:nokey=1", + str(file_path), + ] + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=30, + **subprocess_no_window_kwargs(), + ) + if result.returncode == 0 and result.stdout.strip(): + return float(result.stdout.strip()) + except Exception: + pass + return 0.0 + + def cleanup_task_output(self, task: ConversionTask) -> None: + if task.playlist_file.exists(): + task.playlist_file.unlink() + if task.output_dir.exists(): + for f in list(task.output_dir.iterdir()): + if f.stem.startswith(task.segment_prefix) and f.suffix == '.ts': + f.unlink() + ensure_directory(task.output_dir) + + def build_ffmpeg_command(self, task: ConversionTask) -> list[str]: + segment_template = task.output_dir / f"{task.segment_prefix}_%03d.ts" + command = [ + self.config.ffmpeg_path, + "-y", + "-hide_banner", + "-loglevel", + "warning", + "-threads", + "0", + "-hwaccel", + self.config.hwaccel, + "-i", + str(task.input_file), + "-map", + "0:v:0", + "-map", + "0:a:0?", + "-sn", + "-dn", + "-map_metadata", + "-1", + "-map_chapters", + "-1", + "-vf", + self.build_filter(), + "-c:v", + self.config.video_codec, + "-preset", + self.config.preset, + "-rc", + "vbr", + "-cq", + str(self.config.video_cq), + "-b:v", + self.config.video_bitrate, + "-maxrate", + self.config.video_maxrate, + "-bufsize", + self.config.video_bufsize, + "-profile:v", + "main", + "-spatial_aq", + "1", + "-temporal_aq", + "1", + "-aq-strength", + "8", + "-rc-lookahead", + "20", + "-c:a", + self.config.audio_codec, + "-ac", + "2", + "-ar", + "48000", + "-b:a", + self.config.audio_bitrate, + "-force_key_frames", + f"expr:gte(t,n_forced*{self.config.hls_time})", + "-f", + "hls", + "-hls_time", + str(self.config.hls_time), + "-hls_playlist_type", + "vod", + "-hls_flags", + "independent_segments", + "-hls_list_size", + "0", + "-hls_segment_filename", + str(segment_template), + "-progress", + "pipe:1", + "-nostats", + str(task.playlist_file), + ] + return command + + def convert_task(self, task: ConversionTask) -> ConversionResult: + started = time.perf_counter() + duration_seconds = self.probe_duration(task.input_file) + last_percent = -1 + last_progress_time = time.perf_counter() + + for attempt in range(self.config.max_retries + 1): + if self.stop_event.is_set(): + return ConversionResult( + job_name=task.job_name, + input_file=task.input_file, + output_dir=task.output_dir, + success=False, + message="任务已停止", + elapsed_seconds=time.perf_counter() - started, + ) + + self.cleanup_task_output(task) + command = self.build_ffmpeg_command(task) + self.log( + "info", + f"开始转码 {task.job_name} / {task.input_file.name} -> {task.playlist_file}", + ) + process = subprocess.Popen( + command, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + stdin=subprocess.DEVNULL, + text=True, + encoding="utf-8", + errors="replace", + bufsize=1, + **subprocess_no_window_kwargs(), + ) + self.register_process(process) + diagnostics: list[str] = [] + current_out_time = 0.0 + + try: + while True: + if self.stop_event.is_set(): + process.terminate() + raise RuntimeError("任务已停止") + + line = process.stdout.readline() if process.stdout else "" + if not line and process.poll() is not None: + break + + line = line.strip() + if not line: + continue + + if "=" not in line: + diagnostics.append(line) + diagnostics[:] = diagnostics[-20:] + continue + + key, value = line.split("=", 1) + if key == "out_time_ms": + try: + new_out = float(value) / 1_000_000.0 + current_out_time = new_out + if duration_seconds > 0: + percent = min(int(new_out / duration_seconds * 100), 100) + if percent != last_percent: + last_percent = percent + if self.event_callback: + self.event_callback({ + "kind": "convert_progress", + "job_name": task.job_name, + "file_name": task.input_file.name, + "percent": percent, + "out_time": new_out, + "duration": duration_seconds, + }) + last_progress_time = time.perf_counter() + except ValueError: + pass + elif key == "progress" and value == "end": + # ffmpeg signals completion via progress=end + pass + elif key not in { + "frame", "fps", "stream_0_0_q", "bitrate", + "total_size", "dup_frames", "drop_frames", + "speed", "out_time", "elapsed", "progress", + }: + diagnostics.append(line) + diagnostics[:] = diagnostics[-20:] + + # Stall detection: kill if no progress for 120s and encoding + if duration_seconds > 0 and current_out_time < duration_seconds: + stalled = time.perf_counter() - last_progress_time + if stalled > 120: + self.log("warning", f"FFmpeg 无进度超过 {stalled:.0f}秒，终止重试") + process.kill() + raise RuntimeError("FFmpeg 进度卡死") + + process.wait() + except subprocess.TimeoutExpired: + process.kill() + process.wait() + if attempt >= self.config.max_retries: + return ConversionResult( + job_name=task.job_name, + input_file=task.input_file, + output_dir=task.output_dir, + success=False, + message="转码超时", + elapsed_seconds=time.perf_counter() - started, + ) + self.log("warning", f"FFmpeg 超时，重试 {attempt+1}/{self.config.max_retries}") + time.sleep(2) + continue + except RuntimeError as error: + process.kill() + process.wait() + if attempt >= self.config.max_retries: + return ConversionResult( + job_name=task.job_name, + input_file=task.input_file, + output_dir=task.output_dir, + success=False, + message=str(error), + elapsed_seconds=time.perf_counter() - started, + ) + self.log("warning", f"FFmpeg 异常: {error}，重试 {attempt+1}/{self.config.max_retries}") + time.sleep(2) + continue + except Exception as error: + process.kill() + process.wait() + return ConversionResult( + job_name=task.job_name, + input_file=task.input_file, + output_dir=task.output_dir, + success=False, + message=str(error), + elapsed_seconds=time.perf_counter() - started, + ) + finally: + self.unregister_process(process) + + if process.returncode == 0: + return ConversionResult( + job_name=task.job_name, + input_file=task.input_file, + output_dir=task.output_dir, + success=True, + message=f"成功（CQ {self.config.video_cq}）", + elapsed_seconds=time.perf_counter() - started, + ) + + # Diagnose failure + playlist_exists = task.playlist_file.exists() + segment_files = list(task.output_dir.glob("*.ts")) + message = f"FFmpeg 返回码 {process.returncode}" + if diagnostics: + snippet = " | ".join(diagnostics[-6:]) + message += f" | {snippet}" + if not playlist_exists: + message += " | 未生成 m3u8" + elif len(segment_files) >= 2: + # Output is partial but non-empty; treat as degraded success + self.log("warning", f"FFmpeg 返回码 {process.returncode} 但已生成 {len(segment_files)} 个分片，视为部分成功") + return ConversionResult( + job_name=task.job_name, + input_file=task.input_file, + output_dir=task.output_dir, + success=True, + message=f"部分成功（{len(segment_files)} 分片，返回码 {process.returncode}）", + elapsed_seconds=time.perf_counter() - started, + ) + + if attempt >= self.config.max_retries: + return ConversionResult( + job_name=task.job_name, + input_file=task.input_file, + output_dir=task.output_dir, + success=False, + message=message, + elapsed_seconds=time.perf_counter() - started, + ) + self.log("warning", f"FFmpeg 失败 {message}，重试 {attempt+1}/{self.config.max_retries}") + time.sleep(2) + + return ConversionResult( + job_name=task.job_name, + input_file=task.input_file, + output_dir=task.output_dir, + success=False, + message="未知错误", + elapsed_seconds=time.perf_counter() - started, + ) + + +def run_conversion_jobs( + config: ConversionConfig, + event_callback: Callable[[dict], None] | None = None, + stop_event: threading.Event | None = None, +) -> ConversionSummary: + ensure_directory(config.log_dir) + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + log_file = Path(config.log_dir) / f"convert_{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, + ) + converter = VideoConverter(config=config, logger=logger, event_callback=event_callback, stop_event=stop_event) + + jobs = discover_series_jobs(config.root_dir, config) + if not jobs: + logger.warning(f"没有在 {Path(config.root_dir).resolve()} 找到可处理的视频任务。") + return ConversionSummary( + total=0, succeeded=0, failed=0, log_file=log_file, + elapsed_seconds=0.0, results=[], + ) + + tasks = build_conversion_tasks(jobs, config) + logger.info(f"发现 {len(jobs)} 个剧集目录，待处理 {len(tasks)} 个视频，最大并发 {config.parallel_jobs}") + + if not tasks: + logger.warning("未找到可转码的视频文件。") + return ConversionSummary( + total=0, succeeded=0, failed=0, log_file=log_file, + elapsed_seconds=0.0, results=[], + ) + + started = time.perf_counter() + results: list[ConversionResult] = [] + finished = 0 + + with ThreadPoolExecutor(max_workers=config.parallel_jobs) as executor: + futures = {executor.submit(converter.convert_task, task): task for task in tasks} + for future in as_completed(futures): + result = future.result() + results.append(result) + finished += 1 + logger.info( + f"[{'PASS' if result.success else 'FAIL'}] {finished}/{len(tasks)} " + f"{result.job_name} / {result.input_file.name} | {result.message} | {result.elapsed_seconds:.1f}s" + ) + if event_callback: + event_callback({ + "kind": "convert_result", + "finished": finished, + "total": len(tasks), + "result": result, + }) + if stop_event and stop_event.is_set(): + converter.terminate_active_processes() + + elapsed_seconds = time.perf_counter() - started + succeeded = sum(1 for item in results if item.success) + failed = len(results) - succeeded + logger.info(f"转码完成：成功 {succeeded}/{len(results)}，耗时 {elapsed_seconds:.1f} 秒") + + return ConversionSummary( + total=len(results), succeeded=succeeded, failed=failed, + log_file=log_file, elapsed_seconds=elapsed_seconds, results=results, + ) + +def run_specific_jobs( + jobs: list[ConversionJob], + config: ConversionConfig, + event_callback: Callable[[dict], None] | None = None, + stop_event: threading.Event | None = None, +) -> ConversionSummary: + ensure_directory(config.log_dir) + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + log_file = Path(config.log_dir) / f"convert_{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, + ) + converter = VideoConverter(config=config, logger=logger, event_callback=event_callback, stop_event=stop_event) + + tasks = build_conversion_tasks(jobs, config) + logger.info(f"处理 {len(jobs)} 个剧集目录，待处理 {len(tasks)} 个视频，最大并发 {config.parallel_jobs}") + + if not tasks: + logger.warning("未找到可转码的视频文件。") + return ConversionSummary( + total=0, succeeded=0, failed=0, log_file=log_file, + elapsed_seconds=0.0, results=[], + ) + + started = time.perf_counter() + results: list[ConversionResult] = [] + finished = 0 + + with ThreadPoolExecutor(max_workers=config.parallel_jobs) as executor: + futures = {executor.submit(converter.convert_task, task): task for task in tasks} + for future in as_completed(futures): + result = future.result() + results.append(result) + finished += 1 + logger.info( + f"[{'PASS' if result.success else 'FAIL'}] {finished}/{len(tasks)} " + f"{result.job_name} / {result.input_file.name} | {result.message} | {result.elapsed_seconds:.1f}s" + ) + if event_callback: + event_callback({ + "kind": "convert_result", + "finished": finished, + "total": len(tasks), + "result": result, + }) + if stop_event and stop_event.is_set(): + converter.terminate_active_processes() + + elapsed_seconds = time.perf_counter() - started + succeeded = sum(1 for item in results if item.success) + failed = len(results) - succeeded + logger.info(f"转码完成：成功 {succeeded}/{len(results)}，耗时 {elapsed_seconds:.1f} 秒") + + return ConversionSummary( + total=len(results), succeeded=succeeded, failed=failed, + log_file=log_file, elapsed_seconds=elapsed_seconds, results=results, + ) + +def generate_playlist_txt( + output_dir: str | Path, + url_prefix: str = "http://192.168.10.32:2088", +) -> list[Path]: + """Scan output_dir for series folders, generate playlist txt files. + Supports both flat mode (m3u8 in series dir) and episode folder mode. + One txt per series, placed in output_dir/.""" + output = Path(output_dir) + if not output.is_dir(): + return [] + + created: list[Path] = [] + for series_dir in sorted(output.iterdir(), key=lambda f: natural_sort_key(f.name)): + if not series_dir.is_dir(): + continue + lines: list[str] = [] + + # Flat mode: m3u8 files directly in series dir + flat_m3u8 = sorted(series_dir.glob("*.m3u8"), key=lambda f: natural_sort_key(f.stem)) + if flat_m3u8: + for m3u8_file in flat_m3u8: + ep_name = m3u8_file.stem + display_name = series_dir.name + ep_name + url = f"{url_prefix}/{series_dir.name}/{m3u8_file.name}" + lines.append(f"{display_name},{url}\n") + else: + # Episode folder mode: m3u8 files in subdirectories + episode_dirs = sorted( + [d for d in series_dir.iterdir() if d.is_dir()], + key=lambda d: natural_sort_key(d.name), + ) + for ep_dir in episode_dirs: + m3u8_files = sorted(ep_dir.glob("*.m3u8")) + if not m3u8_files: + continue + m3u8_name = m3u8_files[0].name + display_name = series_dir.name + ep_dir.name + url = f"{url_prefix}/{series_dir.name}/{ep_dir.name}/{m3u8_name}" + lines.append(f"{display_name},{url}\n") + + if not lines: + continue + txt_path = output / f"{series_dir.name}_playlist.txt" + txt_path.write_text("".join(lines), encoding="utf-8") + created.append(txt_path) + + return created diff --git a/videoconvert_gui.py b/videoconvert_gui.py new file mode 100644 index 0000000..813619d --- /dev/null +++ b/videoconvert_gui.py @@ -0,0 +1,732 @@ +from __future__ import annotations + +import argparse +import os +import queue +import threading +import tkinter as tk +from pathlib import Path +from tkinter import filedialog, messagebox, ttk +from tkinter.scrolledtext import ScrolledText + +from app_runtime import ( + RuntimePaths, + ensure_runtime_workspace, + initialize_runtime_workspace, + load_settings_with_defaults, + save_settings as save_runtime_settings, +) +from m3u_tools import ( + ProbeConfig, + ProbeResult, + run_probe_file, + shuffle_text_lines, + trim_m3u8_directory, +) +from video_pipeline import ( + ConversionConfig, + ConversionJob, + ConversionResult, + auto_rename_series, + discover_series_jobs, + discover_single_series, + generate_playlist_txt, + run_specific_jobs, +) + +RUNTIME_PATHS = ensure_runtime_workspace() +APP_DIR = RUNTIME_PATHS.app_dir +INPUT_DIR = RUNTIME_PATHS.input_dir +OUTPUT_DIR = RUNTIME_PATHS.output_dir + + +class VideoConvertApp(tk.Tk): + def __init__(self) -> None: + super().__init__() + self.runtime_paths: RuntimePaths = RUNTIME_PATHS + self.title('VideoConvert \u5de5\u5177\u7bb1') + self.geometry('1480x900') + self.minsize(1200, 760) + + self.settings = load_settings_with_defaults(self.runtime_paths) + self.ui_queue: queue.Queue[dict] = queue.Queue() + self.active_thread: threading.Thread | None = None + self.active_stop_event: threading.Event | None = None + self.active_target: str | None = None + + self.probe_passed = 0 + self.probe_failed = 0 + self.convert_succeeded = 0 + self.convert_failed = 0 + + self.discovered_jobs: list[ConversionJob] = [] + + self._build_variables() + self._build_style() + self._build_layout() + self._load_initial_state() + self.after(120, self._drain_queue) + self.protocol('WM_DELETE_WINDOW', self.on_close) + + def _build_variables(self) -> None: + cpu_count = os.cpu_count() or 8 + default_probe_workers = min(96, max(24, cpu_count * 2)) + default_convert_jobs = min(4, max(1, (cpu_count + 4) // 8)) + + self.probe_input_var = tk.StringVar(value=self.settings.get('probe_input_file', str(RUNTIME_PATHS.test_file))) + self.probe_output_var = tk.StringVar(value=self.settings.get('probe_output_file', str(RUNTIME_PATHS.pass_test_file))) + self.probe_workers_var = tk.IntVar(value=self.settings.get('probe_workers', default_probe_workers)) + self.probe_host_limit_var = tk.IntVar(value=self.settings.get('probe_host_limit', 3)) + self.probe_timeout_var = tk.DoubleVar(value=self.settings.get('probe_timeout', 15.0)) + self.probe_min_speed_var = tk.DoubleVar(value=self.settings.get('probe_min_speed', 400.0)) + self.probe_sample_bytes_var = tk.IntVar(value=self.settings.get('probe_sample_bytes', 2048)) + self.probe_sample_segments_var = tk.IntVar(value=self.settings.get('probe_sample_segments', 3)) + self.probe_ffmpeg_fallback_var = tk.BooleanVar(value=self.settings.get('probe_ffmpeg_fallback', True)) + + self.convert_parallel_var = tk.IntVar(value=self.settings.get('convert_parallel_jobs', default_convert_jobs)) + self.convert_resolution_var = tk.StringVar(value=self.settings.get('convert_resolution', '1280:720')) + self.convert_video_bitrate_var = tk.StringVar(value=self.settings.get('convert_video_bitrate', '0')) + self.convert_audio_bitrate_var = tk.StringVar(value=self.settings.get('convert_audio_bitrate', '96k')) + self.convert_hls_time_var = tk.IntVar(value=self.settings.get('convert_hls_time', 10)) + self.convert_preset_var = tk.StringVar(value=self.settings.get('convert_preset', 'p3')) + self.convert_episode_mode_var = tk.BooleanVar(value=self.settings.get('convert_episode_mode', False)) + self.convert_auto_rename_var = tk.BooleanVar(value=self.settings.get('convert_auto_rename', True)) + + self.shuffle_file_var = tk.StringVar(value=self.settings.get('shuffle_file', str(RUNTIME_PATHS.pass_test_file))) + self.shuffle_keep_header_var = tk.BooleanVar(value=self.settings.get('shuffle_keep_header', True)) + + self.trim_dir_var = tk.StringVar(value=self.settings.get('trim_dir', str(RUNTIME_PATHS.del_line_dir))) + self.trim_remove_var = tk.IntVar(value=self.settings.get('trim_remove_segments', 40)) + self.trim_min_var = tk.IntVar(value=self.settings.get('trim_min_segments', 120)) + self.trim_recursive_var = tk.BooleanVar(value=self.settings.get('trim_recursive', True)) + + self.probe_status_var = tk.StringVar(value='\u5f85\u673a') + self.probe_summary_var = tk.StringVar(value='\u901a\u8fc7 0 / \u5931\u8d25 0') + self.convert_status_var = tk.StringVar(value='\u5f85\u673a') + self.convert_summary_var = tk.StringVar(value='\u6210\u529f 0 / \u5931\u8d25 0') + self.convert_current_var = tk.StringVar(value='\u5f53\u524d\u6587\u4ef6\uff1a\u65e0') + self.playlist_status_var = tk.StringVar(value='') + self.playlist_dir_var = tk.StringVar(value=str(OUTPUT_DIR)) + + def _build_style(self) -> None: + style = ttk.Style(self) + if 'vista' in style.theme_names(): + style.theme_use('vista') + style.configure('Header.TLabel', font=('Microsoft YaHei UI', 15, 'bold')) + style.configure('Hint.TLabel', foreground='#4f5f6f') + style.configure('Status.TLabel', foreground='#1a5fb4') + style.configure('Folder.TLabel', font=('Consolas', 10), foreground='#2e3436') + + def _build_layout(self) -> None: + notebook = ttk.Notebook(self) + notebook.pack(fill='both', expand=True, padx=12, pady=12) + + self.probe_tab = ttk.Frame(notebook, padding=12) + self.convert_tab = ttk.Frame(notebook, padding=12) + self.tools_tab = ttk.Frame(notebook, padding=12) + + notebook.add(self.probe_tab, text='M3U \u68c0\u6d4b') + notebook.add(self.convert_tab, text='\u89c6\u9891\u8f6c\u5207\u7247') + notebook.add(self.tools_tab, text='\u5e38\u7528\u5de5\u5177') + + self._build_probe_tab() + self._build_convert_tab() + self._build_tools_tab() + + # Probe tab + + def _build_probe_tab(self) -> None: + ttk.Label(self.probe_tab, text='M3U \u94fe\u63a5\u68c0\u6d4b', style='Header.TLabel').pack(anchor='w') + ttk.Label(self.probe_tab, + text='HTTP \u76f4\u63a5\u91c7\u6837\u64ad\u653e\u5217\u8868\u548c\u5206\u7247\uff0c\u5fc5\u8981\u65f6\u81ea\u52a8\u56de\u9000 FFmpeg \u7ba1\u9053\u6d4b\u901f\u3002\u4e0d\u518d\u751f\u6210\u4e34\u65f6 ts \u6587\u4ef6\u3002', + style='Hint.TLabel').pack(anchor='w', pady=(2, 10)) + + config_frame = ttk.LabelFrame(self.probe_tab, text='\u914d\u7f6e') + config_frame.pack(fill='x', pady=(0, 10)) + for c in range(4): + config_frame.columnconfigure(c, weight=1) + + self._labeled_entry(config_frame, '\u8f93\u5165\u6587\u4ef6', self.probe_input_var, 0, 0, browse_file=True) + self._labeled_entry(config_frame, '\u8f93\u51fa\u6587\u4ef6', self.probe_output_var, 0, 2, browse_save=True) + self._labeled_entry(config_frame, '\u68c0\u6d4b\u7ebf\u7a0b', self.probe_workers_var, 1, 0, width=10) + self._labeled_entry(config_frame, '\u5355\u4e3b\u673a\u5e76\u53d1', self.probe_host_limit_var, 1, 1, width=10) + self._labeled_entry(config_frame, '\u8bf7\u6c42\u8d85\u65f6(\u79d2)', self.probe_timeout_var, 1, 2, width=10) + self._labeled_entry(config_frame, '\u6700\u4f4e\u901f\u5ea6(KB/s)', self.probe_min_speed_var, 1, 3, width=12) + self._labeled_entry(config_frame, '\u91c7\u6837\u5927\u5c0f(KB)', self.probe_sample_bytes_var, 2, 0, width=10) + self._labeled_entry(config_frame, '\u91c7\u6837\u5206\u7247\u6570', self.probe_sample_segments_var, 2, 1, width=10) + + ttk.Checkbutton(config_frame, text='HTTP \u89e3\u6790\u5931\u8d25\u65f6\u5141\u8bb8 FFmpeg \u56de\u9000\u68c0\u6d4b', + variable=self.probe_ffmpeg_fallback_var).grid(row=2, column=2, columnspan=2, sticky='w', padx=8, pady=8) + + action_frame = ttk.Frame(self.probe_tab) + action_frame.pack(fill='x', pady=(0, 10)) + self.probe_start_button = ttk.Button(action_frame, text='\u5f00\u59cb\u68c0\u6d4b', command=self.start_probe) + self.probe_start_button.pack(side='left') + self.probe_stop_button = ttk.Button(action_frame, text='\u505c\u6b62', command=self.stop_active_task, state='disabled') + self.probe_stop_button.pack(side='left', padx=8) + ttk.Label(action_frame, textvariable=self.probe_status_var, style='Status.TLabel').pack(side='left', padx=(12, 0)) + ttk.Label(action_frame, textvariable=self.probe_summary_var).pack(side='left', padx=(12, 0)) + + self.probe_progress = ttk.Progressbar(self.probe_tab, mode='determinate') + self.probe_progress.pack(fill='x', pady=(0, 10)) + + result_pane = ttk.Panedwindow(self.probe_tab, orient='vertical') + result_pane.pack(fill='both', expand=True) + + tree_frame = ttk.Frame(result_pane) + result_pane.add(tree_frame, weight=3) + columns = ('channel', 'status', 'source', 'speed', 'samples', 'reason') + self.probe_tree = ttk.Treeview(tree_frame, columns=columns, show='headings', height=15) + headings = {'channel': '\u9891\u9053', 'status': '\u72b6\u6001', 'source': '\u68c0\u6d4b\u6e90', + 'speed': '\u901f\u5ea6 KB/s', 'samples': '\u91c7\u6837', 'reason': '\u8bf4\u660e'} + widths = {'channel': 220, 'status': 80, 'source': 90, 'speed': 100, 'samples': 80, 'reason': 520} + for col in columns: + self.probe_tree.heading(col, text=headings[col]) + self.probe_tree.column(col, width=widths[col], anchor='w') + scroll = ttk.Scrollbar(tree_frame, orient='vertical', command=self.probe_tree.yview) + self.probe_tree.configure(yscrollcommand=scroll.set) + self.probe_tree.pack(side='left', fill='both', expand=True) + scroll.pack(side='right', fill='y') + + log_frame = ttk.LabelFrame(result_pane, text='\u65e5\u5fd7') + result_pane.add(log_frame, weight=2) + self.probe_log = ScrolledText(log_frame, height=12, font=('Consolas', 10)) + self.probe_log.pack(fill='both', expand=True, padx=6, pady=6) + + # Convert tab + + def _build_convert_tab(self) -> None: + ttk.Label(self.convert_tab, text='\u89c6\u9891\u8f6c HLS \u5207\u7247', style='Header.TLabel').pack(anchor='w') + ttk.Label(self.convert_tab, + text='HEVC NVENC \u7f16\u7801\uff0c720P\uff0cCQ 30 \u6781\u81f4\u7701\u7a7a\u95f4\u3002CPU+GPU \u6ee1\u8f7d\u52a0\u901f\u3002', + style='Hint.TLabel').pack(anchor='w', pady=(2, 10)) + + folder_frame = ttk.LabelFrame(self.convert_tab, text='\u76ee\u5f55\u7ed3\u6784') + folder_frame.pack(fill='x', pady=(0, 8)) + inp = str(INPUT_DIR) + out = str(OUTPUT_DIR) + ttk.Label(folder_frame, text=f'\u8f93\u5165: {inp}/<\u7535\u89c6\u5267>/', style='Folder.TLabel').pack(anchor='w', padx=8, pady=(4, 0)) + ttk.Label(folder_frame, text=f'\u8f93\u51fa: {out}/<\u7535\u89c6\u5267>/<\u96c6\u540d>/index.m3u8', style='Folder.TLabel').pack(anchor='w', padx=8, pady=(0, 4)) + + action_frame = ttk.Frame(self.convert_tab) + action_frame.pack(fill='x', pady=(0, 8)) + + self.convert_scan_button = ttk.Button(action_frame, text='\u626b\u63cf input/', command=self.scan_conversion_jobs) + self.convert_scan_button.pack(side='left') + + self.convert_pick_button = ttk.Button(action_frame, text='\u9009\u62e9\u5267\u96c6\u76ee\u5f55', command=self.pick_series_folder) + self.convert_pick_button.pack(side='left', padx=8) + + self.convert_start_button = ttk.Button(action_frame, text='\u5f00\u59cb\u8f6c\u7801', command=self.start_convert) + self.convert_start_button.pack(side='left', padx=(20, 0)) + + self.convert_stop_button = ttk.Button(action_frame, text='\u505c\u6b62', command=self.stop_active_task, state='disabled') + self.convert_stop_button.pack(side='left', padx=8) + + ttk.Label(action_frame, textvariable=self.convert_status_var, style='Status.TLabel').pack(side='left', padx=(12, 0)) + ttk.Label(action_frame, textvariable=self.convert_summary_var).pack(side='left', padx=(8, 0)) + ttk.Label(action_frame, textvariable=self.convert_current_var).pack(side='right') + + self.convert_progress = ttk.Progressbar(self.convert_tab, mode='determinate') + self.convert_progress.pack(fill='x', pady=(0, 8)) + + pane = ttk.Panedwindow(self.convert_tab, orient='vertical') + pane.pack(fill='both', expand=True) + + jobs_frame = ttk.LabelFrame(pane, text='\u53d1\u73b0\u7684\u5267\u96c6\u4efb\u52a1') + pane.add(jobs_frame, weight=2) + jcols = ('name', 'mode', 'input', 'output', 'videos') + self.jobs_tree = ttk.Treeview(jobs_frame, columns=jcols, show='headings', height=7) + jheads = {'name': '\u5267\u96c6\u540d\u79f0', 'mode': '\u6765\u6e90', 'input': '\u8f93\u5165\u76ee\u5f55', + 'output': '\u8f93\u51fa\u76ee\u5f55', 'videos': '\u89c6\u9891\u6570'} + jw = {'name': 180, 'mode': 70, 'input': 420, 'output': 420, 'videos': 80} + for col in jcols: + self.jobs_tree.heading(col, text=jheads[col]) + self.jobs_tree.column(col, width=jw[col], anchor='w') + self.jobs_tree.pack(fill='both', expand=True, padx=6, pady=6) + + config_frame = ttk.LabelFrame(pane, text='\u8f6c\u7801\u53c2\u6570') + pane.add(config_frame, weight=0) + for c in range(5): + config_frame.columnconfigure(c, weight=1) + + self._labeled_entry(config_frame, '\u5e76\u53d1\u4efb\u52a1', self.convert_parallel_var, 0, 0, width=8) + self._labeled_entry(config_frame, '\u5206\u8fa8\u7387', self.convert_resolution_var, 0, 1, width=12) + self._labeled_entry(config_frame, '\u89c6\u9891\u7801\u7387(0=CQ)', self.convert_video_bitrate_var, 0, 2, width=10) + self._labeled_entry(config_frame, '\u97f3\u9891\u7801\u7387', self.convert_audio_bitrate_var, 0, 3, width=10) + self._labeled_entry(config_frame, '\u5207\u7247\u65f6\u957f(\u79d2)', self.convert_hls_time_var, 0, 4, width=8) + self._labeled_entry(config_frame, 'NVENC Preset', self.convert_preset_var, 1, 0, width=8) + + check_frame = ttk.Frame(config_frame) + check_frame.grid(row=1, column=1, columnspan=3, sticky='w', padx=8, pady=6) + ttk.Checkbutton(check_frame, text='\u6bcf\u96c6\u5355\u72ec\u5b50\u76ee\u5f55', variable=self.convert_episode_mode_var).pack(side='left') + ttk.Checkbutton(check_frame, text='\u81ea\u52a8\u91cd\u547d\u540d\u6587\u4ef6(\u987a\u5e8f\u7f16\u53f7)', variable=self.convert_auto_rename_var).pack(side='left', padx=(16, 0)) + + result_frame = ttk.LabelFrame(pane, text='\u6587\u4ef6\u7ed3\u679c') + pane.add(result_frame, weight=2) + rcols = ('job', 'file', 'status', 'elapsed', 'message') + self.convert_result_tree = ttk.Treeview(result_frame, columns=rcols, show='headings', height=7) + rheads = {'job': '\u5267\u96c6', 'file': '\u6587\u4ef6', 'status': '\u72b6\u6001', + 'elapsed': '\u8017\u65f6(s)', 'message': '\u8bf4\u660e'} + rw = {'job': 180, 'file': 220, 'status': 80, 'elapsed': 90, 'message': 680} + for col in rcols: + self.convert_result_tree.heading(col, text=rheads[col]) + self.convert_result_tree.column(col, width=rw[col], anchor='w') + self.convert_result_tree.pack(fill='both', expand=True, padx=6, pady=6) + + log_frame = ttk.LabelFrame(pane, text='\u65e5\u5fd7') + pane.add(log_frame, weight=1) + self.convert_log = ScrolledText(log_frame, height=8, font=('Consolas', 10)) + self.convert_log.pack(fill='both', expand=True, padx=6, pady=6) + + # Tools tab + + def _build_tools_tab(self) -> None: + ttk.Label(self.tools_tab, text='\u5c0f\u5de5\u5177', style='Header.TLabel').pack(anchor='w') + ttk.Label(self.tools_tab, text='\u6253\u4e71\u6587\u4ef6\u884c\u987a\u5e8f\u3001\u5220\u9664\u524d N \u4e2a\u5206\u7247\u7b49\u5e38\u7528\u529f\u80fd\u3002', + style='Hint.TLabel').pack(anchor='w', pady=(2, 10)) + + sf = ttk.LabelFrame(self.tools_tab, text='\u6253\u4e71\u6587\u4ef6\u884c\u987a\u5e8f') + sf.pack(fill='x', pady=(0, 10)) + ttk.Label(sf, text='\u76ee\u6807\u6587\u4ef6').grid(row=0, column=0, sticky='w', padx=8, pady=8) + ttk.Entry(sf, textvariable=self.shuffle_file_var).grid(row=0, column=1, sticky='ew', padx=8, pady=8) + ttk.Button(sf, text='\u9009\u62e9\u6587\u4ef6', command=self.pick_shuffle_file).grid(row=0, column=2, padx=8, pady=8) + ttk.Checkbutton(sf, text='\u4fdd\u7559\u7b2c\u4e00\u884c\u6ce8\u91ca\u5934', variable=self.shuffle_keep_header_var).grid(row=1, column=1, sticky='w', padx=8, pady=0) + ttk.Button(sf, text='\u5f00\u59cb\u6253\u4e71', command=self.run_shuffle).grid(row=1, column=2, padx=8, pady=8) + sf.columnconfigure(1, weight=1) + + tf = ttk.LabelFrame(self.tools_tab, text='\u5220\u9664\u524d\u7f6e\u5206\u7247') + tf.pack(fill='x', pady=(0, 10)) + ttk.Label(tf, text='\u76ee\u6807\u76ee\u5f55').grid(row=0, column=0, sticky='w', padx=8, pady=8) + ttk.Entry(tf, textvariable=self.trim_dir_var).grid(row=0, column=1, sticky='ew', padx=8, pady=8) + ttk.Button(tf, text='\u9009\u62e9\u76ee\u5f55', command=self.pick_trim_dir).grid(row=0, column=2, padx=8, pady=8) + self._grid_labeled_entry(tf, '\u5220\u9664\u524d N \u7247', self.trim_remove_var, 1, 0) + self._grid_labeled_entry(tf, '\u6700\u5c11\u5206\u7247\u9608\u503c', self.trim_min_var, 1, 1) + ttk.Checkbutton(tf, text='\u9012\u5f52\u5904\u7406\u5b50\u76ee\u5f55', variable=self.trim_recursive_var).grid(row=1, column=2, sticky='w', padx=8, pady=8) + + bf = ttk.Frame(tf) + bf.grid(row=2, column=2, sticky='e', padx=8, pady=8) + ttk.Button(bf, text='\u9009\u62e9 output/ \u76ee\u5f55', command=self.pick_output_trim_dir).pack(side='left', padx=(0, 8)) + ttk.Button(bf, text='\u5f00\u59cb\u5904\u7406', command=self.run_trim).pack(side='left') + tf.columnconfigure(1, weight=1) + + pf = ttk.LabelFrame(self.tools_tab, text='\u751f\u6210\u64ad\u653e\u5217\u8868') + pf.pack(fill='x', pady=(0, 10)) + ttk.Entry(pf, textvariable=self.playlist_dir_var).pack(side='left', fill='x', expand=True, padx=8, pady=8) + ttk.Button(pf, text='\u9009\u62e9\u76ee\u5f55', command=self.pick_playlist_dir).pack(side='left', padx=8, pady=8) + ttk.Button(pf, text='\u751f\u6210\u64ad\u653e\u5217\u8868', command=self.run_generate_playlist).pack(side='left', padx=8, pady=8) + ttk.Label(pf, textvariable=self.playlist_status_var).pack(side='left', padx=(12, 0)) + + scheme_frame = ttk.LabelFrame(self.tools_tab, text='\u63a8\u8350\u76ee\u5f55\u65b9\u6848') + scheme_frame.pack(fill='both', expand=True) + scheme_text = ScrolledText(scheme_frame, height=14, font=('Consolas', 10)) + scheme_text.pack(fill='both', expand=True, padx=6, pady=6) + scheme_text.insert('1.0', + '\u63a8\u8350\u4f7f\u7528 input/ + output/ \u65b9\u6848\uff1a\n\n' + 'input/\n' + ' \u7504\u5ac2\u4f20/\n' + ' 1.mp4\n 2.mp4\n 3.mp4\n' + ' \u7405\u7430\u699c/\n' + ' 1.mp4\n 2.mp4\n\n' + 'output/\n' + ' \u7504\u5ac2\u4f20/\n' + ' 1/index.m3u8\n 2/index.m3u8\n' + ' \u7405\u7430\u699c/\n' + ' 1/index.m3u8\n\n' + '\u7279\u70b9\uff1a\n' + '1. input/ \u4e0b\u6bcf\u90e8\u7535\u89c6\u5267\u4e00\u4e2a\u6587\u4ef6\u5939\uff0c\u81ea\u52a8\u626b\u63cf\u8bc6\u522b\n' + '2. \u652f\u6301\u81ea\u52a8\u91cd\u547d\u540d\uff1a\u5c06\u6587\u4ef6\u5939\u5185\u6240\u6709\u89c6\u9891\u6309\u987a\u5e8f\u7f16\u53f7\n' + '3. \u7a7a\u76ee\u5f55\u81ea\u52a8\u8df3\u8fc7\uff0c\u7a0b\u5e8f\u4e0d\u4f1a\u5d29\u6e83\n' + '4. \u652f\u6301\u201c\u9009\u62e9\u5267\u96c6\u76ee\u5f55\u201d\u6309\u94ae\u5904\u7406\u4efb\u610f\u4f4d\u7f6e\u7684\u5267\u96c6\u6587\u4ef6\u5939\n' + '5. \u8f93\u51fa\u4fdd\u7559\u539f\u59cb\u76ee\u5f55\u540d\uff0c\u65b9\u4fbf\u540e\u671f\u6574\u7406\u548c\u64ad\u653e\n') + scheme_text.configure(state='disabled') + + # Helper methods + + def _labeled_entry(self, parent, label, variable, row, column, width=None, browse_file=False, browse_save=False): + frame = ttk.Frame(parent) + frame.grid(row=row, column=column, padx=8, pady=6, sticky='ew') + ttk.Label(frame, text=label).pack(anchor='w') + entry = ttk.Entry(frame, textvariable=variable, width=width) + entry.pack(side='left', fill='x', expand=True) + if browse_file: + ttk.Button(frame, text='\u6d4f\u89c8', command=lambda var=variable: self.pick_file(var)).pack(side='left', padx=(6, 0)) + if browse_save: + ttk.Button(frame, text='\u53e6\u5b58', command=lambda var=variable: self.pick_save_file(var)).pack(side='left', padx=(6, 0)) + + def _grid_labeled_entry(self, parent, label, variable, row, column): + frame = ttk.Frame(parent) + frame.grid(row=row, column=column, padx=8, pady=8, sticky='ew') + ttk.Label(frame, text=label).pack(anchor='w') + ttk.Entry(frame, textvariable=variable, width=12).pack(side='left', fill='x', expand=True) + + def _append_log(self, widget, message): + widget.insert('end', message + '\n') + widget.see('end') + + def _clear_probe_ui(self): + self.probe_passed = 0 + self.probe_failed = 0 + self.probe_summary_var.set('\u901a\u8fc7 0 / \u5931\u8d25 0') + self.probe_progress.configure(value=0, maximum=100) + self.probe_tree.delete(*self.probe_tree.get_children()) + self.probe_log.delete('1.0', 'end') + + def _clear_convert_ui(self): + self.convert_succeeded = 0 + self.convert_failed = 0 + self.convert_summary_var.set('\u6210\u529f 0 / \u5931\u8d25 0') + self.convert_progress.configure(value=0, maximum=100) + self.convert_current_var.set('\u5f53\u524d\u6587\u4ef6\uff1a\u65e0') + self.convert_result_tree.delete(*self.convert_result_tree.get_children()) + self.convert_log.delete('1.0', 'end') + + def _set_busy(self, target, busy): + self.active_target = target if busy else None + if target == 'probe': + self.probe_start_button.configure(state='disabled' if busy else 'normal') + self.probe_stop_button.configure(state='normal' if busy else 'disabled') + elif target == 'convert': + self.convert_start_button.configure(state='disabled' if busy else 'normal') + self.convert_stop_button.configure(state='normal' if busy else 'disabled') + self.convert_scan_button.configure(state='disabled' if busy else 'normal') + self.convert_pick_button.configure(state='disabled' if busy else 'normal') + + # Pick / Scan + + def pick_file(self, variable): + path = filedialog.askopenfilename(initialdir=APP_DIR) + if path: + variable.set(path) + + def pick_save_file(self, variable): + path = filedialog.asksaveasfilename(initialdir=APP_DIR) + if path: + variable.set(path) + + def pick_shuffle_file(self): + self.pick_file(self.shuffle_file_var) + + def pick_trim_dir(self): + path = filedialog.askdirectory(initialdir=self.trim_dir_var.get() or APP_DIR) + if path: + self.trim_dir_var.set(path) + + def pick_output_trim_dir(self): + if OUTPUT_DIR.is_dir(): + self.trim_dir_var.set(str(OUTPUT_DIR)) + else: + path = filedialog.askdirectory(initialdir=APP_DIR) + if path: + self.trim_dir_var.set(path) + + def scan_conversion_jobs(self): + self.jobs_tree.delete(*self.jobs_tree.get_children()) + self.discovered_jobs = [] + config = ConversionConfig(root_dir=str(APP_DIR), episode_folder_mode=self.convert_episode_mode_var.get()) + jobs = discover_series_jobs(APP_DIR, config) + exts = ('.mp4', '.mkv', '.avi', '.mov', '.flv', '.wmv', '.m4v') + for j in jobs: + vc = sum(1 for f in j.input_dir.iterdir() if f.is_file() and f.suffix.lower() in exts) + self.jobs_tree.insert('', 'end', values=(j.name, j.mode, str(j.input_dir), str(j.output_dir), vc)) + self.discovered_jobs = jobs + if not jobs: + self._append_log(self.convert_log, f'\u672a\u5728 {INPUT_DIR}/ \u4e0b\u53d1\u73b0\u4efb\u4f55\u5267\u96c6\u76ee\u5f55\uff08\u6216\u76ee\u5f55\u4e3a\u7a7a\uff09\u3002') + + def pick_series_folder(self): + path = filedialog.askdirectory(initialdir=str(INPUT_DIR)) + if not path: + return + folder = Path(path) + exts = ('.mp4', '.mkv', '.avi', '.mov', '.flv', '.wmv', '.m4v') + video_count = sum(1 for f in folder.iterdir() if f.is_file() and f.suffix.lower() in exts) + if video_count == 0: + messagebox.showinfo('\u7a7a\u76ee\u5f55', f'\u6240\u9009\u76ee\u5f55\u4e2d\u6ca1\u6709\u627e\u5230\u89c6\u9891\u6587\u4ef6\uff1a\n{folder}') + return + job = ConversionJob(name=folder.name, input_dir=folder, output_dir=OUTPUT_DIR / folder.name, mode='\u81ea\u5b9a\u4e49') + self.jobs_tree.delete(*self.jobs_tree.get_children()) + self.discovered_jobs = [job] + self.jobs_tree.insert('', 'end', values=(job.name, job.mode, str(job.input_dir), str(job.output_dir), video_count)) + + # Load / Save + + def _load_initial_state(self): + self.scan_conversion_jobs() + + def _save_form_settings(self): + payload = { + 'probe_input_file': self.probe_input_var.get(), + 'probe_output_file': self.probe_output_var.get(), + 'probe_workers': self.probe_workers_var.get(), + 'probe_host_limit': self.probe_host_limit_var.get(), + 'probe_timeout': self.probe_timeout_var.get(), + 'probe_min_speed': self.probe_min_speed_var.get(), + 'probe_sample_bytes': self.probe_sample_bytes_var.get(), + 'probe_sample_segments': self.probe_sample_segments_var.get(), + 'probe_ffmpeg_fallback': self.probe_ffmpeg_fallback_var.get(), + 'convert_parallel_jobs': self.convert_parallel_var.get(), + 'convert_resolution': self.convert_resolution_var.get(), + 'convert_video_bitrate': self.convert_video_bitrate_var.get(), + 'convert_audio_bitrate': self.convert_audio_bitrate_var.get(), + 'convert_hls_time': self.convert_hls_time_var.get(), + 'convert_preset': self.convert_preset_var.get(), + 'convert_episode_mode': self.convert_episode_mode_var.get(), + 'convert_auto_rename': self.convert_auto_rename_var.get(), + 'shuffle_file': self.shuffle_file_var.get(), + 'shuffle_keep_header': self.shuffle_keep_header_var.get(), + 'trim_dir': self.trim_dir_var.get(), + 'trim_remove_segments': self.trim_remove_var.get(), + 'trim_min_segments': self.trim_min_var.get(), + 'trim_recursive': self.trim_recursive_var.get(), + } + save_runtime_settings(self.runtime_paths, payload) + + # Probe Actions + + def start_probe(self): + if self.active_thread and self.active_thread.is_alive(): + messagebox.showinfo('\u5fd9\u788c\u4e2d', '\u5f53\u524d\u5df2\u6709\u4efb\u52a1\u5728\u8fd0\u884c\uff0c\u8bf7\u5148\u505c\u6b62\u6216\u7b49\u5f85\u5b8c\u6210\u3002') + return + input_file = Path(self.probe_input_var.get()) + if not input_file.exists(): + messagebox.showerror('\u6587\u4ef6\u4e0d\u5b58\u5728', f'\u6ca1\u6709\u627e\u5230\u8f93\u5165\u6587\u4ef6\uff1a\n{input_file}') + return + self._save_form_settings() + self._clear_probe_ui() + self.probe_status_var.set('\u68c0\u6d4b\u4e2d') + self._set_busy('probe', True) + self.active_stop_event = threading.Event() + config = ProbeConfig( + input_file=self.probe_input_var.get(), + output_file=self.probe_output_var.get(), + max_workers=max(1, self.probe_workers_var.get()), + per_host_limit=max(1, self.probe_host_limit_var.get()), + request_timeout=max(1.0, self.probe_timeout_var.get()), + min_speed_kbps=max(1.0, self.probe_min_speed_var.get()), + sample_bytes=max(64, self.probe_sample_bytes_var.get()) * 1024, + sample_segments=max(1, self.probe_sample_segments_var.get()), + ffmpeg_fallback=self.probe_ffmpeg_fallback_var.get(), + ) + + def worker(): + summary = run_probe_file( + config, + event_callback=lambda payload: self.ui_queue.put({'target': 'probe', **payload}), + stop_event=self.active_stop_event, + ) + self.ui_queue.put({'target': 'probe', 'kind': 'probe_summary', 'summary': summary}) + + self.active_thread = threading.Thread(target=worker, daemon=True) + self.active_thread.start() + + def start_convert(self): + if self.active_thread and self.active_thread.is_alive(): + messagebox.showinfo('\u5fd9\u788c\u4e2d', '\u5f53\u524d\u5df2\u6709\u4efb\u52a1\u5728\u8fd0\u884c\uff0c\u8bf7\u5148\u505c\u6b62\u6216\u7b49\u5f85\u5b8c\u6210\u3002') + return + if not self.discovered_jobs: + messagebox.showerror('\u65e0\u4efb\u52a1', '\u6ca1\u6709\u627e\u5230\u53ef\u8f6c\u7801\u7684\u89c6\u9891\u4efb\u52a1\u3002\u8bf7\u5c06\u89c6\u9891\u653e\u5165 input/<\u7535\u89c6\u5267>/ \u76ee\u5f55\u540e\u70b9\u51fb\u201c\u626b\u63cf input/\u201d\uff0c\u6216\u4f7f\u7528\u201c\u9009\u62e9\u5267\u96c6\u76ee\u5f55\u201d\u6309\u94ae\u3002') + return + if self.convert_auto_rename_var.get(): + renamed_total = 0 + for job in self.discovered_jobs: + renamed_total += len(auto_rename_series(job.input_dir)) + if renamed_total > 0: + self._append_log(self.convert_log, f'\u5df2\u81ea\u52a8\u91cd\u547d\u540d {renamed_total} \u4e2a\u6587\u4ef6') + self._save_form_settings() + self._clear_convert_ui() + self.convert_status_var.set('\u8f6c\u7801\u4e2d') + self._set_busy('convert', True) + self.active_stop_event = threading.Event() + config = ConversionConfig( + root_dir=str(APP_DIR), + parallel_jobs=max(1, self.convert_parallel_var.get()), + resolution=self.convert_resolution_var.get().strip(), + video_bitrate=self.convert_video_bitrate_var.get().strip(), + audio_bitrate=self.convert_audio_bitrate_var.get().strip(), + hls_time=max(1, self.convert_hls_time_var.get()), + preset=self.convert_preset_var.get().strip(), + episode_folder_mode=self.convert_episode_mode_var.get(), + ) + jobs_copy = list(self.discovered_jobs) + + def worker(): + summary = run_specific_jobs( + jobs_copy, config, + event_callback=lambda payload: self.ui_queue.put({'target': 'convert', **payload}), + stop_event=self.active_stop_event, + ) + self.ui_queue.put({'target': 'convert', 'kind': 'convert_summary', 'summary': summary}) + + self.active_thread = threading.Thread(target=worker, daemon=True) + self.active_thread.start() + + def stop_active_task(self): + if self.active_stop_event: + self.active_stop_event.set() + if self.active_target == 'probe': + self.probe_status_var.set('\u6b63\u5728\u505c\u6b62') + elif self.active_target == 'convert': + self.convert_status_var.set('\u6b63\u5728\u505c\u6b62') + + def run_shuffle(self): + target_file = Path(self.shuffle_file_var.get()) + if not target_file.exists(): + messagebox.showerror('\u6587\u4ef6\u4e0d\u5b58\u5728', f'\u6ca1\u6709\u627e\u5230\u6587\u4ef6\uff1a\n{target_file}') + return + self._save_form_settings() + count, path = shuffle_text_lines(target_file, keep_comment_header=self.shuffle_keep_header_var.get()) + messagebox.showinfo('\u5b8c\u6210', f'\u5df2\u6253\u4e71 {count} \u884c\uff1a\n{path}') + + def run_trim(self): + target_dir = Path(self.trim_dir_var.get()) + if not target_dir.exists(): + messagebox.showerror('\u76ee\u5f55\u4e0d\u5b58\u5728', f'\u6ca1\u6709\u627e\u5230\u76ee\u5f55\uff1a\n{target_dir}') + return + self._save_form_settings() + results = trim_m3u8_directory( + target_dir, + remove_segments=max(1, self.trim_remove_var.get()), + min_segments=max(1, self.trim_min_var.get()), + recursive=self.trim_recursive_var.get(), + ) + processed = sum(1 for s, _, _ in results if s) + skipped = len(results) - processed + messagebox.showinfo('\u5b8c\u6210', f'\u5904\u7406\u5b8c\u6210\uff1a\u6210\u529f {processed} \u4e2a\uff0c\u8df3\u8fc7 {skipped} \u4e2a\u3002') + + # Queue Processing + + def _drain_queue(self): + try: + while True: + item = self.ui_queue.get_nowait() + target = item.get('target', '') + kind = item.get('kind', '') + payload = {k: v for k, v in item.items() if k not in ('target', 'kind')} + if target == 'probe': + self._handle_probe_event(kind, payload) + elif target == 'convert': + self._handle_convert_event(kind, payload) + except queue.Empty: + pass + finally: + self.after(120, self._drain_queue) + + def _handle_probe_event(self, kind, payload): + if kind == 'log': + line = payload.get('line') or f"[{payload.get('level', 'INFO')}] {payload.get('message', '')}" + self._append_log(self.probe_log, line) + return + if kind == 'probe_result': + result = payload['result'] + if result.ok: + self.probe_passed += 1 + else: + self.probe_failed += 1 + finished = payload['finished'] + total = max(payload['total'], 1) + self.probe_progress.configure(maximum=total, value=finished) + self.probe_summary_var.set(f'\u901a\u8fc7 {self.probe_passed} / \u5931\u8d25 {self.probe_failed}') + self.probe_tree.insert('', 0, values=( + result.channel, + '\u901a\u8fc7' if result.ok else '\u5931\u8d25', + result.source_type, + f'{result.avg_speed_kbps:.1f}', + f'{result.successful_samples}/{result.sample_count}', + result.reason, + )) + return + if kind == 'probe_summary': + summary = payload['summary'] + self.probe_status_var.set('\u5df2\u5b8c\u6210') + self.probe_summary_var.set(f'\u901a\u8fc7 {summary.passed} / \u5931\u8d25 {summary.failed}') + self._append_log(self.probe_log, + f'\u7ed3\u679c\u5df2\u5199\u5165\uff1a{summary.output_file}\nCSV \u62a5\u544a\uff1a{summary.csv_report}\nJSON \u62a5\u544a\uff1a{summary.json_report}') + self._set_busy('probe', False) + self.active_thread = None + self.active_stop_event = None + + def _handle_convert_event(self, kind, payload): + if kind == 'log': + line = payload.get('line') or f"[{payload.get('level', 'INFO')}] {payload.get('message', '')}" + self._append_log(self.convert_log, line) + return + if kind == 'convert_progress': + self.convert_current_var.set( + f"\u5f53\u524d\u6587\u4ef6\uff1a{payload.get('job_name')} / {payload.get('file_name')} ({payload.get('percent')}%)") + return + if kind == 'convert_result': + result = payload['result'] + if result.success: + self.convert_succeeded += 1 + else: + self.convert_failed += 1 + finished = payload['finished'] + total = max(payload['total'], 1) + self.convert_progress.configure(maximum=total, value=finished) + self.convert_summary_var.set(f'\u6210\u529f {self.convert_succeeded} / \u5931\u8d25 {self.convert_failed}') + self.convert_result_tree.insert('', 0, values=( + result.job_name, result.input_file.name, + '\u901a\u8fc7' if result.success else '\u5931\u8d25', + f'{result.elapsed_seconds:.1f}', result.message, + )) + return + if kind == 'convert_summary': + summary = payload['summary'] + self.convert_status_var.set('\u5df2\u5b8c\u6210') + self.convert_summary_var.set(f'\u6210\u529f {summary.succeeded} / \u5931\u8d25 {summary.failed}') + self.convert_current_var.set('\u5f53\u524d\u6587\u4ef6\uff1a\u65e0') + self._append_log(self.convert_log, f'\u65e5\u5fd7\u6587\u4ef6\uff1a{summary.log_file}') + self._set_busy('convert', False) + self.active_thread = None + self.active_stop_event = None + self.scan_conversion_jobs() + + def pick_playlist_dir(self): + path = filedialog.askdirectory(initialdir=self.playlist_dir_var.get() or str(OUTPUT_DIR)) + if path: + self.playlist_dir_var.set(path) + + def run_generate_playlist(self): + output_path = Path(self.playlist_dir_var.get()) + if not output_path.is_dir(): + messagebox.showerror('\u76ee\u5f55\u4e0d\u5b58\u5728', f'\u8f93\u51fa\u76ee\u5f55\u4e0d\u5b58\u5728\uff1a\\n{output_path}') + return + created = generate_playlist_txt(output_path) + if created: + names = '\n'.join(str(p.name) for p in created) + self.playlist_status_var.set(f'\u751f\u6210 {len(created)} \u4e2a\u6587\u4ef6') + messagebox.showinfo('\u5b8c\u6210', f'\u5df2\u751f\u6210 {len(created)} \u4e2a\u64ad\u653e\u5217\u6587\u4ef6\uff1a\\n{names}') + else: + self.playlist_status_var.set('\u672a\u627e\u5230\u5206\u7247\u6587\u4ef6') + messagebox.showinfo('\u63d0\u793a', f'\u5728 {output_path}/ \u4e0b\u672a\u627e\u5230\u53ef\u751f\u6210\u64ad\u653e\u5217\u7684\u5267\u96c6\u76ee\u5f55\u3002') + + def on_close(self): + + self._save_form_settings() + if self.active_thread and self.active_thread.is_alive(): + if not messagebox.askyesno('\u4ecd\u6709\u4efb\u52a1\u8fd0\u884c', '\u5f53\u524d\u6709\u4efb\u52a1\u6b63\u5728\u8fd0\u884c\uff0c\u786e\u8ba4\u9000\u51fa\u5417\uff1f'): + return + if self.active_stop_event: + self.active_stop_event.set() + self.destroy() + + +def build_parser(): + parser = argparse.ArgumentParser(description='VideoConvert \u684c\u9762\u5de5\u5177') + parser.add_argument('--init-only', action='store_true', + help='\u53ea\u521d\u59cb\u5316 exe \u540c\u7ea7\u76ee\u5f55\u4e0b\u7684\u9ed8\u8ba4\u6587\u4ef6\u548c\u6587\u4ef6\u5939\uff0c\u7136\u540e\u9000\u51fa') + return parser + + +def main(): + args = build_parser().parse_args() + if args.init_only: + info = initialize_runtime_workspace() + print(f'\u521d\u59cb\u5316\u5b8c\u6210\uff1a{info["app_dir"]}') + return + app = VideoConvertApp() + app.mainloop() + + +if __name__ == '__main__': + main()