776 lines
27 KiB
Python
776 lines
27 KiB
Python
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/<series_name>/<episode>/
|
||
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/<series_name>/ 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
|