import gzip import shutil import subprocess from pathlib import Path import httpx FFMPEG_DIR = Path(__file__).resolve().parents[2] / "tools" / "ffmpeg" FFMPEG_EXE = FFMPEG_DIR / "ffmpeg.exe" FFMPEG_URL = "https://github.com/eugeneware/ffmpeg-static/releases/download/b6.1.1/ffmpeg-win32-x64.gz" def ffmpeg_path() -> Path | None: if FFMPEG_EXE.is_file(): return FFMPEG_EXE found = shutil.which("ffmpeg") return Path(found) if found else None def ensure_ffmpeg() -> Path: current = ffmpeg_path() if current is not None: return current FFMPEG_DIR.mkdir(parents=True, exist_ok=True) archive = FFMPEG_DIR / "ffmpeg-win32-x64.gz" try: with httpx.Client(follow_redirects=True, timeout=120.0, trust_env=False) as client: response = client.get(FFMPEG_URL) response.raise_for_status() archive.write_bytes(response.content) import gzip FFMPEG_EXE.write_bytes(gzip.decompress(archive.read_bytes())) finally: archive.unlink(missing_ok=True) if not FFMPEG_EXE.is_file(): raise RuntimeError("没有找到 ffmpeg") return FFMPEG_EXE def convert_to_mp4(episode_dir: Path, mp4_path: Path) -> None: exe = ensure_ffmpeg() mp4_path.parent.mkdir(parents=True, exist_ok=True) temporary = mp4_path.with_name(f"{mp4_path.stem}.part.mp4") completed = subprocess.run( [ str(exe), "-y", "-allowed_extensions", "ALL", "-i", "video.m3u8", "-c", "copy", "-movflags", "+faststart", "-f", "mp4", str(temporary), ], cwd=episode_dir, capture_output=True, check=False, ) if completed.returncode != 0 or not temporary.is_file() or temporary.stat().st_size <= 0: temporary.unlink(missing_ok=True) detail = (completed.stderr or b"").decode("utf-8", "replace")[-400:] raise RuntimeError("转换成 MP4 失败" + (f":{detail}" if detail else "")) temporary.replace(mp4_path) def unused_zip_helper(path: Path) -> None: if path.suffix == ".zip": with zipfile.ZipFile(path) as archive: archive.extractall(path.parent)