#!/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 '<none>'}")
|
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 '<none>'}")
|
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())
|
