#!/usr/bin/env python3 """Validate and publish complete Bilibili videos without exposing credentials. The bridge deliberately accepts only canonical Bilibili video URLs. It never reads Chrome profiles, cookies, extension tokens, or signed media URLs. A download is published only after ffprobe confirms that the file contains both video and audio streams. An authorized browser download may be handed to the bridge as a completed local file; the bridge never initiates that browser download or inspects the authenticated session. """ from __future__ import annotations import argparse import hashlib import json import math import os import re import shutil import subprocess import sys import time import uuid from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, Iterable, Sequence from urllib.parse import urlsplit SCHEMA_VERSION = "1.0" BVID_RE = re.compile(r"BV[0-9A-Za-z]{10}\Z") BATCH_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}\Z") CONTROL_RE = re.compile(r"[\x00-\x1f\x7f]") URL_RE = re.compile(r"https?://\S+", re.IGNORECASE) MEDIA_SUFFIXES = {".flv", ".m4v", ".mkv", ".mov", ".mp4", ".webm"} FORBIDDEN_SUFFIXES = {".part", ".m4s", ".webp"} DURATION_ABSOLUTE_TOLERANCE_SECONDS = 3.0 DURATION_RELATIVE_TOLERANCE = 0.001 LOCAL_FILE_SETTLE_SECONDS = 1.0 CREATE_NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0) BELOW_NORMAL_PRIORITY_CLASS = getattr(subprocess, "BELOW_NORMAL_PRIORITY_CLASS", 0) RunCommand = Callable[..., subprocess.CompletedProcess[str]] Sleeper = Callable[[float], None] class BridgeError(Exception): """Base class for safe, user-facing bridge failures.""" class InputError(BridgeError): """The batch input or command arguments are invalid.""" class AccessError(BridgeError): """The public, credential-free remote operation failed.""" class ValidationError(BridgeError): """A downloaded artifact is not a complete video.""" class CollisionError(BridgeError): """A formal output already exists and cannot be overwritten.""" @dataclass(frozen=True) class VideoItem: bvid: str source_url: str published_at: str title: str | None = None @dataclass(frozen=True) class BatchSpec: batch_id: str items: tuple[VideoItem, ...] @dataclass(frozen=True) class MediaFacts: duration_seconds: float format_name: str video_codec: str audio_codec: str def _creationflags() -> int: return CREATE_NO_WINDOW | BELOW_NORMAL_PRIORITY_CLASS def _require_plain_text(value: Any, field: str, *, maximum: int) -> str: if not isinstance(value, str): raise InputError(f"{field} must be a string") if not value or len(value) > maximum or CONTROL_RE.search(value): raise InputError(f"{field} is empty, too long, or contains control characters") return value def _canonical_source(value: Any, bvid: str, field: str) -> str: source = _require_plain_text(value, field, maximum=200) parsed = urlsplit(source) expected_path = f"/video/{bvid}" if ( parsed.scheme != "https" or parsed.hostname != "www.bilibili.com" or parsed.username is not None or parsed.password is not None or parsed.port is not None or parsed.path.rstrip("/") != expected_path or parsed.query or parsed.fragment ): raise InputError(f"{field} must be the canonical credential-free URL for {bvid}") return f"https://www.bilibili.com{expected_path}" def _published_at(value: Any, field: str) -> str: text = _require_plain_text(value, field, maximum=64) try: parsed = datetime.fromisoformat(text) except ValueError as exc: raise InputError(f"{field} must be an ISO-8601 timestamp") from exc if parsed.utcoffset() is None: raise InputError(f"{field} must include a timezone offset") return text def load_batch(path: Path) -> BatchSpec: try: raw = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError) as exc: raise InputError(f"cannot read UTF-8 batch JSON: {path}") from exc if not isinstance(raw, dict) or raw.get("schema_version") != SCHEMA_VERSION: raise InputError(f"schema_version must be {SCHEMA_VERSION!r}") batch_id = _require_plain_text(raw.get("batch_id"), "batch_id", maximum=64) if not BATCH_ID_RE.fullmatch(batch_id): raise InputError("batch_id contains unsupported characters") raw_items = raw.get("items") if not isinstance(raw_items, list) or not 1 <= len(raw_items) <= 10: raise InputError("items must contain 1 to 10 entries") items: list[VideoItem] = [] seen: set[str] = set() for index, raw_item in enumerate(raw_items): prefix = f"items[{index}]" if not isinstance(raw_item, dict): raise InputError(f"{prefix} must be an object") unknown = set(raw_item) - {"bvid", "source_url", "published_at", "title"} if unknown: raise InputError(f"{prefix} contains unsupported fields: {sorted(unknown)}") bvid = _require_plain_text(raw_item.get("bvid"), f"{prefix}.bvid", maximum=12) if not BVID_RE.fullmatch(bvid): raise InputError(f"{prefix}.bvid is invalid") if bvid.casefold() in seen: raise InputError(f"duplicate bvid: {bvid}") seen.add(bvid.casefold()) title_value = raw_item.get("title") title = None if title_value is None else _require_plain_text( title_value, f"{prefix}.title", maximum=300 ) items.append( VideoItem( bvid=bvid, source_url=_canonical_source( raw_item.get("source_url"), bvid, f"{prefix}.source_url" ), published_at=_published_at(raw_item.get("published_at"), f"{prefix}.published_at"), title=title, ) ) return BatchSpec(batch_id=batch_id, items=tuple(items)) def _resolve_executable(value: str, field: str) -> str: candidate = shutil.which(value) if candidate: return candidate path = Path(value).expanduser() if path.is_file(): return str(path.resolve()) raise InputError(f"{field} executable was not found: {value}") def _privacy_args() -> list[str]: return [ "--ignore-config", "--no-cache-dir", "--no-cookies", "--no-cookies-from-browser", "--no-playlist", ] def build_probe_command(yt_dlp: str, item: VideoItem) -> list[str]: return [ yt_dlp, *_privacy_args(), "--simulate", "--format", "bv*+ba/b", "--dump-single-json", "--no-warnings", item.source_url, ] def build_download_command( yt_dlp: str, ffmpeg: str, item: VideoItem, stage_dir: Path, result_path_file: Path, ) -> list[str]: return [ yt_dlp, *_privacy_args(), "--ffmpeg-location", ffmpeg, "--format", "bv*+ba/b", "--merge-output-format", "mp4", "--retries", "2", "--fragment-retries", "2", "--file-access-retries", "2", "--no-overwrites", "--paths", str(stage_dir), "--paths", f"temp:{stage_dir / 'temp'}", "--output", "%(id)s.%(ext)s", "--print-to-file", f"after_move:filepath", str(result_path_file), item.source_url, ] def _safe_process_error(stderr: str, *, maximum_lines: int = 8) -> str: lines = [line.strip() for line in stderr.splitlines() if line.strip()] safe = [URL_RE.sub("[URL_REDACTED]", line) for line in lines[-maximum_lines:]] return " | ".join(safe)[:2000] or "no diagnostic text" def _run( command: Sequence[str], *, runner: RunCommand = subprocess.run, cwd: Path | None = None, ) -> subprocess.CompletedProcess[str]: return runner( list(command), cwd=None if cwd is None else str(cwd), capture_output=True, text=True, encoding="utf-8", errors="replace", check=False, shell=False, creationflags=_creationflags(), ) def probe_item( item: VideoItem, yt_dlp: str, *, runner: RunCommand = subprocess.run, ) -> dict[str, Any]: completed = _run(build_probe_command(yt_dlp, item), runner=runner) if completed.returncode != 0: raise AccessError( "credential-free metadata probe failed; login/access may be required: " + _safe_process_error(completed.stderr) ) try: metadata = json.loads(completed.stdout) except json.JSONDecodeError as exc: raise AccessError("yt-dlp returned invalid metadata JSON") from exc if not isinstance(metadata, dict) or metadata.get("id") != item.bvid: raise AccessError(f"remote identity did not match requested {item.bvid}") if metadata.get("_type") not in {None, "video"} or metadata.get("entries") is not None: raise AccessError(f"{item.bvid} resolved to a playlist or multi-video result") duration = metadata.get("duration") if ( isinstance(duration, bool) or not isinstance(duration, (int, float)) or not math.isfinite(float(duration)) or duration <= 0 ): raise AccessError(f"{item.bvid} has no positive finite duration") if metadata.get("is_live") is True or metadata.get("live_status") in { "is_live", "is_upcoming", }: raise AccessError(f"{item.bvid} is live or upcoming, not a complete video") if metadata.get("availability") in { "needs_auth", "premium_only", "private", "subscriber_only", }: raise AccessError(f"{item.bvid} is not available without authentication") remote_title = str(metadata.get("title") or item.title or item.bvid) remote_title = CONTROL_RE.sub(" ", remote_title).strip()[:300] or item.bvid return { "bvid": item.bvid, "source_url": item.source_url, "title": remote_title, "duration_seconds": float(duration), "availability": metadata.get("availability"), "probe": "METADATA_PASS_ONLY", } def _read_result_path(path_file: Path, stage_dir: Path) -> Path: try: lines = [line.strip() for line in path_file.read_text(encoding="utf-8").splitlines() if line.strip()] except OSError as exc: raise ValidationError("yt-dlp did not record a completed output path") from exc if len(lines) != 1: raise ValidationError("yt-dlp must produce exactly one completed media path") output = Path(lines[0]) if not output.is_absolute(): output = stage_dir / output resolved = output.resolve() try: resolved.relative_to(stage_dir.resolve()) except ValueError as exc: raise ValidationError("yt-dlp output escaped the isolated staging directory") from exc if not resolved.is_file(): raise ValidationError("recorded yt-dlp output is not a regular file") return resolved def probe_media_file( path: Path, ffprobe: str, *, runner: RunCommand = subprocess.run, ) -> MediaFacts: suffix = path.suffix.casefold() if suffix in FORBIDDEN_SUFFIXES or suffix not in MEDIA_SUFFIXES: raise ValidationError(f"unsupported or incomplete media suffix: {suffix or ''}") command = [ ffprobe, "-v", "error", "-show_entries", "format=duration,format_name:stream=codec_type,codec_name", "-of", "json", str(path), ] completed = _run(command, runner=runner) if completed.returncode != 0: raise ValidationError("ffprobe failed: " + _safe_process_error(completed.stderr)) try: raw = json.loads(completed.stdout) streams = raw["streams"] duration = float(raw["format"]["duration"]) format_name = str(raw["format"]["format_name"]) except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: raise ValidationError("ffprobe returned incomplete media metadata") from exc video = next((stream for stream in streams if stream.get("codec_type") == "video"), None) audio = next((stream for stream in streams if stream.get("codec_type") == "audio"), None) if ( not math.isfinite(duration) or duration <= 0 or not isinstance(video, dict) or not isinstance(audio, dict) ): raise ValidationError("artifact must have positive duration plus video and audio streams") return MediaFacts( duration_seconds=duration, format_name=format_name, video_codec=str(video.get("codec_name") or "unknown"), audio_codec=str(audio.get("codec_name") or "unknown"), ) def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as source: for block in iter(lambda: source.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def validate_stable_local_media( path: Path, *, sleeper: Sleeper = time.sleep, ) -> Path: """Return a resolved, completed local media file without modifying it.""" try: resolved = path.expanduser().resolve(strict=True) except OSError as exc: raise ValidationError("browser handoff media file does not exist") from exc if not resolved.is_file(): raise ValidationError("browser handoff media path must be a regular file") suffix = resolved.suffix.casefold() if suffix in FORBIDDEN_SUFFIXES or suffix not in MEDIA_SUFFIXES: raise ValidationError(f"unsupported or incomplete media suffix: {suffix or ''}") for partial_suffix in (".crdownload", ".part"): if Path(f"{resolved}{partial_suffix}").exists(): raise ValidationError("browser download is still partial; wait for it to finish") try: before = resolved.stat() sleeper(LOCAL_FILE_SETTLE_SECONDS) after = resolved.stat() except OSError as exc: raise ValidationError("cannot inspect browser handoff media stability") from exc before_facts = (before.st_size, before.st_mtime_ns) after_facts = (after.st_size, after.st_mtime_ns) if before.st_size <= 0 or before_facts != after_facts: raise ValidationError("browser handoff media is not stable; wait for download completion") return resolved def copy_local_media_to_stage(source: Path, staged: Path) -> str: """Copy a stable handoff file to target-volume staging and verify its hash.""" try: before = source.stat() digest = hashlib.sha256() with source.open("rb") as source_handle, staged.open("xb") as target_handle: for block in iter(lambda: source_handle.read(1024 * 1024), b""): digest.update(block) target_handle.write(block) target_handle.flush() os.fsync(target_handle.fileno()) after = source.stat() except BaseException: try: staged.unlink(missing_ok=True) except OSError: pass raise if (before.st_size, before.st_mtime_ns) != (after.st_size, after.st_mtime_ns): staged.unlink(missing_ok=True) raise ValidationError("browser handoff media changed while it was copied") source_sha = digest.hexdigest() if sha256_file(staged) != source_sha: staged.unlink(missing_ok=True) raise ValidationError("staged copy hash did not match browser handoff media") return source_sha def validate_duration_consistency(remote_seconds: float, local_seconds: float) -> dict[str, float]: if ( not math.isfinite(remote_seconds) or not math.isfinite(local_seconds) or remote_seconds <= 0 or local_seconds <= 0 ): raise ValidationError("remote and local durations must be positive finite values") tolerance = max( DURATION_ABSOLUTE_TOLERANCE_SECONDS, remote_seconds * DURATION_RELATIVE_TOLERANCE, ) delta = abs(local_seconds - remote_seconds) if delta > tolerance: raise ValidationError( "downloaded duration does not match remote duration; possible preview or wrong part: " f"remote={remote_seconds:.6f}s local={local_seconds:.6f}s " f"tolerance={tolerance:.6f}s" ) return { "remote_duration_seconds": remote_seconds, "local_duration_seconds": local_seconds, "duration_delta_seconds": delta, "duration_tolerance_seconds": tolerance, } def _write_json_create_new(path: Path, payload: dict[str, Any]) -> None: encoded = (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode("utf-8") created = False try: with path.open("xb") as target: created = True target.write(encoded) target.flush() os.fsync(target.fileno()) except BaseException: if created: try: path.unlink(missing_ok=True) except OSError: pass raise def _preflight_destination(destination: Path, bvid: str) -> None: collisions = [ child for child in destination.iterdir() if child.is_file() and child.name.casefold().startswith(f"{bvid}.".casefold()) ] if collisions: raise CollisionError(f"output already exists for {bvid}; overwrite is forbidden") def publish_validated_media( source: Path, destination: Path, item: VideoItem, metadata: dict[str, Any], facts: MediaFacts, *, mapping_extra: dict[str, Any] | None = None, ) -> dict[str, Any]: duration_evidence = validate_duration_consistency( float(metadata["duration_seconds"]), facts.duration_seconds ) _preflight_destination(destination, item.bvid) media_path = destination / f"{item.bvid}{source.suffix.casefold()}" mapping_path = destination / f"{item.bvid}.download.json" size_bytes = source.stat().st_size if size_bytes <= 0: raise ValidationError("downloaded artifact is empty") source_sha = sha256_file(source) created: list[Path] = [] try: os.link(source, media_path) created.append(media_path) published_sha = sha256_file(media_path) if published_sha != source_sha: raise ValidationError("published media hash did not match validated staging media") mapping: dict[str, Any] = { "schema_version": SCHEMA_VERSION, "bvid": item.bvid, "source": item.source_url, "published_at": item.published_at, "title": metadata["title"], "local_file": media_path.name, "bytes": size_bytes, "sha256": published_sha, "duration_seconds": facts.duration_seconds, **duration_evidence, "format_name": facts.format_name, "video_codec": facts.video_codec, "audio_codec": facts.audio_codec, "completed_at": datetime.now(timezone.utc).isoformat(), } if mapping_extra: overlap = set(mapping).intersection(mapping_extra) if overlap: raise ValidationError(f"mapping extension overlaps protected fields: {sorted(overlap)}") mapping.update(mapping_extra) _write_json_create_new(mapping_path, mapping) created.append(mapping_path) return mapping except BaseException: for path in reversed(created): try: path.unlink(missing_ok=True) except OSError: pass raise def _cleanup_stage( stage_dir: Path, stage_parent: Path, *, committed: bool, ) -> str | None: active_error = sys.exc_info()[1] try: shutil.rmtree(stage_dir, ignore_errors=False) if stage_parent.exists() and not any(stage_parent.iterdir()): stage_parent.rmdir() except OSError as exc: if committed: return f"staging cleanup requires attention: {type(exc).__name__}" if active_error is not None: active_error.add_note(f"staging cleanup also failed: {type(exc).__name__}") else: raise ValidationError(f"staging cleanup failed: {type(exc).__name__}") from exc return None def download_item( item: VideoItem, destination: Path, batch_id: str, yt_dlp: str, ffmpeg: str, ffprobe: str, *, runner: RunCommand = subprocess.run, ) -> dict[str, Any]: _preflight_destination(destination, item.bvid) metadata = probe_item(item, yt_dlp, runner=runner) stage_parent = destination / ".bili-download-staging" stage_dir = stage_parent / f"{batch_id}-{item.bvid}-{uuid.uuid4().hex[:8]}" stage_dir.mkdir(parents=True, exist_ok=False) result_path_file = stage_dir / "completed-path.txt" committed = False cleanup_warning: str | None = None try: command = build_download_command(yt_dlp, ffmpeg, item, stage_dir, result_path_file) completed = _run(command, runner=runner, cwd=stage_dir) if completed.returncode != 0: raise AccessError("complete video download failed: " + _safe_process_error(completed.stderr)) source = _read_result_path(result_path_file, stage_dir) facts = probe_media_file(source, ffprobe, runner=runner) mapping = publish_validated_media(source, destination, item, metadata, facts) committed = True finally: cleanup_warning = _cleanup_stage(stage_dir, stage_parent, committed=committed) result: dict[str, Any] = {"status": "COMPLETE", **mapping} if cleanup_warning: result["warning"] = cleanup_warning return result def _find_batch_item(batch: BatchSpec, bvid: str) -> VideoItem: if not BVID_RE.fullmatch(bvid): raise InputError("bvid is invalid") matches = [item for item in batch.items if item.bvid == bvid] if len(matches) != 1: raise InputError(f"bvid must identify exactly one item in the batch: {bvid}") return matches[0] def accept_browser_file( item: VideoItem, source_file: Path, destination: Path, batch_id: str, yt_dlp: str, ffprobe: str, *, runner: RunCommand = subprocess.run, sleeper: Sleeper = time.sleep, ) -> dict[str, Any]: """Validate a completed browser-produced file and publish a read-only copy.""" if not destination.is_dir(): raise InputError("destination must already exist and be a directory") _preflight_destination(destination, item.bvid) source = validate_stable_local_media(source_file, sleeper=sleeper) metadata = probe_item(item, yt_dlp, runner=runner) stage_parent = destination / ".bili-download-staging" stage_dir = stage_parent / f"{batch_id}-{item.bvid}-{uuid.uuid4().hex[:8]}" stage_dir.mkdir(parents=True, exist_ok=False) staged = stage_dir / f"{item.bvid}{source.suffix.casefold()}" committed = False cleanup_warning: str | None = None try: source_sha = copy_local_media_to_stage(source, staged) facts = probe_media_file(staged, ffprobe, runner=runner) mapping = publish_validated_media( staged, destination, item, metadata, facts, mapping_extra={ "acquisition_mode": "authorized_browser_file_handoff", "handoff_source_sha256": source_sha, }, ) committed = True finally: cleanup_warning = _cleanup_stage(stage_dir, stage_parent, committed=committed) result: dict[str, Any] = {"status": "COMPLETE", **mapping} if cleanup_warning: result["warning"] = cleanup_warning return result def probe_batch( batch: BatchSpec, yt_dlp: str, *, runner: RunCommand = subprocess.run, ) -> tuple[list[dict[str, Any]], int]: results: list[dict[str, Any]] = [] failures = 0 for item in batch.items: try: results.append(probe_item(item, yt_dlp, runner=runner)) except BridgeError as exc: failures += 1 results.append({"bvid": item.bvid, "probe": "FAIL", "error": str(exc)}) return results, failures def download_batch( batch: BatchSpec, destination: Path, yt_dlp: str, ffmpeg: str, ffprobe: str, *, runner: RunCommand = subprocess.run, ) -> tuple[list[dict[str, Any]], int]: if not destination.is_dir(): raise InputError("destination must already exist and be a directory") results: list[dict[str, Any]] = [] failures = 0 for item in batch.items: try: results.append( download_item( item, destination, batch.batch_id, yt_dlp, ffmpeg, ffprobe, runner=runner, ) ) except BridgeError as exc: failures += 1 results.append({"bvid": item.bvid, "status": "FAIL", "error": str(exc)}) return results, failures def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description=( "Safely download public Bilibili videos or accept a completed browser file " "without exposing browser credentials." ) ) parser.add_argument("--input", required=True, type=Path, help="UTF-8 batch JSON") parser.add_argument("--yt-dlp", default="yt-dlp", help="yt-dlp executable") subparsers = parser.add_subparsers(dest="command", required=True) subparsers.add_parser("probe", help="credential-free metadata check; downloads no media") download = subparsers.add_parser("download", help="download, validate, and publish complete media") download.add_argument("--destination", required=True, type=Path) download.add_argument("--ffmpeg", default="ffmpeg", help="ffmpeg executable") download.add_argument("--ffprobe", default="ffprobe", help="ffprobe executable") handoff = subparsers.add_parser( "accept-browser-file", help="validate and publish one already-completed local browser media file", ) handoff.add_argument("--bvid", required=True, help="exact BVID from the batch JSON") handoff.add_argument("--media-file", required=True, type=Path) handoff.add_argument("--destination", required=True, type=Path) handoff.add_argument("--ffprobe", default="ffprobe", help="ffprobe executable") return parser def main(argv: Iterable[str] | None = None) -> int: args = _parser().parse_args(list(argv) if argv is not None else None) try: batch = load_batch(args.input) yt_dlp = _resolve_executable(args.yt_dlp, "yt-dlp") if args.command == "probe": results, failures = probe_batch(batch, yt_dlp) elif args.command == "download": ffmpeg = _resolve_executable(args.ffmpeg, "ffmpeg") ffprobe = _resolve_executable(args.ffprobe, "ffprobe") results, failures = download_batch(batch, args.destination, yt_dlp, ffmpeg, ffprobe) else: ffprobe = _resolve_executable(args.ffprobe, "ffprobe") item = _find_batch_item(batch, args.bvid) results = [ accept_browser_file( item, args.media_file, args.destination, batch.batch_id, yt_dlp, ffprobe, ) ] failures = 0 payload = { "batch_id": batch.batch_id, "command": args.command, "result": ( "METADATA_PASS_ONLY" if args.command == "probe" and failures == 0 else "PASS" if failures == 0 else "PARTIAL_OR_FAILED" ), "success_count": len(results) - failures, "failure_count": failures, "items": results, } print(json.dumps(payload, ensure_ascii=False, indent=2)) return 0 if failures == 0 else 2 except BridgeError as exc: print(json.dumps({"result": "SAFETY_STOP", "error": str(exc)}, ensure_ascii=False)) return 3 if __name__ == "__main__": sys.exit(main())