From 129e0ebd2ca859b3463ad2c31ae335c72bace22d Mon Sep 17 00:00:00 2001
From: Cai <cai@nbcai.cc>
Date: Sat, 05 Sep 2026 23:02:28 +0800
Subject: [PATCH] 刷新2026H1股票估值覆盖与半导体三子链
---
dev/project-dev/bili_authenticated_extension/worker.py | 1582 +++++++++++++++++++++++++++++++++++++++++++++++++++++----
1 files changed, 1,453 insertions(+), 129 deletions(-)
diff --git a/dev/project-dev/bili_authenticated_extension/worker.py b/dev/project-dev/bili_authenticated_extension/worker.py
index 6d1a2ca..3516a7c 100644
--- a/dev/project-dev/bili_authenticated_extension/worker.py
+++ b/dev/project-dev/bili_authenticated_extension/worker.py
@@ -7,17 +7,22 @@
from __future__ import annotations
+import ctypes
import hashlib
import importlib.util
import io
import json
import math
import os
+import re
import shutil
+import stat
import subprocess
import sys
import time
import uuid
+from collections.abc import Mapping
+from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Iterable, Sequence
@@ -25,27 +30,63 @@
from .constants import (
BRIDGE_TIMEOUT_SECONDS,
- CANONICAL_URL,
- DURATION_TOLERANCE_MS,
- EXPECTED_DURATION_MS,
+ EXTENSION_BUILD,
EXTRACTOR_RETRIES,
FILE_ACCESS_RETRIES,
FRAGMENT_RETRIES,
HTTP_RETRIES,
+ RELOAD_GENERATION,
SOCKET_TIMEOUT_SECONDS,
- TARGET_BVID,
YTDLP_MODULE_SHA256,
YTDLP_VERSION,
+ duration_tolerance_ms,
+ validate_bvid,
+ validate_creator_uid,
)
-from .protocol import ProtocolError, strict_json_loads, validate_start
+from .formal_legacy_identity_manifest import (
+ FORMAL_LEGACY_CREATOR_UID,
+ FORMAL_LEGACY_INTEGER_UID_ROWS,
+ FORMAL_LEGACY_ROWS,
+ FORMAL_PREFIX_BYTES,
+ FORMAL_PREFIX_LINES,
+ FORMAL_PREFIX_SHA256,
+)
+from .protocol import (
+ ProtocolError, strict_json_loads, validate_media_complete_identity, validate_start,
+)
-FROZEN_BRIDGE_SHA256 = "749FC486B0F42315BD463F11771FE2A7C71CAB53DD9AC2E411CE82E1175DFF13"
+FROZEN_BRIDGE_SHA256 = "00F11DAF8387160DB863C89F0B33AB8480422233FF42199189222C989C7ED07E"
+SUBPROCESS_POLICY_ERROR_CODES = frozenset({
+ "E_SUBPROCESS_POLICY_ARGUMENTS",
+ "E_SUBPROCESS_POLICY_ENVIRONMENT",
+ "E_SUBPROCESS_POLICY_EVENT_SHAPE",
+ "E_SUBPROCESS_POLICY_EXECUTABLE",
+ "E_SUBPROCESS_POLICY_LOCAL_PATH",
+ "E_SUBPROCESS_POLICY_SECRET",
+})
+_BRIDGE_MAPPING_KEYS = frozenset({
+ "schema_version", "bvid", "source", "published_at", "title", "local_file",
+ "bytes", "sha256", "duration_seconds", "remote_duration_seconds",
+ "local_duration_seconds", "duration_delta_seconds", "duration_tolerance_seconds",
+ "format_name", "video_codec", "audio_codec", "completed_at", "acquisition_mode",
+ "handoff_source_sha256",
+})
+_BRIDGE_ITEM_KEYS = _BRIDGE_MAPPING_KEYS | {"status"}
+_BRIDGE_ITEM_WARNING_KEYS = _BRIDGE_ITEM_KEYS | {"warning"}
+_LOWER_SHA256_RE = re.compile(r"[0-9a-f]{64}")
+_BRIDGE_CLEANUP_WARNING_RE = re.compile(
+ r"staging cleanup requires attention: [A-Za-z][A-Za-z0-9_]{0,63}"
+)
+_UTC_ISO_RE = re.compile(
+ r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(?:\.[0-9]{1,6})?\+00:00"
+)
class WorkerError(Exception):
- def __init__(self, code: str) -> None:
+ def __init__(self, code: str, diagnostic: dict[str, object] | None = None) -> None:
super().__init__(code)
self.code = code
+ self.diagnostic = diagnostic
class CancelRequested(BaseException):
@@ -116,13 +157,17 @@
@dataclass(frozen=True)
class HostConfig:
+ creator_allowlist: frozenset[str]
ffmpeg: Path
ffprobe: Path
bridge_python: Path
bridge_script: Path
- batch_json: Path
yt_dlp_executable: Path
destination: Path
+ queue_lock_path: Path | None = None
+ formal_manifest_path: Path | None = None
+ processing_handoff_path: Path | None = None
+ creator_name: str = ""
@staticmethod
def _safe_absolute_file(value: Any, expected_hash: Any) -> Path:
@@ -148,8 +193,13 @@
raise WorkerError("E_CONFIG") from exc
expected = {
"schema",
- "target",
- "canonical_url",
+ "creator_allowlist",
+ "queue_path",
+ "queue_state_path",
+ "queue_lock_path",
+ "reload_state_path",
+ "reload_generation",
+ "required_extension_build",
"ffmpeg",
"ffmpeg_sha256",
"ffprobe",
@@ -158,16 +208,86 @@
"bridge_python_sha256",
"bridge_script",
"bridge_script_sha256",
- "batch_json",
- "batch_json_sha256",
"yt_dlp_executable",
"yt_dlp_executable_sha256",
"destination",
+ "formal_manifest_path",
+ "processing_handoff_path",
+ "creator_name",
}
if not isinstance(raw, dict) or set(raw) != expected:
raise WorkerError("E_CONFIG")
- if raw["schema"] != 1 or raw["target"] != TARGET_BVID or raw["canonical_url"] != CANONICAL_URL:
+ if raw["schema"] != 2:
raise WorkerError("E_CONFIG")
+ if raw["required_extension_build"] != EXTENSION_BUILD or raw["reload_generation"] != RELOAD_GENERATION:
+ raise WorkerError("E_CONFIG")
+ creators = raw["creator_allowlist"]
+ try:
+ if (
+ not isinstance(creators, list) or not creators or len(creators) > 64
+ or [validate_creator_uid(item) for item in creators] != sorted(set(creators))
+ ):
+ raise WorkerError("E_CONFIG")
+ except ValueError as exc:
+ raise WorkerError("E_CONFIG") from exc
+ for name in (
+ "queue_path", "queue_state_path", "queue_lock_path", "reload_state_path",
+ "formal_manifest_path", "processing_handoff_path",
+ ):
+ value = raw[name]
+ if not isinstance(value, str):
+ raise WorkerError("E_CONFIG")
+ creator_name = raw["creator_name"]
+ if (
+ not isinstance(creator_name, str) or not creator_name.strip()
+ or len(creator_name.encode("utf-8")) > 240
+ or any(ord(ch) < 32 or ord(ch) == 127 for ch in creator_name)
+ ):
+ raise WorkerError("E_CONFIG")
+ queue_lock_path = Path(raw["queue_lock_path"])
+ formal_manifest_path = Path(raw["formal_manifest_path"])
+ processing_handoff_path = Path(raw["processing_handoff_path"])
+ governed_paths: list[Path] = []
+ for governed_path, must_exist in (
+ (queue_lock_path, False),
+ (formal_manifest_path, True),
+ (processing_handoff_path, False),
+ ):
+ if not governed_path.is_absolute() or str(governed_path).startswith("\\\\"):
+ raise WorkerError("E_CONFIG")
+ try:
+ lexical_parent = governed_path.parent
+ parent_stat = lexical_parent.lstat()
+ resolved_parent = lexical_parent.resolve(strict=True)
+ if (
+ not stat.S_ISDIR(parent_stat.st_mode)
+ or lexical_parent.is_symlink()
+ or _is_reparse(lexical_parent)
+ or resolved_parent != lexical_parent
+ ):
+ raise WorkerError("E_CONFIG")
+ resolved_path = resolved_parent / governed_path.name
+ try:
+ path_stat = governed_path.lstat()
+ except FileNotFoundError:
+ if must_exist:
+ raise WorkerError("E_CONFIG")
+ else:
+ if (
+ not stat.S_ISREG(path_stat.st_mode)
+ or governed_path.is_symlink()
+ or _is_reparse(governed_path)
+ or governed_path.resolve(strict=True) != resolved_path
+ ):
+ raise WorkerError("E_CONFIG")
+ except WorkerError:
+ raise
+ except OSError as exc:
+ raise WorkerError("E_CONFIG") from exc
+ governed_paths.append(resolved_path)
+ if len(set(governed_paths)) != len(governed_paths):
+ raise WorkerError("E_CONFIG")
+ queue_lock_path, formal_manifest_path, processing_handoff_path = governed_paths
bridge_script = cls._safe_absolute_file(raw["bridge_script"], raw["bridge_script_sha256"])
if raw["bridge_script_sha256"].upper() != FROZEN_BRIDGE_SHA256:
raise WorkerError("E_CONFIG_HASH")
@@ -178,17 +298,21 @@
if not destination.is_dir() or destination.is_symlink():
raise WorkerError("E_CONFIG")
return cls(
+ creator_allowlist=frozenset(creators),
ffmpeg=cls._safe_absolute_file(raw["ffmpeg"], raw["ffmpeg_sha256"]),
ffprobe=cls._safe_absolute_file(raw["ffprobe"], raw["ffprobe_sha256"]),
bridge_python=cls._safe_absolute_file(
raw["bridge_python"], raw["bridge_python_sha256"]
),
bridge_script=bridge_script,
- batch_json=cls._safe_absolute_file(raw["batch_json"], raw["batch_json_sha256"]),
yt_dlp_executable=cls._safe_absolute_file(
raw["yt_dlp_executable"], raw["yt_dlp_executable_sha256"]
),
destination=destination,
+ queue_lock_path=queue_lock_path,
+ formal_manifest_path=formal_manifest_path,
+ processing_handoff_path=processing_handoff_path,
+ creator_name=creator_name,
)
@@ -229,13 +353,17 @@
return resolved
-def fixed_stage_root() -> Path:
+def fixed_stage_root(bvid: str) -> Path:
local_app_data = validated_local_app_data()
+ try:
+ bvid = validate_bvid(bvid)
+ except ValueError as exc:
+ raise WorkerError("E_STAGE") from exc
logical_root = (
local_app_data
/ "project-info"
/ "bili-auth-ingress"
- / TARGET_BVID
+ / bvid
)
_reject_reparse_path(logical_root, local_app_data)
resolved = logical_root.resolve(strict=False)
@@ -258,7 +386,7 @@
def cleanup_stale_runs(root: Path, *, boundary: Path | None = None) -> None:
"""Remove only uncommitted run-* directories below the fixed stage root."""
- allowed_root = fixed_stage_root() if boundary is None else boundary.resolve()
+ allowed_root = root.resolve() if boundary is None else boundary.resolve()
_ensure_within(root, allowed_root)
if not root.exists():
return
@@ -270,8 +398,8 @@
shutil.rmtree(child)
-def create_run_directory(root: Path | None = None) -> Path:
- stage_root = fixed_stage_root() if root is None else root.resolve()
+def create_run_directory(root: Path) -> Path:
+ stage_root = root.resolve()
_ensure_within(stage_root, stage_root)
stage_root.mkdir(parents=True, exist_ok=True)
_reject_reparse_chain(stage_root, stage_root)
@@ -285,10 +413,10 @@
raise WorkerError("E_STAGE")
-def prepare_run_directory(root: Path | None = None) -> Path:
+def prepare_run_directory(root: Path) -> Path:
"""Clean stale runs and create the secret-free task lease."""
- stage_root = fixed_stage_root() if root is None else root.resolve()
- cleanup_stale_runs(stage_root, boundary=stage_root if root is not None else None)
+ stage_root = root.resolve()
+ cleanup_stale_runs(stage_root, boundary=stage_root)
return create_run_directory(stage_root)
@@ -345,8 +473,8 @@
return converted
-def validate_processed_info(info: Any) -> dict[str, Any]:
- if not isinstance(info, dict) or info.get("id") != TARGET_BVID:
+def validate_processed_info(info: Any, job: dict[str, Any]) -> dict[str, Any]:
+ if not isinstance(info, dict) or info.get("id") != job["bvid"]:
raise WorkerError("E_METADATA")
if info.get("entries") not in (None, []) or info.get("_type") not in (None, "video"):
raise WorkerError("E_MULTI_PART")
@@ -358,8 +486,11 @@
raise WorkerError("E_DRM")
if info.get("availability") not in (None, "public", "unlisted"):
raise WorkerError("E_ENTITLEMENT")
+ owner_ids = {str(value) for value in (info.get("uploader_id"), info.get("channel_id")) if value is not None}
+ if job["creator_uid"] not in owner_ids:
+ raise WorkerError("E_OWNER")
duration_ms = round(_finite_number(info.get("duration")) * 1000)
- if abs(duration_ms - EXPECTED_DURATION_MS) > DURATION_TOLERANCE_MS:
+ if abs(duration_ms - job["expected_duration_ms"]) > duration_tolerance_ms(job["expected_duration_ms"]):
raise WorkerError("E_DURATION")
formats = info.get("formats")
if not isinstance(formats, list) or not formats:
@@ -385,10 +516,11 @@
def validate_download_info(
download_info: Any,
params: dict[str, Any],
+ job: dict[str, Any],
*,
downloader_resolver: Callable[..., Any] | None = None,
) -> tuple[list[dict[str, Any]], bool]:
- if not isinstance(download_info, dict) or download_info.get("id") != TARGET_BVID:
+ if not isinstance(download_info, dict) or download_info.get("id") != job["bvid"]:
raise WorkerError("E_FORMAT")
leaves, single = _format_leaves(download_info)
if single:
@@ -426,6 +558,7 @@
def prepare_download_info(
ydl: Any,
+ job: dict[str, Any],
*,
downloader_resolver: Callable[..., Any] | None = None,
) -> tuple[dict[str, Any], bool, tuple[str, ...]]:
@@ -440,8 +573,8 @@
return original_extract(*args, **kwargs)
ydl.extract_info = one_extract
- processed = ydl.extract_info(CANONICAL_URL, download=False, process=True)
- validate_processed_info(processed)
+ processed = ydl.extract_info(job["canonical_url"], download=False, process=True)
+ validate_processed_info(processed, job)
selector = ydl.build_format_selector("bestvideo+bestaudio/best")
selected = list(ydl._select_formats(ydl._get_formats(processed), selector))
if len(selected) != 1:
@@ -451,6 +584,7 @@
leaves, single = validate_download_info(
download_info,
ydl.params,
+ job,
downloader_resolver=downloader_resolver,
)
signed_urls = tuple(str(leaf["url"]) for leaf in leaves)
@@ -467,6 +601,23 @@
class SubprocessPolicy:
"""Pre-CreateProcess audit for local-only child command lines."""
+ _FORBIDDEN = ("://", "-headers", "-cookies", "authorization", "cookie:", "referer:", "user-agent:")
+ _FFMPEG_FLAGS = frozenset({"-nostdin", "-y"})
+ _FFMPEG_SCALAR_OPTIONS = frozenset(
+ {"-v", "-loglevel", "-map", "-c", "-map_metadata", "-f", "-movflags"}
+ )
+ _FFMPEG_REPEATABLE_OPTIONS = frozenset({"-map"})
+ _FFPROBE_FLAGS = frozenset({"-hide_banner", "-show_format", "-show_streams"})
+ _FFPROBE_SCALAR_OPTIONS = frozenset({"-v", "-show_entries", "-of", "-print_format"})
+ _WINDOWS_RESERVED_NAMES = frozenset(
+ {"CON", "PRN", "AUX", "NUL", *(f"COM{index}" for index in range(1, 10)),
+ *(f"LPT{index}" for index in range(1, 10))}
+ )
+ _MAX_ARGUMENTS = 256
+ _MAX_COMMAND_LINE = 32_767
+ _MAX_ENVIRONMENT_ITEMS = 256
+ _MAX_LOCAL_FILE_OPERANDS = 16
+
def __init__(
self,
run_root: Path,
@@ -477,46 +628,269 @@
self.executables = {os.path.normcase(str(item.resolve())) for item in executables}
self.secrets = {item.casefold() for item in secrets if item}
+ @staticmethod
+ def _fail(code: str) -> None:
+ if code not in SUBPROCESS_POLICY_ERROR_CODES:
+ code = "E_SUBPROCESS_POLICY_EVENT_SHAPE"
+ raise WorkerError(code)
+
+ @classmethod
+ def _windows_arguments(cls, command_line: Any) -> list[str]:
+ if (
+ not isinstance(command_line, str)
+ or not command_line
+ or len(command_line) > cls._MAX_COMMAND_LINE
+ or "\x00" in command_line
+ ):
+ cls._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
+ argc = ctypes.c_int()
+ command_line_to_argv = ctypes.windll.shell32.CommandLineToArgvW
+ command_line_to_argv.argtypes = (ctypes.c_wchar_p, ctypes.POINTER(ctypes.c_int))
+ command_line_to_argv.restype = ctypes.POINTER(ctypes.c_wchar_p)
+ pointer = command_line_to_argv(command_line, ctypes.byref(argc))
+ if not pointer:
+ cls._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ try:
+ result = [pointer[index] for index in range(argc.value)]
+ finally:
+ local_free = ctypes.windll.kernel32.LocalFree
+ local_free.argtypes = (ctypes.c_void_p,)
+ local_free.restype = ctypes.c_void_p
+ local_free(ctypes.cast(pointer, ctypes.c_void_p))
+ if (
+ not result
+ or len(result) > cls._MAX_ARGUMENTS
+ or any(not isinstance(item, str) or "\x00" in item for item in result)
+ or subprocess.list2cmdline(result) != command_line
+ ):
+ cls._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ return result
+
+ @classmethod
+ def _portable_arguments(cls, raw: Any) -> list[str]:
+ if not isinstance(raw, (list, tuple)) or not raw or len(raw) > cls._MAX_ARGUMENTS:
+ cls._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
+ if any(type(item) is not str or "\x00" in item for item in raw):
+ cls._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ return list(raw)
+
+ @staticmethod
+ def _absolute_executable(value: Any) -> tuple[str, Path]:
+ if not isinstance(value, (str, os.PathLike)):
+ SubprocessPolicy._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
+ filesystem_value = os.fspath(value)
+ if not isinstance(filesystem_value, str) or not filesystem_value or "\x00" in filesystem_value:
+ SubprocessPolicy._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
+ path = Path(filesystem_value)
+ if not path.is_absolute() or str(path).startswith("\\\\"):
+ SubprocessPolicy._fail("E_SUBPROCESS_POLICY_EXECUTABLE")
+ try:
+ resolved = path.resolve(strict=True)
+ except OSError as exc:
+ raise WorkerError("E_SUBPROCESS_POLICY_EXECUTABLE") from exc
+ if not resolved.is_file() or resolved.is_symlink() or _is_reparse(resolved):
+ SubprocessPolicy._fail("E_SUBPROCESS_POLICY_EXECUTABLE")
+ return os.path.normcase(str(resolved)), resolved
+
+ def _validate_environment(self, environment: Any) -> None:
+ if environment is None:
+ return
+ if not isinstance(environment, Mapping) or len(environment) > self._MAX_ENVIRONMENT_ITEMS:
+ self._fail("E_SUBPROCESS_POLICY_ENVIRONMENT")
+ encoded: list[str] = []
+ for key, value in environment.items():
+ if (
+ type(key) is not str
+ or type(value) is not str
+ or not key
+ or "\x00" in key
+ or "\x00" in value
+ ):
+ self._fail("E_SUBPROCESS_POLICY_ENVIRONMENT")
+ encoded.append(f"{key}={value}")
+ folded = "\x00".join(encoded).casefold()
+ if any(item in folded for item in self._FORBIDDEN) or any(item in folded for item in self.secrets):
+ self._fail("E_SUBPROCESS_POLICY_ENVIRONMENT")
+
def check(self, event: str, arguments: tuple[Any, ...]) -> None:
if event != "subprocess.Popen":
return
- executable, argv, _cwd, environment = arguments
- if not isinstance(executable, (str, os.PathLike)) or not isinstance(argv, (list, tuple)):
- raise WorkerError("E_SUBPROCESS_POLICY")
- executable_key = os.path.normcase(str(Path(executable).resolve()))
+ if not isinstance(arguments, tuple) or len(arguments) != 4:
+ self._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
+ executable, raw_arguments, cwd, environment = arguments
+ if cwd is not None:
+ self._fail("E_SUBPROCESS_POLICY_EVENT_SHAPE")
+ text_args = (
+ self._windows_arguments(raw_arguments)
+ if os.name == "nt"
+ else self._portable_arguments(raw_arguments)
+ )
+ argv_executable_key, _ = self._absolute_executable(text_args[0])
+ if executable is None:
+ executable_key = argv_executable_key
+ else:
+ executable_key, _ = self._absolute_executable(executable)
+ if executable_key != argv_executable_key:
+ self._fail("E_SUBPROCESS_POLICY_EXECUTABLE")
if executable_key not in self.executables:
- raise WorkerError("E_SUBPROCESS_POLICY")
- text_args = [str(item) for item in argv]
+ self._fail("E_SUBPROCESS_POLICY_EXECUTABLE")
folded = "\x00".join(text_args).casefold()
- forbidden = ("://", "-headers", "-cookies", "authorization", "cookie:", "referer:", "user-agent:")
- if any(item in folded for item in forbidden) or any(item in folded for item in self.secrets):
- raise WorkerError("E_SUBPROCESS_POLICY")
- if environment is not None:
- encoded_env = "\x00".join(f"{key}={value}" for key, value in environment.items()).casefold()
- if any(item in encoded_env for item in self.secrets):
- raise WorkerError("E_SUBPROCESS_POLICY")
- if executable_key.endswith("ffmpeg.exe") or executable_key.endswith("ffprobe.exe"):
- for index, argument in enumerate(text_args[:-1]):
- if argument == "-i":
- self._local_path(text_args[index + 1], must_exist=True)
- output = text_args[-1]
- if output not in {"-", "NUL"} and not output.startswith("-"):
- self._local_path(output, must_exist=False)
+ if any(item in folded for item in self._FORBIDDEN) or any(item in folded for item in self.secrets):
+ self._fail("E_SUBPROCESS_POLICY_SECRET")
+ self._validate_environment(environment)
+ executable_name = Path(executable_key).name.casefold()
+ if executable_name == "ffmpeg.exe":
+ self._validate_ffmpeg_arguments(text_args)
+ elif executable_name == "ffprobe.exe":
+ self._validate_ffprobe_arguments(text_args)
+
+ @staticmethod
+ def _valid_ffmpeg_scalar(option: str, value: str) -> bool:
+ if option == "-v":
+ return value == "error"
+ if option == "-loglevel":
+ return value == "repeat+info"
+ if option == "-map":
+ return re.fullmatch(r"\d+(?::[av](?::\d+)?)?", value) is not None
+ if option == "-c":
+ return value == "copy"
+ if option == "-map_metadata":
+ return value == "-1"
+ if option == "-f":
+ return value == "matroska"
+ if option == "-movflags":
+ return value == "+faststart"
+ return False
+
+ @staticmethod
+ def _valid_ffprobe_scalar(option: str, value: str) -> bool:
+ if option == "-v":
+ return value == "error"
+ if option == "-show_entries":
+ return value == "format=format_name,duration:stream=codec_type"
+ if option in {"-of", "-print_format"}:
+ return value == "json"
+ return False
+
+ def _validate_ffmpeg_arguments(self, arguments: Sequence[str]) -> None:
+ if tuple(arguments[1:]) == ("-bsfs",):
+ return
+ index = 1
+ input_count = 0
+ file_operand_count = 0
+ output_seen = False
+ seen_options: set[str] = set()
+ while index < len(arguments):
+ option = arguments[index]
+ if option in self._FFMPEG_FLAGS:
+ if option in seen_options:
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ seen_options.add(option)
+ index += 1
+ continue
+ if option in {"-i", "-attach"}:
+ if index + 1 >= len(arguments) or arguments[index + 1].startswith("-"):
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ self._local_path(arguments[index + 1], must_exist=True)
+ file_operand_count += 1
+ input_count += option == "-i"
+ if file_operand_count > self._MAX_LOCAL_FILE_OPERANDS:
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ index += 2
+ continue
+ if option in self._FFMPEG_SCALAR_OPTIONS:
+ if index + 1 >= len(arguments):
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ value = arguments[index + 1]
+ if not self._valid_ffmpeg_scalar(option, value):
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ if option not in self._FFMPEG_REPEATABLE_OPTIONS:
+ if option in seen_options:
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ seen_options.add(option)
+ index += 2
+ continue
+ if re.fullmatch(r"-bsf:a:\d+", option):
+ if index + 1 >= len(arguments) or arguments[index + 1] != "aac_adtstoasc":
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ if option in seen_options:
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ seen_options.add(option)
+ index += 2
+ continue
+ if option.startswith("-") or output_seen or index != len(arguments) - 1:
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ self._local_path(option, must_exist=False)
+ output_seen = True
+ file_operand_count += 1
+ index += 1
+ if input_count < 1 or not output_seen or file_operand_count > self._MAX_LOCAL_FILE_OPERANDS:
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+
+ def _validate_ffprobe_arguments(self, arguments: Sequence[str]) -> None:
+ if tuple(arguments[1:]) == ("-bsfs",):
+ return
+ index = 1
+ input_seen = False
+ seen_options: set[str] = set()
+ while index < len(arguments):
+ option = arguments[index]
+ if option in self._FFPROBE_FLAGS:
+ if option in seen_options:
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ seen_options.add(option)
+ index += 1
+ continue
+ if option in self._FFPROBE_SCALAR_OPTIONS:
+ if index + 1 >= len(arguments) or not self._valid_ffprobe_scalar(
+ option, arguments[index + 1]
+ ):
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ if option in seen_options:
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ seen_options.add(option)
+ index += 2
+ continue
+ if option.startswith("-") or input_seen or index != len(arguments) - 1:
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
+ self._local_path(option, must_exist=True)
+ input_seen = True
+ index += 1
+ if not input_seen:
+ self._fail("E_SUBPROCESS_POLICY_ARGUMENTS")
def _local_path(self, value: str, *, must_exist: bool) -> Path:
+ if value.startswith("file:"):
+ value = value[5:]
path = Path(value)
- if not path.is_absolute() or str(path).startswith("\\\\"):
- raise WorkerError("E_SUBPROCESS_POLICY")
+ if (
+ not value
+ or not path.is_absolute()
+ or str(path).startswith("\\\\")
+ or ":" in value[2:]
+ or any(part == ".." or part.endswith((" ", ".")) for part in path.parts)
+ or path.name.split(".", 1)[0].upper() in self._WINDOWS_RESERVED_NAMES
+ ):
+ self._fail("E_SUBPROCESS_POLICY_LOCAL_PATH")
try:
- resolved = path.resolve(strict=must_exist)
- except OSError as exc:
- raise WorkerError("E_SUBPROCESS_POLICY") from exc
- try:
- resolved.relative_to(self.run_root)
- except ValueError as exc:
- raise WorkerError("E_SUBPROCESS_POLICY") from exc
- if must_exist and (not resolved.is_file() or resolved.is_symlink()):
- raise WorkerError("E_SUBPROCESS_POLICY")
+ path.relative_to(self.run_root)
+ _reject_reparse_chain(path, self.run_root)
+ if must_exist:
+ resolved = path.resolve(strict=True)
+ resolved.relative_to(self.run_root)
+ if not resolved.is_file() or resolved.is_symlink() or _is_reparse(resolved):
+ self._fail("E_SUBPROCESS_POLICY_LOCAL_PATH")
+ else:
+ if path.exists() or path.is_symlink():
+ self._fail("E_SUBPROCESS_POLICY_LOCAL_PATH")
+ parent = path.parent.resolve(strict=True)
+ parent.relative_to(self.run_root)
+ _reject_reparse_chain(path.parent, self.run_root)
+ if not parent.is_dir() or parent.is_symlink() or _is_reparse(parent):
+ self._fail("E_SUBPROCESS_POLICY_LOCAL_PATH")
+ resolved = parent / path.name
+ except (OSError, ValueError, WorkerError) as exc:
+ raise WorkerError("E_SUBPROCESS_POLICY_LOCAL_PATH") from exc
return resolved
def install(self) -> None:
@@ -619,7 +993,7 @@
raise WorkerError("E_MERGE")
-def probe_mkv(ffprobe: Path, candidate: Path) -> None:
+def probe_mkv(ffprobe: Path, candidate: Path, job: dict[str, Any] | None = None) -> None:
try:
result = _run_local(
[
@@ -627,7 +1001,7 @@
"-v",
"error",
"-show_entries",
- "format=format_name:stream=codec_type",
+ "format=format_name,duration:stream=codec_type",
"-of",
"json",
str(candidate),
@@ -645,6 +1019,13 @@
format_name = payload.get("format", {}).get("format_name", "")
if result.returncode != 0 or {"video", "audio"} - stream_types or "matroska" not in format_name:
raise WorkerError("E_MEDIA_VALIDATION")
+ if job is not None:
+ raw_duration = payload.get("format", {}).get("duration")
+ if isinstance(raw_duration, str) and re.fullmatch(r"[0-9]+(?:\.[0-9]+)?", raw_duration):
+ raw_duration = float(raw_duration)
+ duration_ms = round(_finite_number(raw_duration) * 1000)
+ if abs(duration_ms - job["expected_duration_ms"]) > duration_tolerance_ms(job["expected_duration_ms"]):
+ raise WorkerError("E_DURATION")
def validate_unique_candidate(run_directory: Path) -> Path:
@@ -658,96 +1039,981 @@
return candidates[0]
-def _bridge_command(config: HostConfig, candidate: Path) -> list[str]:
+def _ordinary_exact_file(path: Path, parent: Path) -> os.stat_result:
+ try:
+ if path.parent.resolve(strict=True) != parent.resolve(strict=True):
+ raise WorkerError("E_COLLISION")
+ _reject_reparse_path(path, parent)
+ value = path.lstat()
+ except OSError as exc:
+ raise WorkerError("E_COLLISION") from exc
+ if not stat.S_ISREG(value.st_mode) or _is_reparse(path):
+ raise WorkerError("E_COLLISION")
+ return value
+
+
+def _stable_file_identity(value: os.stat_result) -> tuple[int, ...]:
+ """Return cross-API file identity fields; content is bound separately by SHA."""
+ return (
+ value.st_dev,
+ value.st_ino,
+ value.st_mode,
+ value.st_nlink,
+ value.st_size,
+ )
+
+
+def _stable_pair_test_seam(_: str) -> None:
+ """Named no-op seams used only by production-shaped race regressions."""
+ return None
+
+
+def _completion_test_seam(_: str) -> None:
+ """Named no-op seams for durable completion transaction regressions."""
+ return None
+
+
+def _canonical_json_line(value: Mapping[str, Any]) -> bytes:
+ return json.dumps(
+ dict(value), ensure_ascii=False, allow_nan=False, separators=(",", ":")
+ ).encode("utf-8") + b"\n"
+
+
+_FORMAL_PRIOR_REQUIRED_KEYS = frozenset({
+ "stable_id", "creator_uid", "source_url", "published_at", "item_type", "status",
+})
+_FORMAL_PRIOR_STATUS_RE = re.compile(r"VIDEO_[A-Z0-9_]{1,127}\Z")
+_FORMAL_LEGACY_BY_LINE = {row[0]: row for row in FORMAL_LEGACY_ROWS}
+_FORMAL_LEGACY_INTEGER_UID_BY_LINE = {
+ row[0]: row for row in FORMAL_LEGACY_INTEGER_UID_ROWS
+}
+
+
+def _formal_raw_lines(payload: bytes) -> list[bytes]:
+ complete = payload if not payload or payload.endswith(b"\n") else payload[:payload.rfind(b"\n") + 1]
+ return complete.splitlines()
+
+
+def _validate_formal_legacy_manifest() -> None:
+ if (
+ FORMAL_PREFIX_BYTES != 103_766
+ or FORMAL_PREFIX_LINES != 119
+ or not re.fullmatch(r"[A-F0-9]{64}", FORMAL_PREFIX_SHA256)
+ or not re.fullmatch(r"[1-9][0-9]{1,19}", FORMAL_LEGACY_CREATOR_UID)
+ or len(FORMAL_LEGACY_ROWS) != 26
+ or len(_FORMAL_LEGACY_BY_LINE) != len(FORMAL_LEGACY_ROWS)
+ or len(FORMAL_LEGACY_INTEGER_UID_ROWS) != 1
+ or len(_FORMAL_LEGACY_INTEGER_UID_BY_LINE)
+ != len(FORMAL_LEGACY_INTEGER_UID_ROWS)
+ or set(_FORMAL_LEGACY_BY_LINE) & set(_FORMAL_LEGACY_INTEGER_UID_BY_LINE)
+ ):
+ raise WorkerError("E_COMPLETION_FORMAL")
+ for row in FORMAL_LEGACY_ROWS:
+ if (
+ not isinstance(row, tuple) or len(row) != 8
+ or not isinstance(row[0], int) or not 1 <= row[0] <= FORMAL_PREFIX_LINES
+ or not isinstance(row[1], int) or row[1] < 2
+ or not isinstance(row[2], str) or not re.fullmatch(r"[A-F0-9]{64}", row[2])
+ or not all(isinstance(value, str) and value for value in row[3:])
+ or row[6] != "video" or _FORMAL_PRIOR_STATUS_RE.fullmatch(row[7]) is None
+ ):
+ raise WorkerError("E_COMPLETION_FORMAL")
+ for row in FORMAL_LEGACY_INTEGER_UID_ROWS:
+ if (
+ not isinstance(row, tuple) or len(row) != 9
+ or type(row[0]) is not int or not 1 <= row[0] <= FORMAL_PREFIX_LINES
+ or type(row[1]) is not int or row[1] < 2
+ or not isinstance(row[2], str) or not re.fullmatch(r"[A-F0-9]{64}", row[2])
+ or type(row[3]) is not int or row[3] <= 0
+ or str(row[3]) != FORMAL_LEGACY_CREATOR_UID
+ or not all(isinstance(value, str) and value for value in row[4:])
+ or row[7] != "video" or _FORMAL_PRIOR_STATUS_RE.fullmatch(row[8]) is None
+ ):
+ raise WorkerError("E_COMPLETION_FORMAL")
+
+
+def _validate_legacy_formal_prior(
+ payload: bytes,
+ raw_lines: Sequence[bytes],
+ line_ordinal: int,
+ value: dict[str, Any],
+ config: HostConfig,
+ job: dict[str, Any],
+) -> str:
+ _validate_formal_legacy_manifest()
+ if (
+ len(payload) < FORMAL_PREFIX_BYTES
+ or len(raw_lines) < FORMAL_PREFIX_LINES
+ or hashlib.sha256(payload[:FORMAL_PREFIX_BYTES]).hexdigest().upper() != FORMAL_PREFIX_SHA256
+ or payload[:FORMAL_PREFIX_BYTES].count(b"\n") != FORMAL_PREFIX_LINES
+ or not payload[:FORMAL_PREFIX_BYTES].endswith(b"\n")
+ or job["creator_uid"] != FORMAL_LEGACY_CREATOR_UID
+ ):
+ raise WorkerError("E_COMPLETION_FORMAL")
+ expected = _FORMAL_LEGACY_BY_LINE.get(line_ordinal)
+ if expected is None:
+ raise WorkerError("E_COMPLETION_FORMAL")
+ raw_line = raw_lines[line_ordinal - 1]
+ if (
+ len(raw_line) != expected[1]
+ or hashlib.sha256(raw_line).hexdigest().upper() != expected[2]
+ or set(_FORMAL_PRIOR_REQUIRED_KEYS) - set(value) != {"creator_uid"}
+ or value.get("stable_id") != expected[3]
+ or value.get("source_url") != expected[4]
+ or value.get("published_at") != expected[5]
+ or value.get("item_type") != expected[6]
+ or value.get("status") != expected[7]
+ or value.get("schema_version") != 1
+ or value.get("creator") != config.creator_name
+ or expected[3] != job["bvid"]
+ or expected[4] != job["canonical_url"]
+ or expected[5] != job["published_at"]
+ ):
+ raise WorkerError("E_COMPLETION_FORMAL")
+ return expected[7]
+
+
+def _validate_current_formal_prior(
+ value: dict[str, Any], config: HostConfig, job: dict[str, Any]
+) -> str:
+ if (
+ not _FORMAL_PRIOR_REQUIRED_KEYS.issubset(value)
+ or type(value.get("schema_version")) is not int
+ or value["schema_version"] != 1
+ or value.get("creator") != config.creator_name
+ or not isinstance(value.get("stable_id"), str)
+ or not isinstance(value.get("creator_uid"), str)
+ or not isinstance(value.get("source_url"), str)
+ or not isinstance(value.get("published_at"), str)
+ or not isinstance(value.get("item_type"), str)
+ or not isinstance(value.get("status"), str)
+ or value["stable_id"] != job["bvid"]
+ or value["creator_uid"] != job["creator_uid"]
+ or value["source_url"] != job["canonical_url"]
+ or value["published_at"] != job["published_at"]
+ or value["item_type"] != "video"
+ or _FORMAL_PRIOR_STATUS_RE.fullmatch(value["status"]) is None
+ ):
+ raise WorkerError("E_COMPLETION_FORMAL")
+ return value["status"]
+
+
+def _validate_legacy_integer_uid_formal_prior(
+ payload: bytes,
+ raw_lines: Sequence[bytes],
+ line_ordinal: int,
+ value: dict[str, Any],
+ config: HostConfig,
+ job: dict[str, Any],
+) -> str:
+ _validate_formal_legacy_manifest()
+ if (
+ len(payload) < FORMAL_PREFIX_BYTES
+ or len(raw_lines) < FORMAL_PREFIX_LINES
+ or hashlib.sha256(payload[:FORMAL_PREFIX_BYTES]).hexdigest().upper()
+ != FORMAL_PREFIX_SHA256
+ or payload[:FORMAL_PREFIX_BYTES].count(b"\n") != FORMAL_PREFIX_LINES
+ or not payload[:FORMAL_PREFIX_BYTES].endswith(b"\n")
+ ):
+ raise WorkerError("E_COMPLETION_FORMAL")
+ expected = _FORMAL_LEGACY_INTEGER_UID_BY_LINE.get(line_ordinal)
+ if expected is None:
+ raise WorkerError("E_COMPLETION_FORMAL")
+ raw_line = raw_lines[line_ordinal - 1]
+ if (
+ len(raw_line) != expected[1]
+ or hashlib.sha256(raw_line).hexdigest().upper() != expected[2]
+ or not _FORMAL_PRIOR_REQUIRED_KEYS.issubset(value)
+ or type(value.get("schema_version")) is not int
+ or value["schema_version"] != 1
+ or value.get("creator") != config.creator_name
+ or type(value.get("creator_uid")) is not int
+ or value["creator_uid"] != expected[3]
+ or value.get("stable_id") != expected[4]
+ or value.get("source_url") != expected[5]
+ or value.get("published_at") != expected[6]
+ or value.get("item_type") != expected[7]
+ or value.get("status") != expected[8]
+ or str(expected[3]) != job["creator_uid"]
+ or expected[4] != job["bvid"]
+ or expected[5] != job["canonical_url"]
+ or expected[6] != job["published_at"]
+ ):
+ raise WorkerError("E_COMPLETION_FORMAL")
+ return expected[8]
+
+
+def _read_jsonl_objects(path: Path, *, limit: int, error_code: str) -> tuple[bytes, list[dict[str, Any]]]:
+ try:
+ if path.exists():
+ parent = path.parent.resolve(strict=True)
+ _ordinary_exact_file(path, parent)
+ payload = path.read_bytes()
+ else:
+ payload = b""
+ if len(payload) > limit:
+ raise WorkerError(error_code)
+ complete = payload if not payload or payload.endswith(b"\n") else payload[:payload.rfind(b"\n") + 1]
+ records: list[dict[str, Any]] = []
+ for raw_line in complete.splitlines():
+ if not raw_line:
+ raise WorkerError(error_code)
+ value = strict_json_loads(raw_line)
+ records.append(value)
+ return payload, records
+ except WorkerError:
+ raise
+ except (OSError, ProtocolError, UnicodeError, json.JSONDecodeError) as exc:
+ raise WorkerError(error_code) from exc
+
+
+def _append_jsonl_idempotent(
+ path: Path,
+ record: dict[str, Any],
+ *,
+ identity_key: str,
+ identity_value: str,
+ limit: int,
+ error_code: str,
+) -> None:
+ expected = _canonical_json_line(record)
+ payload, records = _read_jsonl_objects(path, limit=limit, error_code=error_code)
+ matches = [value for value in records if value.get(identity_key) == identity_value]
+ if len(matches) > 1 or (matches and matches[0] != record):
+ raise WorkerError("E_COMPLETION_REPLAY")
+ if matches:
+ return
+ suffix = b"" if not payload or payload.endswith(b"\n") else payload[payload.rfind(b"\n") + 1:]
+ if suffix and (len(suffix) >= len(expected) or expected[:len(suffix)] != suffix):
+ raise WorkerError(error_code)
+ parent = path.parent.resolve(strict=True)
+ try:
+ if not path.exists():
+ with path.open("xb") as created:
+ created.flush()
+ os.fsync(created.fileno())
+ _ordinary_exact_file(path, parent)
+ with path.open("r+b", buffering=0) as stream:
+ current = stream.read()
+ if current != payload:
+ raise WorkerError(error_code)
+ stream.seek(0, os.SEEK_END)
+ stream.write(expected[len(suffix):])
+ stream.flush()
+ os.fsync(stream.fileno())
+ _completion_test_seam(f"AFTER_{identity_key.upper()}_APPEND")
+ final_payload, final_records = _read_jsonl_objects(path, limit=limit, error_code=error_code)
+ if not final_payload.endswith(b"\n") or sum(
+ value.get(identity_key) == identity_value and value == record for value in final_records
+ ) != 1:
+ raise WorkerError(error_code)
+ except WorkerError:
+ raise
+ except OSError as exc:
+ raise WorkerError(error_code) from exc
+
+
+@contextmanager
+def _completion_lock(path: Path) -> Iterable[None]:
+ try:
+ path.parent.resolve(strict=True)
+ with path.open("a+b") as stream:
+ if stream.seek(0, os.SEEK_END) == 0:
+ stream.write(b"\0")
+ stream.flush()
+ os.fsync(stream.fileno())
+ _ordinary_exact_file(path, path.parent)
+ stream.seek(0)
+ if os.name == "nt":
+ import msvcrt # noqa: PLC0415
+
+ msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1)
+ try:
+ yield
+ finally:
+ stream.seek(0)
+ msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1)
+ else:
+ import fcntl # noqa: PLC0415
+
+ fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+ try:
+ yield
+ finally:
+ fcntl.flock(stream.fileno(), fcntl.LOCK_UN)
+ except WorkerError:
+ raise
+ except OSError as exc:
+ raise WorkerError("E_COMPLETION_BUSY") from exc
+
+
+def _commit_formal_and_handoff(
+ config: HostConfig,
+ job: dict[str, Any],
+ formal_name: str,
+ mapping_name: str,
+ persisted: dict[str, Any],
+ *,
+ media_complete_acknowledged: bool = False,
+) -> None:
+ """Idempotently close formal publication and processing handoff before COMPLETE."""
+ if (
+ not isinstance(config.queue_lock_path, Path)
+ or not isinstance(config.formal_manifest_path, Path)
+ or not isinstance(config.processing_handoff_path, Path)
+ or not isinstance(config.creator_name, str) or not config.creator_name
+ ):
+ raise WorkerError("E_COMPLETION_CONFIG")
+ handoff_id = f"HANDOFF-BILI-MEDIA-{job['job_id'][:32].upper()}"
+ with _completion_lock(config.queue_lock_path):
+ formal_payload, formal_records = _read_jsonl_objects(
+ config.formal_manifest_path, limit=32 * 1024 * 1024, error_code="E_COMPLETION_FORMAL"
+ )
+ raw_lines = _formal_raw_lines(formal_payload)
+ if len(raw_lines) != len(formal_records):
+ raise WorkerError("E_COMPLETION_FORMAL")
+ prior_status = None
+ seen_prior_identities: set[tuple[str, str, str, str, str, str]] = set()
+ for line_ordinal, value in enumerate(formal_records, 1):
+ if value.get("queue_job_id") == job["job_id"]:
+ continue
+ stable_matches = value.get("stable_id") == job["bvid"]
+ source_matches = value.get("source_url") == job["canonical_url"]
+ if not stable_matches and not source_matches:
+ continue
+ item_type = value.get("item_type")
+ if item_type == "video_transcript" and stable_matches and source_matches:
+ continue
+ if not stable_matches or not source_matches:
+ raise WorkerError("E_COMPLETION_FORMAL")
+ missing = _FORMAL_PRIOR_REQUIRED_KEYS - set(value)
+ if line_ordinal in _FORMAL_LEGACY_INTEGER_UID_BY_LINE:
+ prior_status = _validate_legacy_integer_uid_formal_prior(
+ formal_payload, raw_lines, line_ordinal, value, config, job
+ )
+ creator_uid = job["creator_uid"]
+ elif missing == {"creator_uid"}:
+ prior_status = _validate_legacy_formal_prior(
+ formal_payload, raw_lines, line_ordinal, value, config, job
+ )
+ creator_uid = job["creator_uid"]
+ else:
+ prior_status = _validate_current_formal_prior(value, config, job)
+ creator_uid = value["creator_uid"]
+ prior_identity = (
+ value["stable_id"], creator_uid, value["source_url"],
+ value["published_at"], value["item_type"], prior_status,
+ )
+ if prior_identity in seen_prior_identities:
+ raise WorkerError("E_COMPLETION_FORMAL")
+ seen_prior_identities.add(prior_identity)
+ if prior_status is None and media_complete_acknowledged is not True:
+ # A first formal row has no historical status to supersede. It is
+ # permitted only after the typed Host ACK proves the governed queue
+ # job's exact MEDIA_COMPLETE identity is already durable.
+ raise WorkerError("E_COMPLETION_FORMAL")
+ formal_record = {
+ "schema_version": 1,
+ "creator": config.creator_name,
+ "creator_uid": job["creator_uid"],
+ "item_type": "video",
+ "stable_id": job["bvid"],
+ "title": job["title"],
+ "source_url": job["canonical_url"],
+ "published_at": job["published_at"],
+ "collected_at": persisted["completed_at"],
+ "status": "VIDEO_DOWNLOADED_COMPLETE_HANDOFF_SENT",
+ "video_path": str(config.destination / formal_name),
+ "mapping_path": str(config.destination / mapping_name),
+ "bytes": persisted["bytes"],
+ "sha256": persisted["sha256"],
+ "duration_seconds": persisted["duration_seconds"],
+ "video_codec": persisted["video_codec"],
+ "audio_codec": persisted["audio_codec"],
+ "processing_handoff_id": handoff_id,
+ "queue_job_id": job["job_id"],
+ }
+ if prior_status is not None:
+ formal_record["supersedes_status"] = prior_status
+ handoff_record = {
+ "schema": 1,
+ "type": "media-processing-handoff",
+ "status": "READY",
+ "handoff_id": handoff_id,
+ "queue_job_id": job["job_id"],
+ "creator_uid": job["creator_uid"],
+ "bvid": job["bvid"],
+ "source_url": job["canonical_url"],
+ "media_path": str(config.destination / formal_name),
+ "mapping_path": str(config.destination / mapping_name),
+ "bytes": persisted["bytes"],
+ "sha256": persisted["sha256"],
+ "duration_seconds": persisted["duration_seconds"],
+ "video_codec": persisted["video_codec"],
+ "audio_codec": persisted["audio_codec"],
+ "created_at": persisted["completed_at"],
+ }
+ _append_jsonl_idempotent(
+ config.formal_manifest_path, formal_record,
+ identity_key="queue_job_id", identity_value=job["job_id"],
+ limit=32 * 1024 * 1024, error_code="E_COMPLETION_FORMAL",
+ )
+ _completion_test_seam("BETWEEN_FORMAL_AND_HANDOFF")
+ _append_jsonl_idempotent(
+ config.processing_handoff_path, handoff_record,
+ identity_key="queue_job_id", identity_value=job["job_id"],
+ limit=8 * 1024 * 1024, error_code="E_COMPLETION_HANDOFF",
+ )
+ _completion_test_seam("BEFORE_COMPLETION_RETURN")
+
+
+@dataclass
+class _LockedPublishedFile:
+ """Read-only handle whose sharing mode denies writers, deletion and replacement."""
+
+ path: Path
+ parent: Path
+ error_code: str
+ stream: Any
+ identity: tuple[int, ...]
+
+ @classmethod
+ def open(cls, path: Path, parent: Path, error_code: str) -> "_LockedPublishedFile":
+ stream: Any | None = None
+ try:
+ before = _ordinary_exact_file(path, parent)
+ if os.name != "nt":
+ # The deployed Host is Windows-only. Keep non-Windows imports
+ # fail-closed while retaining a no-follow advisory read lock.
+ import fcntl # noqa: PLC0415
+
+ flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0)
+ descriptor = os.open(path, flags)
+ try:
+ fcntl.flock(descriptor, fcntl.LOCK_SH | fcntl.LOCK_NB)
+ stream = os.fdopen(descriptor, "rb", closefd=True)
+ except BaseException:
+ os.close(descriptor)
+ raise
+ else:
+ import msvcrt # noqa: PLC0415
+
+ kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+ create_file = kernel32.CreateFileW
+ create_file.argtypes = (
+ ctypes.c_wchar_p, ctypes.c_uint32, ctypes.c_uint32, ctypes.c_void_p,
+ ctypes.c_uint32, ctypes.c_uint32, ctypes.c_void_p,
+ )
+ create_file.restype = ctypes.c_void_p
+ close_handle = kernel32.CloseHandle
+ close_handle.argtypes = (ctypes.c_void_p,)
+ close_handle.restype = ctypes.c_int
+ handle = create_file(
+ str(path),
+ 0x80000000, # GENERIC_READ
+ 0x00000001, # FILE_SHARE_READ: deny write/delete/path replacement
+ None,
+ 3, # OPEN_EXISTING
+ 0x00000080 | 0x00200000 | 0x08000000,
+ # FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OPEN_REPARSE_POINT |
+ # FILE_FLAG_SEQUENTIAL_SCAN
+ None,
+ )
+ if handle in (None, ctypes.c_void_p(-1).value):
+ code = ctypes.get_last_error()
+ raise OSError(code, ctypes.FormatError(code), str(path))
+ descriptor: int | None = None
+ try:
+ descriptor = msvcrt.open_osfhandle(
+ int(handle), os.O_RDONLY | getattr(os, "O_BINARY", 0)
+ )
+ handle = None
+ stream = os.fdopen(descriptor, "rb", closefd=True)
+ descriptor = None
+ finally:
+ if descriptor is not None:
+ os.close(descriptor)
+ if handle is not None:
+ close_handle(handle)
+ after = _ordinary_exact_file(path, parent)
+ identity = _stable_file_identity(os.fstat(stream.fileno()))
+ if identity != _stable_file_identity(before) or identity != _stable_file_identity(after):
+ raise WorkerError(error_code)
+ return cls(path=path, parent=parent, error_code=error_code, stream=stream, identity=identity)
+ except BaseException as exc:
+ if stream is not None:
+ stream.close()
+ if isinstance(exc, WorkerError) and exc.code == error_code:
+ raise
+ if isinstance(exc, (KeyboardInterrupt, SystemExit)):
+ raise
+ raise WorkerError(error_code) from exc
+
+ def close(self) -> None:
+ self.stream.close()
+
+ def assert_path_identity(self) -> os.stat_result:
+ try:
+ path_stat = _ordinary_exact_file(self.path, self.parent)
+ handle_stat = os.fstat(self.stream.fileno())
+ except (OSError, WorkerError) as exc:
+ raise WorkerError(self.error_code) from exc
+ if (
+ _stable_file_identity(path_stat) != self.identity
+ or _stable_file_identity(handle_stat) != self.identity
+ ):
+ raise WorkerError(self.error_code)
+ return handle_stat
+
+ def read_all(self, maximum: int) -> bytes:
+ try:
+ self.stream.seek(0)
+ payload = self.stream.read(maximum + 1)
+ if len(payload) > maximum or self.stream.read(1) != b"":
+ raise WorkerError(self.error_code)
+ self.assert_path_identity()
+ return payload
+ except WorkerError:
+ raise
+ except OSError as exc:
+ raise WorkerError(self.error_code) from exc
+
+ def sha256(self) -> str:
+ digest = hashlib.sha256()
+ try:
+ self.stream.seek(0)
+ for chunk in iter(lambda: self.stream.read(1024 * 1024), b""):
+ digest.update(chunk)
+ self.assert_path_identity()
+ return digest.hexdigest()
+ except WorkerError:
+ raise
+ except OSError as exc:
+ raise WorkerError(self.error_code) from exc
+
+
+def _bridge_number(value: Any) -> float:
+ if isinstance(value, bool) or not isinstance(value, (int, float)):
+ raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
+ number = float(value)
+ if not math.isfinite(number) or number < 0:
+ raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
+ return number
+
+
+def _bridge_remote_matches_expected(expected_duration_ms: int, remote: float) -> bool:
+ expected = expected_duration_ms / 1000
+ if abs(remote - expected) <= 0.001:
+ return True
+ return (
+ expected_duration_ms % 1000 == 0
+ and 0 < expected - remote < 1
+ and math.ceil(remote) == int(expected)
+ )
+
+
+def _validate_complete_bridge_item(
+ item: Any,
+ job: dict[str, Any],
+ *,
+ persisted: bool,
+) -> dict[str, Any]:
+ if not isinstance(item, dict):
+ raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
+ expected_keys = _BRIDGE_MAPPING_KEYS if persisted else _BRIDGE_ITEM_KEYS
+ if set(item) != expected_keys:
+ if persisted or set(item) != _BRIDGE_ITEM_WARNING_KEYS:
+ raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
+ warning = item["warning"]
+ if (
+ not isinstance(warning, str)
+ or _BRIDGE_CLEANUP_WARNING_RE.fullmatch(warning) is None
+ ):
+ raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
+ if not persisted and item["status"] != "COMPLETE":
+ raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
+ if (
+ item["schema_version"] != "1.0"
+ or item["bvid"] != job["bvid"]
+ or item["source"] != job["canonical_url"]
+ or item["published_at"] != job["published_at"]
+ or item["local_file"] != f"{job['bvid']}.mkv"
+ or item["acquisition_mode"] != "authorized_browser_file_handoff"
+ or not isinstance(item["title"], str)
+ or not 1 <= len(item["title"]) <= 1024
+ or any(ord(character) < 0x20 for character in item["title"])
+ or not isinstance(item["bytes"], int)
+ or isinstance(item["bytes"], bool)
+ or item["bytes"] <= 0
+ or not isinstance(item["sha256"], str)
+ or _LOWER_SHA256_RE.fullmatch(item["sha256"]) is None
+ or item["handoff_source_sha256"] != item["sha256"]
+ or not isinstance(item["format_name"], str)
+ or "matroska" not in item["format_name"].split(",")
+ or not isinstance(item["video_codec"], str)
+ or not re.fullmatch(r"[A-Za-z0-9_.-]{1,64}", item["video_codec"])
+ or not isinstance(item["audio_codec"], str)
+ or not re.fullmatch(r"[A-Za-z0-9_.-]{1,64}", item["audio_codec"])
+ or not isinstance(item["completed_at"], str)
+ or _UTC_ISO_RE.fullmatch(item["completed_at"]) is None
+ ):
+ raise WorkerError("E_BRIDGE_OUTPUT_SCHEMA")
+ duration = _bridge_number(item["duration_seconds"])
+ remote = _bridge_number(item["remote_duration_seconds"])
+ local = _bridge_number(item["local_duration_seconds"])
+ delta = _bridge_number(item["duration_delta_seconds"])
+ tolerance = _bridge_number(item["duration_tolerance_seconds"])
+ expected_tolerance = max(3.0, remote * 0.001)
+ if (
+ not _bridge_remote_matches_expected(job["expected_duration_ms"], remote)
+ or abs(duration - local) > 1e-9
+ or abs(delta - abs(local - remote)) > 1e-9
+ or abs(tolerance - expected_tolerance) > 1e-9
+ or delta > tolerance
+ ):
+ raise WorkerError("E_BRIDGE_DURATION_SHA")
+ return item
+
+
+def _read_exact_published_bridge_result(
+ config: HostConfig,
+ job: dict[str, Any],
+ *,
+ expected_item: dict[str, Any] | None = None,
+ on_media_verified: Callable[[str, str, dict[str, Any], dict[str, Any]], None] | None = None,
+ on_verified: Callable[[str, str, dict[str, Any]], None] | None = None,
+) -> tuple[str, str] | None:
+ destination = config.destination.resolve(strict=True)
+ formal_name = f"{job['bvid']}.mkv"
+ mapping_name = f"{job['bvid']}.download.json"
+ formal_path = destination / formal_name
+ mapping_path = destination / mapping_name
+ present = (formal_path.exists(), mapping_path.exists())
+ if present == (False, False):
+ return None
+ if present != (True, True):
+ raise WorkerError("E_BRIDGE_MEDIA_MAPPING")
+ formal_lock = _LockedPublishedFile.open(formal_path, destination, "E_BRIDGE_DURATION_SHA")
+ try:
+ mapping_lock = _LockedPublishedFile.open(
+ mapping_path, destination, "E_BRIDGE_MAPPING_READBACK"
+ )
+ try:
+ _stable_pair_test_seam("LOCKS_ACQUIRED")
+ mapping_stat = mapping_lock.assert_path_identity()
+ if mapping_stat.st_size <= 0 or mapping_stat.st_size > 64 * 1024:
+ raise WorkerError("E_BRIDGE_MAPPING_READBACK")
+ mapping_bytes = mapping_lock.read_all(64 * 1024)
+ try:
+ persisted = _validate_complete_bridge_item(
+ strict_json_loads(mapping_bytes), job, persisted=True
+ )
+ except WorkerError as exc:
+ if exc.code == "E_BRIDGE_DURATION_SHA":
+ raise
+ raise WorkerError("E_BRIDGE_MAPPING_READBACK") from exc
+ except (ProtocolError, UnicodeError, json.JSONDecodeError) as exc:
+ raise WorkerError("E_BRIDGE_MAPPING_READBACK") from exc
+ if expected_item is not None:
+ if {key: expected_item[key] for key in _BRIDGE_MAPPING_KEYS} != persisted:
+ raise WorkerError("E_BRIDGE_MAPPING_READBACK")
+ formal_stat = formal_lock.assert_path_identity()
+ formal_sha = formal_lock.sha256().casefold()
+ if persisted["bytes"] != formal_stat.st_size or persisted["sha256"] != formal_sha:
+ raise WorkerError("E_BRIDGE_DURATION_SHA")
+ _stable_pair_test_seam("AFTER_INITIAL_PAIR")
+
+ # The two Windows handles deny write/delete/path replacement while
+ # FFprobe opens its read-only view. Hashing and JSON parsing use these
+ # same handles, and the consumer commit executes before handle release.
+ probe_mkv(config.ffprobe, formal_path, job)
+ formal_lock.assert_path_identity()
+ mapping_lock.assert_path_identity()
+ final_formal_sha = formal_lock.sha256().casefold()
+ _stable_pair_test_seam("AFTER_FINAL_MEDIA_HASH")
+ final_mapping_bytes = mapping_lock.read_all(64 * 1024)
+ _stable_pair_test_seam("AFTER_FINAL_MAPPING_READ")
+ if final_formal_sha != persisted["sha256"]:
+ raise WorkerError("E_BRIDGE_DURATION_SHA")
+ if final_mapping_bytes != mapping_bytes:
+ raise WorkerError("E_BRIDGE_MAPPING_READBACK")
+ try:
+ final_persisted = _validate_complete_bridge_item(
+ strict_json_loads(final_mapping_bytes), job, persisted=True
+ )
+ except (WorkerError, ProtocolError, UnicodeError, json.JSONDecodeError) as exc:
+ raise WorkerError("E_BRIDGE_MAPPING_READBACK") from exc
+ if final_persisted != persisted:
+ raise WorkerError("E_BRIDGE_MAPPING_READBACK")
+ formal_lock.assert_path_identity()
+ mapping_lock.assert_path_identity()
+ _stable_pair_test_seam("BEFORE_COMMIT")
+ media_identity = validate_media_complete_identity(
+ {
+ "formal_filename": formal_name,
+ "mapping_filename": mapping_name,
+ "media_bytes": persisted["bytes"],
+ "media_sha256": persisted["sha256"].upper(),
+ "mapping_bytes": len(final_mapping_bytes),
+ "mapping_sha256": hashlib.sha256(final_mapping_bytes).hexdigest().upper(),
+ "duration_milliseconds": int(round(
+ float(persisted["local_duration_seconds"]) * 1_000
+ )),
+ "video_codec": persisted["video_codec"],
+ "audio_codec": persisted["audio_codec"],
+ },
+ job,
+ )
+ if on_media_verified is not None:
+ on_media_verified(formal_name, mapping_name, persisted, media_identity)
+ if on_verified is not None:
+ on_verified(formal_name, mapping_name, persisted)
+ return formal_name, mapping_name
+ finally:
+ mapping_lock.close()
+ finally:
+ formal_lock.close()
+
+
+def recover_published_task(
+ config: HostConfig,
+ job: dict[str, Any],
+ *,
+ cancel_check: Callable[[], bool],
+ report: Callable[..., None],
+ commit_begin: Callable[[], None],
+) -> tuple[str, str]:
+ """Verify and consume one exact published pair before secret transfer."""
+ if job["creator_uid"] not in config.creator_allowlist:
+ raise WorkerError("E_ALLOWLIST")
+ committed: tuple[str, str] | None = None
+
+ def media_verified(
+ _formal_name: str, _mapping_name: str, _persisted: dict[str, Any],
+ media_identity: dict[str, Any],
+ ) -> None:
+ report("MEDIA_COMPLETE", 100, media_identity)
+
+ def consume(formal_name: str, mapping_name: str, persisted: dict[str, Any]) -> None:
+ nonlocal committed
+ if cancel_check():
+ raise CancelRequested()
+ report("POSTPROCESS_PENDING", 100)
+ commit_begin()
+ _commit_formal_and_handoff(
+ config, job, formal_name, mapping_name, persisted,
+ media_complete_acknowledged=True,
+ )
+ committed = (formal_name, mapping_name)
+
+ recovered = _read_exact_published_bridge_result(
+ config, job, on_media_verified=media_verified, on_verified=consume
+ )
+ if recovered is None:
+ raise WorkerError("E_BRIDGE_MEDIA_MAPPING")
+ if committed != recovered:
+ raise WorkerError("E_BRIDGE_MAPPING_READBACK")
+ return recovered
+
+
+def _bridge_command(
+ config: HostConfig, candidate: Path, job: dict[str, Any], batch_json: Path
+) -> list[str]:
return [
str(config.bridge_python),
str(config.bridge_script),
"--input",
- str(config.batch_json),
- "--yt-dlp",
- str(config.yt_dlp_executable),
+ str(batch_json),
"accept-browser-file",
"--bvid",
- TARGET_BVID,
+ job["bvid"],
"--media-file",
str(candidate),
"--destination",
str(config.destination),
"--ffprobe",
str(config.ffprobe),
+ "--expected-duration-ms",
+ str(job["expected_duration_ms"]),
]
-def run_frozen_bridge(config: HostConfig, candidate: Path) -> tuple[str, str]:
+def run_frozen_bridge(
+ config: HostConfig,
+ candidate: Path,
+ job: dict[str, Any],
+ *,
+ on_media_verified: Callable[[str, str, dict[str, Any], dict[str, Any]], None] | None = None,
+ on_verified: Callable[[str, str, dict[str, Any]], None] | None = None,
+) -> tuple[str, str]:
+ started = time.monotonic()
+
+ def failure(code: str, state: str, reason: str) -> WorkerError:
+ return WorkerError(code, {
+ "attempts": 1,
+ "elapsed_ms": max(0, min(7_200_000, int((time.monotonic() - started) * 1000))),
+ "state": state,
+ "reason": reason,
+ })
+
+ batch_json = candidate.parent / "bridge-input.json"
+ batch_payload = {
+ "schema_version": "1.0",
+ "batch_id": f"generic-{job['job_id'][:16]}",
+ "items": [{
+ "bvid": job["bvid"],
+ "source_url": job["canonical_url"],
+ "published_at": job["published_at"],
+ "title": job["title"],
+ "expected_duration_ms": job["expected_duration_ms"],
+ }],
+ }
+ try:
+ with batch_json.open("xb") as stream:
+ stream.write((json.dumps(batch_payload, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8"))
+ stream.flush()
+ os.fsync(stream.fileno())
+ except BaseException as exc:
+ batch_json.unlink(missing_ok=True)
+ raise failure(
+ "E_BRIDGE_INVOCATION", "BRIDGE_INVOCATION", "BATCH_RECEIPT_CREATE_FAILED"
+ ) from exc
try:
result = _run_local(
- _bridge_command(config, candidate),
+ _bridge_command(config, candidate, job, batch_json),
BRIDGE_TIMEOUT_SECONDS,
capture_stdout=True,
)
except subprocess.TimeoutExpired as exc:
- raise WorkerError("E_BACKHALF") from exc
+ raise failure(
+ "E_BRIDGE_INVOCATION", "BRIDGE_INVOCATION", "SUBPROCESS_TIMEOUT"
+ ) from exc
+ except OSError as exc:
+ raise failure(
+ "E_BRIDGE_INVOCATION", "BRIDGE_INVOCATION", "SUBPROCESS_INVOCATION_FAILED"
+ ) from exc
+ finally:
+ batch_json.unlink(missing_ok=True)
if result.returncode != 0:
- raise WorkerError("E_BACKHALF")
+ try:
+ stopped = strict_json_loads(result.stdout)
+ except (ProtocolError, UnicodeError, json.JSONDecodeError):
+ stopped = None
+ bridge_stops = {
+ "E_BRIDGE_SOURCE_STABILITY": (
+ "BRIDGE_SOURCE_STABILITY", "SOURCE_FILE_INVALID"
+ ),
+ "E_BRIDGE_METADATA_BINDING": (
+ "BRIDGE_METADATA_BINDING", "EXPECTED_METADATA_MISMATCH"
+ ),
+ "E_BRIDGE_FFPROBE": (
+ "BRIDGE_FFPROBE", "LOCAL_MEDIA_PROBE_FAILED"
+ ),
+ "E_BRIDGE_DURATION_SHA": (
+ "DURATION_SHA_VERIFICATION", "DURATION_OR_SHA_MISMATCH"
+ ),
+ "E_BRIDGE_PUBLISH": (
+ "CREATE_NEW_PUBLISH", "BRIDGE_REPORTED_PUBLISH_FAILURE"
+ ),
+ }
+ if (
+ isinstance(stopped, dict)
+ and set(stopped) == {"result", "error_code"}
+ and stopped.get("result") == "SAFETY_STOP"
+ and stopped.get("error_code") in bridge_stops
+ ):
+ code = stopped["error_code"]
+ state, reason = bridge_stops[code]
+ raise failure(code, state, reason)
+ raise failure("E_BRIDGE_EXIT", "BRIDGE_EXIT", "NONZERO_EXIT")
+
+ def read_published(expected_item: dict[str, Any] | None) -> tuple[str, str]:
+ try:
+ published_value = _read_exact_published_bridge_result(
+ config, job, expected_item=expected_item,
+ on_media_verified=on_media_verified, on_verified=on_verified,
+ )
+ except WorkerError as exc:
+ if exc.code.startswith("E_COMPLETION_"):
+ raise
+ code = exc.code if exc.code.startswith("E_BRIDGE_") else "E_BRIDGE_MAPPING_READBACK"
+ state, reason = {
+ "E_BRIDGE_MEDIA_MAPPING": ("MEDIA_MAPPING_PRESENCE", "MEDIA_OR_MAPPING_MISSING"),
+ "E_BRIDGE_DURATION_SHA": ("DURATION_SHA_VERIFICATION", "DURATION_OR_SHA_MISMATCH"),
+ }.get(code, ("MAPPING_READBACK", "PERSISTED_MAPPING_INVALID"))
+ raise failure(code, state, reason) from exc
+ if published_value is None:
+ raise failure(
+ "E_BRIDGE_MEDIA_MAPPING", "MEDIA_MAPPING_PRESENCE", "MEDIA_OR_MAPPING_MISSING"
+ )
+ return published_value
+
+ def preserve_verified_media_before_output_failure(_cause: BaseException | None = None) -> bool:
+ """Persist media truth, but do not commit postprocess through an invalid wire result."""
+ try:
+ if os.path.lexists(config.destination / ".bili-download-staging"):
+ raise failure(
+ "E_BRIDGE_OUTPUT_SCHEMA", "BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID"
+ )
+ published_value = _read_exact_published_bridge_result(
+ config, job, on_media_verified=on_media_verified
+ )
+ if published_value is None:
+ raise WorkerError("E_BRIDGE_MEDIA_MAPPING")
+ return True
+ except WorkerError as exc:
+ if exc.code.startswith("E_COMPLETION_"):
+ raise
+ return False
+
try:
- payload = json.loads(result.stdout.decode("utf-8"))
- except (UnicodeError, json.JSONDecodeError) as exc:
- raise WorkerError("E_BACKHALF") from exc
- if not isinstance(payload, dict) or payload.get("result") != "PASS":
- raise WorkerError("E_BACKHALF")
+ payload = strict_json_loads(result.stdout)
+ except (ProtocolError, UnicodeError, json.JSONDecodeError) as exc:
+ preserve_verified_media_before_output_failure(exc)
+ raise failure(
+ "E_BRIDGE_OUTPUT_SCHEMA", "BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID"
+ ) from exc
+ if (
+ not isinstance(payload, dict)
+ or set(payload) != {"batch_id", "command", "result", "success_count", "failure_count", "items"}
+ or payload.get("batch_id") != f"generic-{job['job_id'][:16]}"
+ or payload.get("result") != "PASS"
+ or payload.get("command") != "accept-browser-file"
+ or payload.get("success_count") != 1
+ or payload.get("failure_count") != 0
+ ):
+ preserve_verified_media_before_output_failure()
+ raise failure(
+ "E_BRIDGE_OUTPUT_SCHEMA", "BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID"
+ )
items = payload.get("items")
if not isinstance(items, list) or len(items) != 1 or not isinstance(items[0], dict):
- raise WorkerError("E_BACKHALF")
- item = items[0]
- required = {"bvid", "local_file", "bytes", "sha256", "duration_seconds", "acquisition_mode"}
- if not required.issubset(item) or item["bvid"] != TARGET_BVID:
- raise WorkerError("E_BACKHALF")
- if item["acquisition_mode"] != "authorized_browser_file_handoff":
- raise WorkerError("E_BACKHALF")
- formal = item["local_file"]
- mapping = f"{TARGET_BVID}.download.json"
- if formal != f"{TARGET_BVID}.mkv":
- raise WorkerError("E_BACKHALF")
- formal_path = config.destination / formal
- mapping_path = config.destination / mapping
- if not formal_path.is_file() or not mapping_path.is_file():
- raise WorkerError("E_BACKHALF")
+ preserve_verified_media_before_output_failure()
+ raise failure(
+ "E_BRIDGE_OUTPUT_SCHEMA", "BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID"
+ )
try:
- duration_ms = round(_finite_number(item["duration_seconds"]) * 1000)
+ item = _validate_complete_bridge_item(items[0], job, persisted=False)
except WorkerError as exc:
- raise WorkerError("E_BACKHALF") from exc
- if abs(duration_ms - EXPECTED_DURATION_MS) > DURATION_TOLERANCE_MS:
- raise WorkerError("E_BACKHALF")
- formal_sha = sha256_file(formal_path)
- if (
- isinstance(item["bytes"], bool)
- or not isinstance(item["bytes"], int)
- or item["bytes"] != formal_path.stat().st_size
- or not isinstance(item["sha256"], str)
- or item["sha256"] != formal_sha
- ):
- raise WorkerError("E_BACKHALF")
- try:
- persisted = strict_json_loads(mapping_path.read_bytes())
- except (OSError, ProtocolError) as exc:
- raise WorkerError("E_BACKHALF") from exc
- matched_fields = {
- "bvid": TARGET_BVID,
- "source": CANONICAL_URL,
- "local_file": formal,
- "bytes": item["bytes"],
- "sha256": formal_sha,
- "acquisition_mode": "authorized_browser_file_handoff",
- }
- if any(persisted.get(key) != expected for key, expected in matched_fields.items()):
- raise WorkerError("E_BACKHALF")
- try:
- persisted_duration_ms = round(_finite_number(persisted.get("duration_seconds")) * 1000)
- except WorkerError as exc:
- raise WorkerError("E_BACKHALF") from exc
- if persisted_duration_ms != duration_ms:
- raise WorkerError("E_BACKHALF")
- return formal, mapping
+ preserve_verified_media_before_output_failure(exc)
+ code = exc.code if exc.code == "E_BRIDGE_DURATION_SHA" else "E_BRIDGE_OUTPUT_SCHEMA"
+ state, reason = (
+ ("DURATION_SHA_VERIFICATION", "DURATION_OR_SHA_MISMATCH")
+ if code == "E_BRIDGE_DURATION_SHA"
+ else ("BRIDGE_OUTPUT_SCHEMA", "OUTPUT_SCHEMA_INVALID")
+ )
+ raise failure(code, state, reason) from exc
+ if "warning" in item:
+ try:
+ if os.path.lexists(config.destination / ".bili-download-staging"):
+ raise failure(
+ "E_BRIDGE_MAPPING_READBACK", "MAPPING_READBACK", "PERSISTED_MAPPING_INVALID"
+ )
+ except OSError as exc:
+ raise failure(
+ "E_BRIDGE_MAPPING_READBACK", "MAPPING_READBACK", "PERSISTED_MAPPING_INVALID"
+ ) from exc
+ return read_published(item)
def ytdlp_options(
@@ -799,15 +2065,19 @@
config: HostConfig,
*,
cancel_check: Callable[[], bool],
- report: Callable[[str, int], None],
+ report: Callable[..., None],
stage_root: Path | None = None,
prepared_run_directory: Path | None = None,
commit_begin: Callable[[], None] | None = None,
closure_report: Callable[[bool], None] | None = None,
+ recovery_required: bool = False,
) -> tuple[str, str, bool]:
"""Run the exact task. The caller must already own this worker in a job."""
validate_start(start)
- root = fixed_stage_root() if stage_root is None else stage_root.resolve()
+ job = start["job"]
+ if job["creator_uid"] not in config.creator_allowlist:
+ raise WorkerError("E_ALLOWLIST")
+ root = fixed_stage_root(job["bvid"]) if stage_root is None else stage_root.resolve()
if prepared_run_directory is None:
run_directory = prepare_run_directory(root)
else:
@@ -825,6 +2095,38 @@
raise CancelRequested()
checkpoint()
+ def consume_recovered(
+ formal_name: str, mapping_name: str, persisted: dict[str, Any]
+ ) -> None:
+ nonlocal cookie_closed, committed, outcome
+ checkpoint()
+ report("POSTPROCESS_PENDING", 100)
+ if commit_begin is not None:
+ commit_begin()
+ _commit_formal_and_handoff(
+ config, job, formal_name, mapping_name, persisted,
+ media_complete_acknowledged=True,
+ )
+ cookie_closed = True
+ committed = True
+ outcome = (formal_name, mapping_name, True)
+
+ def recovered_media_verified(
+ _formal_name: str, _mapping_name: str, _persisted: dict[str, Any],
+ media_identity: dict[str, Any],
+ ) -> None:
+ report("MEDIA_COMPLETE", 100, media_identity)
+
+ recovered = _read_exact_published_bridge_result(
+ config, job, on_media_verified=recovered_media_verified,
+ on_verified=consume_recovered,
+ )
+ if recovered is not None:
+ if outcome != (recovered[0], recovered[1], True):
+ raise WorkerError("E_BRIDGE_MAPPING_READBACK")
+ return outcome
+ if recovery_required:
+ raise WorkerError("E_BRIDGE_MEDIA_MAPPING")
cookie_stream = build_cookie_stream(start)
yt_dlp, _ = bootstrap_ytdlp()
secret_values = [
@@ -852,7 +2154,7 @@
opts = ytdlp_options(config, run_directory, cookie_stream, progress_hook)
with yt_dlp.YoutubeDL(opts) as ydl:
report("CHECKING", 0)
- download_info, single, signed_urls = prepare_download_info(ydl)
+ download_info, single, signed_urls = prepare_download_info(ydl, job)
policy.secrets.update(url.casefold() for url in signed_urls)
checkpoint()
ydl.process_info(download_info)
@@ -882,7 +2184,7 @@
checkpoint()
candidate = validate_unique_candidate(run_directory)
checkpoint()
- probe_mkv(config.ffprobe, candidate)
+ probe_mkv(config.ffprobe, candidate, job)
checkpoint()
cookie_closed = close_cookie_stream(cookie_stream)
cookie_stream = None
@@ -892,7 +2194,29 @@
if commit_begin is not None:
commit_begin()
checkpoint()
- formal, mapping = run_frozen_bridge(config, candidate)
+ completion_committed = False
+
+ def consume_published(
+ formal_name: str, mapping_name: str, persisted: dict[str, Any]
+ ) -> None:
+ nonlocal completion_committed
+ checkpoint()
+ report("POSTPROCESS_PENDING", 100)
+ _commit_formal_and_handoff(
+ config, job, formal_name, mapping_name, persisted,
+ media_complete_acknowledged=True,
+ )
+ completion_committed = True
+
+ formal, mapping = run_frozen_bridge(
+ config, candidate, job,
+ on_media_verified=lambda _formal, _mapping, _persisted, media: report(
+ "MEDIA_COMPLETE", 100, media
+ ),
+ on_verified=consume_published,
+ )
+ if not completion_committed:
+ raise WorkerError("E_COMPLETION_HANDOFF")
committed = True
outcome = (formal, mapping, cookie_closed)
finally:
--
Gitblit v1.9.3