"""Governed dynamic-catalog to schema-1 authenticated queue producer.
|
|
The command is intentionally separate from the Native Messaging host. It
|
does not open Chrome, call the host, read credentials, or accept caller-owned
|
job records. Its only mutation is an append to the configured queue through
|
``QueueStore.append_ingress_jobs`` and that method's shared first-byte lock.
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
import base64
|
import hashlib
|
import json
|
import os
|
import re
|
import stat
|
import sys
|
from datetime import datetime
|
from decimal import Decimal, InvalidOperation
|
from pathlib import Path, PurePosixPath
|
from typing import Any
|
|
from .constants import (
|
EXTENSION_BUILD,
|
EXPECTED_EXTENSION_ID,
|
EXPECTED_ORIGIN,
|
HOST_BUILD,
|
MESSAGE_ID_RE,
|
RELOAD_GENERATION,
|
canonical_url,
|
stable_job_id,
|
validate_bvid,
|
validate_creator_uid,
|
)
|
from .protocol import ProtocolError, strict_json_loads
|
from .queue_state import (
|
AuthorizedSuccessor,
|
QueueStore,
|
ReleaseApproval,
|
validate_ingress_record,
|
)
|
|
|
PRODUCER_ID = "project-info-bili-auth-queue-producer/2"
|
_LEGACY_PRODUCER_ID = "project-info-bili-auth-queue-producer/1"
|
_TASK_ID = "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001"
|
_PROJECT_ID = "project-info"
|
_HEX64 = re.compile(r"[0-9A-Fa-f]{64}\Z")
|
_CONTROL = re.compile(r"[\x00-\x1f\x7f]")
|
_FORBIDDEN_KEYS = re.compile(r"(?:cookie|password|passwd|access_token|refresh_token|signed_url|local_?storage|captcha)", re.I)
|
_HOST_KEYS = {
|
"schema", "creator_allowlist", "queue_path", "queue_state_path", "queue_lock_path",
|
"reload_state_path", "reload_generation", "required_extension_build", "ffmpeg",
|
"ffmpeg_sha256", "ffprobe", "ffprobe_sha256", "bridge_python",
|
"bridge_python_sha256", "bridge_script", "bridge_script_sha256",
|
"yt_dlp_executable", "yt_dlp_executable_sha256", "destination",
|
"creator_name", "formal_manifest_path", "processing_handoff_path",
|
}
|
_CONFIG_V1_KEYS = {
|
"schema", "producer_id", "project_root", "host_config_path", "host_config_sha256",
|
"creator_name", "creator_uid", "dynamic_manifest_path", "registered_catalog_path",
|
"registered_catalog_sha256", "bvid_allowlist",
|
}
|
_CONFIG_V2_KEYS = _CONFIG_V1_KEYS | {"successor_authorization_message_id"}
|
_REGISTERED_KEYS = {"schema", "source", "creator_name", "creator_uid", "items"}
|
_INGRESS_KEYS = {
|
"schema", "bvid", "creator_uid", "expected_duration_ms",
|
"discovered_at_unix_ms", "published_at", "title",
|
}
|
_TERMINAL_VIDEO_STATUSES = frozenset({
|
"VIDEO_DOWNLOADED_COMPLETE_HANDOFF_SENT",
|
"VIDEO_DOWNLOADED_COMPLETE",
|
"VIDEO_COMPLETE",
|
})
|
_SOURCE_MANIFEST_KEYS = {
|
"schema", "scope", "extension_id", "extension_build", "host_build",
|
"archive_metadata_contract", "dependency_artifact_manifest_bytes",
|
"dependency_artifact_manifest_sha256", "files",
|
}
|
_ARCHIVE_METADATA_KEYS = {
|
"schema", "root", "relative_files", "distribution_name", "distribution_version",
|
"allowed_type_codes", "source_date_epoch", "tree_hash_algorithm",
|
"canonical_tree_sha256",
|
}
|
_ARTIFACT_FILE_KEYS = {"path", "bytes", "sha256"}
|
_BUILD_APPROVAL_KEYS = {
|
"schema", "task_id", "approval_scope", "approved_by_role", "status",
|
"source_artifact_manifest_bytes", "source_artifact_manifest_sha256",
|
}
|
_BUILD_RECEIPT_KEYS = {
|
"schema", "scope", "extension_id", "extension_build", "host_build", "packaging",
|
"pyinstaller_version", "yt_dlp_version", "builder_python_sha256",
|
"pyinstaller_executable_bytes", "pyinstaller_executable_sha256",
|
"builder_provision_receipt_bytes", "builder_provision_receipt_sha256",
|
"build_script_sha256", "source_artifact_manifest_bytes",
|
"source_artifact_manifest_sha256", "dependency_artifact_manifest_bytes",
|
"dependency_artifact_manifest_sha256", "yt_dlp_wheel_sha256",
|
"archive_verification", "files",
|
}
|
_ARCHIVE_VERIFICATION_KEYS = {
|
"status", "method", "required_modules", "metadata_entry", "metadata_files",
|
"metadata_type_codes", "metadata_tree_sha256",
|
}
|
_INSTALL_APPROVAL_KEYS = {
|
"schema", "task_id", "approval_scope", "approved_by_role", "status",
|
"source_artifact_manifest_sha256", "build_artifact_manifest_bytes",
|
"build_artifact_manifest_sha256", "host_executable_bytes", "host_executable_sha256",
|
}
|
_INSTALL_RECEIPT_KEYS = {
|
"schema", "task_id", "host_build", "required_extension_build", "extension_id",
|
"host_name", "installed_root", "installed_files",
|
}
|
_FAST_PATH_INSTALL_RECEIPT_KEYS = _INSTALL_RECEIPT_KEYS | {
|
"validation_scope", "validated_by_role", "status",
|
"continuous_authorization_handoff_id", "owner_ai_id", "owner_thread_id",
|
"owner_role_instance_id", "authorization_file_relative_path",
|
"authorization_file_bytes", "authorization_file_sha256", "successor_scope_sha256",
|
"implementation_review_audit_id", "implementation_review_audit_bytes",
|
"implementation_review_audit_sha256", "source_artifact_manifest_bytes",
|
"source_artifact_manifest_sha256", "source_receipt_bytes", "source_receipt_sha256",
|
"build_artifact_manifest_bytes", "build_artifact_manifest_sha256",
|
"build_receipt_bytes", "build_receipt_sha256", "host_executable_bytes",
|
"host_executable_sha256", "native_messaging_host_manifest",
|
"host_build_source_manifest", "host_source_binding_mode", "producer_only_changed_files",
|
"host_archive_excluded_modules",
|
"projection_contract_sha256", "projection_tree_sha256", "reload_state_path",
|
"reload_state_bytes", "reload_state_lines", "reload_state_sha256", "queue_path",
|
"queue_prefix_bytes", "queue_prefix_lines", "queue_prefix_sha256",
|
"queue_state_path", "queue_state_prefix_bytes", "queue_state_prefix_lines",
|
"queue_state_prefix_sha256", "formal_manifest_path", "formal_manifest_prefix_bytes",
|
"formal_manifest_prefix_lines", "formal_manifest_prefix_sha256", "secret_field_count",
|
}
|
_CONTINUOUS_FAST_PATH_HANDOFF_ID = (
|
"HANDOFF-INFOADMIN-INFODEV2-BILI-AUTH-V010-GENERATION9-PASS0-"
|
"CONTINUOUS-TARGET-FAST-PATH-RELEASE-RUNTIME-20260820-001"
|
)
|
_FAST_PATH_OWNER = (
|
"infodev-2",
|
"019fbcbb-bed7-7c90-83ab-f50610f80d3a",
|
"dev.developer.project.secondary",
|
)
|
|
|
def _is_reparse(stat_result: os.stat_result) -> bool:
|
attributes = int(getattr(stat_result, "st_file_attributes", 0))
|
reparse = int(getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400))
|
return stat.S_ISLNK(stat_result.st_mode) or bool(attributes & reparse)
|
|
|
def _absolute_local_path(value: Any, code: str) -> Path:
|
if not isinstance(value, str) or not value or _CONTROL.search(value):
|
raise ProtocolError(code)
|
path = Path(value)
|
if not path.is_absolute() or str(path).startswith("\\\\"):
|
raise ProtocolError(code)
|
return Path(os.path.abspath(path))
|
|
|
def _check_existing_chain(path: Path, *, require_file: bool | None) -> None:
|
current = Path(path.anchor)
|
for part in path.parts[1:]:
|
current = current / part
|
try:
|
info = os.lstat(current)
|
except FileNotFoundError:
|
if require_file:
|
raise ProtocolError("E_CONFIG")
|
return
|
if _is_reparse(info):
|
raise ProtocolError("E_CONFIG")
|
if require_file is True and not path.is_file():
|
raise ProtocolError("E_CONFIG")
|
if require_file is False and path.exists() and not path.is_dir():
|
raise ProtocolError("E_CONFIG")
|
|
|
def _within(path: Path, root: Path) -> bool:
|
try:
|
return os.path.commonpath((str(path), str(root))) == str(root)
|
except ValueError:
|
return False
|
|
|
def _reject_secret_keys(value: Any) -> None:
|
if isinstance(value, dict):
|
for key, child in value.items():
|
if not isinstance(key, str) or _FORBIDDEN_KEYS.search(key):
|
raise ProtocolError("E_SECRET_FIELD")
|
_reject_secret_keys(child)
|
elif isinstance(value, list):
|
for child in value:
|
_reject_secret_keys(child)
|
|
|
def _safe_text(value: Any, maximum: int = 600) -> str:
|
if not isinstance(value, str) or not value.strip() or _CONTROL.search(value):
|
raise ProtocolError("E_CATALOG")
|
if len(value.encode("utf-8")) > maximum:
|
raise ProtocolError("E_CATALOG")
|
return value
|
|
|
def _time_ms(value: Any) -> int:
|
if not isinstance(value, str) or _CONTROL.search(value):
|
raise ProtocolError("E_CATALOG")
|
try:
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
except ValueError as exc:
|
raise ProtocolError("E_CATALOG") from exc
|
if parsed.utcoffset() is None:
|
raise ProtocolError("E_CATALOG")
|
return int(parsed.timestamp() * 1000)
|
|
|
def _duration_ms(value: Any) -> int:
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
raise ProtocolError("E_CATALOG")
|
try:
|
milliseconds = Decimal(str(value)) * 1000
|
except InvalidOperation as exc:
|
raise ProtocolError("E_CATALOG") from exc
|
if milliseconds != milliseconds.to_integral_value():
|
raise ProtocolError("E_CATALOG")
|
result = int(milliseconds)
|
if not 1_000 <= result <= 86_400_000:
|
raise ProtocolError("E_CATALOG")
|
return result
|
|
|
def _read_strict_payload(path: Path, maximum: int, code: str) -> bytes:
|
try:
|
_check_existing_chain(path, require_file=True)
|
with path.open("rb") as source:
|
payload = source.read(maximum + 1)
|
except (OSError, ProtocolError) as exc:
|
raise ProtocolError(code) from exc
|
if len(payload) > maximum:
|
raise ProtocolError(code)
|
return payload
|
|
|
def _enumerate_exact_tree(root: Path, expected_files: set[str], code: str) -> None:
|
"""Require the actual no-follow tree to equal an exact case-sensitive file set."""
|
try:
|
_check_existing_chain(root, require_file=False)
|
root_info = os.lstat(root)
|
except (OSError, ProtocolError) as exc:
|
raise ProtocolError(code) from exc
|
if _is_reparse(root_info) or not stat.S_ISDIR(root_info.st_mode):
|
raise ProtocolError(code)
|
expected_directories = {
|
PurePosixPath(path).parent.as_posix()
|
for path in expected_files
|
if PurePosixPath(path).parent.as_posix() != "."
|
}
|
actual_files: set[str] = set()
|
actual_directories: set[str] = set()
|
pending: list[tuple[Path, PurePosixPath]] = [(root, PurePosixPath("."))]
|
while pending:
|
directory, relative_directory = pending.pop()
|
try:
|
entries = list(os.scandir(directory))
|
except OSError as exc:
|
raise ProtocolError(code) from exc
|
for entry in entries:
|
try:
|
info = entry.stat(follow_symlinks=False)
|
except OSError as exc:
|
raise ProtocolError(code) from exc
|
if _is_reparse(info):
|
raise ProtocolError(code)
|
relative = PurePosixPath(entry.name) if relative_directory == PurePosixPath(".") else relative_directory / entry.name
|
normalized = relative.as_posix()
|
if stat.S_ISDIR(info.st_mode):
|
actual_directories.add(normalized)
|
pending.append((Path(entry.path), relative))
|
elif stat.S_ISREG(info.st_mode):
|
actual_files.add(normalized)
|
else:
|
raise ProtocolError(code)
|
if actual_files != expected_files or actual_directories != expected_directories:
|
raise ProtocolError(code)
|
|
|
def _validate_source_tree(root: Path, source_payload: bytes, deployment: dict[str, Any]) -> dict[str, Any]:
|
code = "E_DEPLOYMENT_NOT_READY"
|
source = _plain_mapping(strict_json_loads(source_payload), _SOURCE_MANIFEST_KEYS, code)
|
if (
|
source["schema"] != 1 or source["scope"] != "generic-bilibili-queue"
|
or source["extension_id"] != EXPECTED_EXTENSION_ID
|
or source["extension_build"] != EXTENSION_BUILD or source["host_build"] != HOST_BUILD
|
):
|
raise ProtocolError(code)
|
archive = _plain_mapping(source["archive_metadata_contract"], _ARCHIVE_METADATA_KEYS, code)
|
if (
|
archive["schema"] != 1 or archive["root"] != "yt_dlp-2026.7.4.dist-info"
|
or archive["relative_files"] != [
|
"INSTALLER", "METADATA", "RECORD", "REQUESTED", "WHEEL",
|
"entry_points.txt", "licenses/LICENSE",
|
]
|
or archive["distribution_name"] != "yt-dlp" or archive["distribution_version"] != "2026.7.4"
|
or archive["allowed_type_codes"] != ["b", "x"]
|
or archive["source_date_epoch"] != 1786207924
|
or archive["tree_hash_algorithm"] != "sha256(path-utf8,nul,decimal-bytes-ascii,nul,payload-sha256-lower-hex-ascii,lf)-upper-hex-v1"
|
or archive["canonical_tree_sha256"] != "32F1DC23F6704966AE4511CB173318CD11FCBA031323028D268D9F2488589E70"
|
):
|
raise ProtocolError(code)
|
dependency_size = _positive_integer(source["dependency_artifact_manifest_bytes"], code)
|
dependency_hash = _upper_sha(source["dependency_artifact_manifest_sha256"], code)
|
files = source["files"]
|
if not isinstance(files, list) or len(files) != 21:
|
raise ProtocolError(code)
|
source_root = root / "dev" / "project-dev" / "bili_authenticated_extension"
|
expected: dict[str, tuple[int, str]] = {}
|
for item in files:
|
item = _plain_mapping(item, _ARTIFACT_FILE_KEYS, code)
|
relative = _relative_path(item["path"], ("",), code)
|
size = _positive_integer(item["bytes"], code)
|
digest = _upper_sha(item["sha256"], code)
|
if relative == "source-artifact-manifest.json" or relative in expected:
|
raise ProtocolError(code)
|
expected[relative] = (size, digest)
|
if list(expected) != sorted(expected):
|
raise ProtocolError(code)
|
_enumerate_exact_tree(source_root, set(expected) | {"source-artifact-manifest.json"}, code)
|
for relative, (size, digest) in expected.items():
|
payload = _read_strict_payload(source_root / Path(relative), size + 1, code)
|
if len(payload) != size or hashlib.sha256(payload).hexdigest().upper() != digest:
|
raise ProtocolError(code)
|
dependency = expected.get("dependencies/dependency-artifact-manifest.json")
|
if dependency != (dependency_size, dependency_hash):
|
raise ProtocolError(code)
|
return source
|
|
|
def _strict_file(path: Path, maximum: int, code: str = "E_CATALOG") -> Any:
|
payload = _read_strict_payload(path, maximum, code)
|
return strict_json_loads(payload)
|
|
|
def load_producer_configuration(path: Path) -> dict[str, Any]:
|
path = Path(os.path.abspath(path))
|
_check_existing_chain(path, require_file=True)
|
raw = _strict_file(path, 128 * 1024, "E_CONFIG")
|
if not isinstance(raw, dict) or raw.get("schema") not in {1, 2}:
|
raise ProtocolError("E_CONFIG")
|
expected_keys = _CONFIG_V1_KEYS if raw["schema"] == 1 else _CONFIG_V2_KEYS
|
if set(raw) != expected_keys or raw.get("producer_id") not in {_LEGACY_PRODUCER_ID, PRODUCER_ID}:
|
raise ProtocolError("E_CONFIG")
|
if raw["schema"] == 2:
|
authorization_message_id = raw["successor_authorization_message_id"]
|
if authorization_message_id is not None and (
|
not isinstance(authorization_message_id, str)
|
or not MESSAGE_ID_RE.fullmatch(authorization_message_id)
|
):
|
raise ProtocolError("E_CONFIG")
|
_reject_secret_keys(raw)
|
project_root = _absolute_local_path(raw["project_root"], "E_CONFIG")
|
_check_existing_chain(project_root, require_file=False)
|
if not project_root.is_dir() or not _within(path, project_root):
|
raise ProtocolError("E_CONFIG")
|
creator_name = _safe_text(raw["creator_name"], 128)
|
try:
|
creator_uid = validate_creator_uid(raw["creator_uid"])
|
except ValueError as exc:
|
raise ProtocolError("E_CONFIG") from exc
|
host_path = _absolute_local_path(raw["host_config_path"], "E_CONFIG")
|
_check_existing_chain(host_path, require_file=True)
|
expected_host_hash = raw["host_config_sha256"]
|
host_payload = _read_strict_payload(host_path, 128 * 1024, "E_CONFIG")
|
actual_host_hash = hashlib.sha256(host_payload).hexdigest().upper()
|
if not isinstance(expected_host_hash, str) or not _HEX64.fullmatch(expected_host_hash) or actual_host_hash != expected_host_hash.upper():
|
raise ProtocolError("E_CONFIG_HASH")
|
host = strict_json_loads(host_payload)
|
if not isinstance(host, dict) or set(host) != _HOST_KEYS or host.get("schema") != 2:
|
raise ProtocolError("E_CONFIG")
|
creators = host.get("creator_allowlist")
|
if not isinstance(creators, list) or creators != sorted(set(creators)) or creator_uid not in creators:
|
raise ProtocolError("E_ALLOWLIST")
|
if host.get("required_extension_build") != EXTENSION_BUILD:
|
raise ProtocolError("E_CONFIG")
|
if host.get("reload_generation") != RELOAD_GENERATION:
|
raise ProtocolError("E_CONFIG")
|
queue_paths = {
|
name: _absolute_local_path(host[name], "E_CONFIG")
|
for name in ("queue_path", "queue_state_path", "queue_lock_path", "reload_state_path")
|
}
|
if len(set(queue_paths.values())) != 4 or len({value.parent for value in queue_paths.values()}) != 1:
|
raise ProtocolError("E_CONFIG")
|
for candidate in queue_paths.values():
|
_check_existing_chain(candidate, require_file=None)
|
if candidate.exists() and not candidate.is_file():
|
raise ProtocolError("E_CONFIG")
|
dynamic_path = _absolute_local_path(raw["dynamic_manifest_path"], "E_CONFIG")
|
if not _within(dynamic_path, project_root):
|
raise ProtocolError("E_CONFIG")
|
_check_existing_chain(dynamic_path, require_file=True)
|
formal_path = _absolute_local_path(host["formal_manifest_path"], "E_CONFIG")
|
handoff_path = _absolute_local_path(host["processing_handoff_path"], "E_CONFIG")
|
if (
|
host["creator_name"] != creator_name or formal_path != dynamic_path
|
or not _within(handoff_path, project_root) or handoff_path == formal_path
|
):
|
raise ProtocolError("E_CONFIG")
|
_check_existing_chain(handoff_path, require_file=None)
|
if handoff_path.exists() and not handoff_path.is_file():
|
raise ProtocolError("E_CONFIG")
|
registered_path = None
|
registered_payload = None
|
actual_registered_hash = None
|
registered_hash = raw["registered_catalog_sha256"]
|
if raw["registered_catalog_path"] is None:
|
if registered_hash is not None:
|
raise ProtocolError("E_CONFIG")
|
else:
|
registered_path = _absolute_local_path(raw["registered_catalog_path"], "E_CONFIG")
|
if not _within(registered_path, project_root):
|
raise ProtocolError("E_CONFIG")
|
_check_existing_chain(registered_path, require_file=True)
|
registered_payload = _read_strict_payload(registered_path, 4 * 1024 * 1024, "E_CATALOG")
|
actual_registered_hash = hashlib.sha256(registered_payload).hexdigest().upper()
|
if not isinstance(registered_hash, str) or not _HEX64.fullmatch(registered_hash) or actual_registered_hash != registered_hash.upper():
|
raise ProtocolError("E_CONFIG_HASH")
|
allowlist = raw["bvid_allowlist"]
|
if allowlist is not None:
|
if not isinstance(allowlist, list) or not allowlist or allowlist != sorted(set(allowlist)):
|
raise ProtocolError("E_CONFIG")
|
try:
|
allowlist = [validate_bvid(value) for value in allowlist]
|
except ValueError as exc:
|
raise ProtocolError("E_CONFIG") from exc
|
return {
|
**raw,
|
"project_root": project_root,
|
"host_config_path": host_path,
|
"dynamic_manifest_path": dynamic_path,
|
"registered_catalog_path": registered_path,
|
"creator_name": creator_name,
|
"creator_uid": creator_uid,
|
"bvid_allowlist": None if allowlist is None else frozenset(allowlist),
|
"queue_paths": queue_paths,
|
"_host_config_sha256": actual_host_hash,
|
"_registered_catalog_payload": registered_payload,
|
"_registered_catalog_sha256": actual_registered_hash,
|
}
|
|
|
_RELEASE_KEYS = {
|
"schema", "scope", "project_id", "task_id", "approved_by_role",
|
"authorization_message_id", "authorization_handoff_id", "authorization_file",
|
"implementation_review", "deployment",
|
}
|
_FILE_ID_KEYS = {"relative_path", "bytes", "sha256"}
|
_REVIEW_KEYS = {
|
"result_message_id", "audit_id", "audit_bytes", "audit_sha256",
|
"verdict", "blocking_findings",
|
}
|
_DEPLOYMENT_KEYS = {
|
"extension_build", "host_build", "reload_generation", "source_manifest_sha256",
|
"build_approval", "build_receipt", "exe", "install_approval", "install_receipt",
|
"installed_config_sha256", "installed_manifest_sha256", "installed_exe_sha256",
|
"projection_contract_sha256", "projection_tree_sha256", "extension_id",
|
}
|
_AUTH_KEYS = {
|
"schema", "scope", "task_id", "authorized_by_role", "authorization_message_id",
|
"authorization_handoff_id", "repair", "successors",
|
}
|
_AUTH_REPAIR_KEYS = {
|
"review_result_message_id", "audit_id", "audit_bytes", "audit_sha256",
|
"verdict", "blocking_findings",
|
}
|
_AUTH_SUCCESSOR_KEYS = {
|
"creator_uid", "bvid", "predecessor_job_id", "retry_generation",
|
"terminal_error_code",
|
}
|
|
|
def _project_root_from_source() -> Path:
|
root = Path(__file__).resolve().parents[3]
|
_check_existing_chain(root, require_file=False)
|
project_config = root / "mbx.project.yaml"
|
payload = _read_strict_payload(project_config, 4 * 1024 * 1024, "E_AUTH_TRUST")
|
text = payload.decode("utf-8", errors="strict")
|
if not re.search(r"(?m)^project:\s*\r?\n(?:^[ \t].*\r?\n)*?^ id: project-info\s*$", text) or not re.search(r"(?m)^ project_admin: project\.admin\s*$", text):
|
raise ProtocolError("E_AUTH_TRUST")
|
if not re.search(r"(?ms)^- id: project\.admin\s*$.*?^ private_dir: ai-infoadmin\s*$", text):
|
raise ProtocolError("E_AUTH_TRUST")
|
return root
|
|
|
def _relative_path(value: Any, prefixes: tuple[str, ...], code: str) -> str:
|
if not isinstance(value, str) or not value or "\\" in value or _CONTROL.search(value):
|
raise ProtocolError(code)
|
path = PurePosixPath(value)
|
if path.is_absolute() or any(part in {"", ".", ".."} for part in path.parts):
|
raise ProtocolError(code)
|
normalized = path.as_posix()
|
if normalized != value or not any(normalized.startswith(prefix) for prefix in prefixes):
|
raise ProtocolError(code)
|
return normalized
|
|
|
def _file_identity(root: Path, value: Any, prefixes: tuple[str, ...], code: str) -> tuple[Path, bytes]:
|
if not isinstance(value, dict) or set(value) != _FILE_ID_KEYS:
|
raise ProtocolError(code)
|
relative = _relative_path(value.get("relative_path"), prefixes, code)
|
size = value.get("bytes")
|
digest = value.get("sha256")
|
if (
|
isinstance(size, bool) or not isinstance(size, int) or size <= 0
|
or not isinstance(digest, str) or not _HEX64.fullmatch(digest)
|
or digest != digest.upper()
|
):
|
raise ProtocolError(code)
|
path = root / Path(relative)
|
payload = _read_strict_payload(path, max(size, 1) + 1, code)
|
if len(payload) != size or hashlib.sha256(payload).hexdigest().upper() != digest:
|
raise ProtocolError(code)
|
return path, payload
|
|
|
def _plain_mapping(value: Any, keys: set[str], code: str) -> dict[str, Any]:
|
if not isinstance(value, dict) or set(value) != keys:
|
raise ProtocolError(code)
|
return value
|
|
|
def _upper_sha(value: Any, code: str) -> str:
|
if not isinstance(value, str) or not _HEX64.fullmatch(value) or value != value.upper():
|
raise ProtocolError(code)
|
return value
|
|
|
def _positive_integer(value: Any, code: str) -> int:
|
if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
|
raise ProtocolError(code)
|
return value
|
|
|
def _extension_id_from_manifest_key(value: Any) -> str:
|
if not isinstance(value, str) or _CONTROL.search(value):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
try:
|
public_key = base64.b64decode(value, validate=True)
|
except (ValueError, TypeError) as exc:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY") from exc
|
alphabet = "abcdefghijklmnop"
|
prefix = hashlib.sha256(public_key).digest()[:16]
|
return "".join(alphabet[byte >> 4] + alphabet[byte & 15] for byte in prefix)
|
|
|
def _projection_tree(entries: list[dict[str, Any]]) -> str:
|
builder = bytearray()
|
for item in sorted(entries, key=lambda candidate: candidate["path"]):
|
builder.extend(item["path"].encode("utf-8"))
|
builder.extend(b"\0")
|
builder.extend(str(item["bytes"]).encode("ascii"))
|
builder.extend(b"\0")
|
builder.extend(item["sha256"].encode("ascii"))
|
builder.extend(b"\n")
|
return hashlib.sha256(builder).hexdigest().upper()
|
|
|
def _read_native_host_registry_default() -> str:
|
try:
|
import winreg
|
|
key_path = rf"Software\Google\Chrome\NativeMessagingHosts\com.project_info.bili_auth_ingress"
|
with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path, 0, winreg.KEY_READ) as key:
|
value, kind = winreg.QueryValueEx(key, None)
|
if kind != winreg.REG_SZ or not isinstance(value, str):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
return value
|
except (OSError, ImportError) as exc:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY") from exc
|
|
|
def _read_native_host_registry_exact() -> str:
|
"""Return the sole default REG_SZ value; reject extra values or subkeys."""
|
try:
|
import winreg
|
|
key_path = rf"Software\Google\Chrome\NativeMessagingHosts\com.project_info.bili_auth_ingress"
|
with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path, 0, winreg.KEY_READ) as key:
|
subkey_count, value_count, _ = winreg.QueryInfoKey(key)
|
if subkey_count != 0 or value_count != 1:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
name, value, kind = winreg.EnumValue(key, 0)
|
if name != "" or kind != winreg.REG_SZ or not isinstance(value, str):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
return value
|
except (OSError, ImportError) as exc:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY") from exc
|
|
|
def _fast_path_canonical_paths() -> dict[str, Path]:
|
return {
|
"queue": Path(r"C:\Users\Cai\AppData\Local\project-info\bili-auth-generic-runtime\queue.jsonl"),
|
"queue_state": Path(r"C:\Users\Cai\AppData\Local\project-info\bili-auth-generic-runtime\queue-state.jsonl"),
|
"reload_state": Path(r"C:\Users\Cai\AppData\Local\project-info\bili-auth-generic-runtime\reload-state.jsonl"),
|
"formal_manifest": Path(r"E:\mb-ms-doc\project-info\ana-data\news-青枫浦上Q\manifest.jsonl"),
|
}
|
|
|
def _successor_scope_sha256(successors: Any) -> str:
|
if not isinstance(successors, list):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
normalized: list[dict[str, Any]] = []
|
for value in successors:
|
value = _plain_mapping(value, _AUTH_SUCCESSOR_KEYS, "E_DEPLOYMENT_NOT_READY")
|
normalized.append({name: value[name] for name in sorted(_AUTH_SUCCESSOR_KEYS)})
|
normalized.sort(key=lambda value: (
|
value["creator_uid"], value["bvid"], value["predecessor_job_id"],
|
value["retry_generation"], value["terminal_error_code"],
|
))
|
payload = json.dumps(
|
normalized, ensure_ascii=True, sort_keys=True, separators=(",", ":"),
|
).encode("ascii")
|
return hashlib.sha256(payload).hexdigest().upper()
|
|
|
def _validate_prefix_identity(
|
path: Path, receipt: dict[str, Any], prefix: str, *, exact: bool,
|
) -> None:
|
code = "E_DEPLOYMENT_NOT_READY"
|
expected_path = _absolute_local_path(receipt[f"{prefix}_path"], code)
|
if path != expected_path:
|
raise ProtocolError(code)
|
size_name = f"{prefix}_bytes" if exact else f"{prefix}_prefix_bytes"
|
lines_name = f"{prefix}_lines" if exact else f"{prefix}_prefix_lines"
|
sha_name = f"{prefix}_sha256" if exact else f"{prefix}_prefix_sha256"
|
size = _positive_integer(receipt[size_name], code)
|
lines = _positive_integer(receipt[lines_name], code)
|
digest = _upper_sha(receipt[sha_name], code)
|
payload = _read_strict_payload(path, 32 * 1024 * 1024, code)
|
if exact:
|
candidate = payload
|
if len(payload) != size:
|
raise ProtocolError(code)
|
else:
|
if len(payload) < size:
|
raise ProtocolError(code)
|
candidate = payload[:size]
|
if (
|
not candidate.endswith(b"\n") or candidate.count(b"\n") != lines
|
or hashlib.sha256(candidate).hexdigest().upper() != digest
|
):
|
raise ProtocolError(code)
|
|
|
def _validate_review(value: Any, code: str) -> dict[str, Any]:
|
review = _plain_mapping(value, _REVIEW_KEYS, code)
|
if (
|
not isinstance(review["result_message_id"], str)
|
or not MESSAGE_ID_RE.fullmatch(review["result_message_id"])
|
or not isinstance(review["audit_id"], str) or not review["audit_id"].startswith("DEV-AUDIT-")
|
or _positive_integer(review["audit_bytes"], code) != review["audit_bytes"]
|
or _upper_sha(review["audit_sha256"], code) != review["audit_sha256"]
|
or review["verdict"] != "PASS/0" or review["blocking_findings"] != 0
|
):
|
raise ProtocolError(code)
|
return review
|
|
|
def _validate_audit_prefix(root: Path, review: dict[str, Any], code: str) -> None:
|
audit_path = root / "dev-doc" / "开发审计报告.md"
|
payload = _read_strict_payload(audit_path, 16 * 1024 * 1024, code)
|
size = review["audit_bytes"]
|
if len(payload) < size or hashlib.sha256(payload[:size]).hexdigest().upper() != review["audit_sha256"]:
|
raise ProtocolError(code)
|
try:
|
prefix = payload[:size].decode("utf-8", errors="strict")
|
except UnicodeError as exc:
|
raise ProtocolError(code) from exc
|
if review["audit_id"] not in prefix:
|
raise ProtocolError(code)
|
|
|
def _validate_projection(root: Path, deployment: dict[str, Any]) -> None:
|
contract_path = root / "dev" / "project-dev" / "bili_authenticated_extension_unpacked_contract.json"
|
contract_payload = _read_strict_payload(contract_path, 128 * 1024, "E_DEPLOYMENT_NOT_READY")
|
if hashlib.sha256(contract_payload).hexdigest().upper() != deployment["projection_contract_sha256"]:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
contract = strict_json_loads(contract_payload)
|
required = {
|
"schema", "task_id", "project_id", "source_root", "projection_root",
|
"expected_extension_id", "public_key_der_sha256", "tree_hash_algorithm",
|
"tree_sha256", "files",
|
}
|
if not isinstance(contract, dict) or set(contract) != required or contract.get("schema") != 1:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
if (
|
contract["task_id"] != _TASK_ID or contract["project_id"] != _PROJECT_ID
|
or contract["expected_extension_id"] != EXPECTED_EXTENSION_ID
|
or contract["tree_sha256"] != deployment["projection_tree_sha256"]
|
or contract["tree_hash_algorithm"] != "path-nul-bytes-nul-sha256-upper-lf-v1"
|
):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
source_root = root / Path(_relative_path(contract["source_root"], ("dev/project-dev/bili_authenticated_extension",), "E_DEPLOYMENT_NOT_READY"))
|
projection_root = root / Path(_relative_path(contract["projection_root"], ("dev/project-dev/bili_authenticated_extension_unpacked",), "E_DEPLOYMENT_NOT_READY"))
|
_check_existing_chain(source_root, require_file=False)
|
_check_existing_chain(projection_root, require_file=False)
|
files = contract["files"]
|
if not isinstance(files, list) or len(files) != 5:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
expected_paths = ["background.js", "manifest.json", "sidepanel.css", "sidepanel.html", "sidepanel.js"]
|
_enumerate_exact_tree(projection_root, set(expected_paths), "E_DEPLOYMENT_NOT_READY")
|
actual_paths: list[str] = []
|
entries: list[dict[str, Any]] = []
|
for spec in files:
|
if not isinstance(spec, dict) or set(spec) != {"path", "bytes", "sha256"}:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
name = spec["path"]
|
if name not in expected_paths or name in actual_paths:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
size = _positive_integer(spec["bytes"], "E_DEPLOYMENT_NOT_READY")
|
digest = _upper_sha(spec["sha256"], "E_DEPLOYMENT_NOT_READY")
|
payload = _read_strict_payload(projection_root / name, size + 1, "E_DEPLOYMENT_NOT_READY")
|
source_payload = _read_strict_payload(source_root / name, size + 1, "E_DEPLOYMENT_NOT_READY")
|
if len(payload) != size or payload != source_payload or hashlib.sha256(payload).hexdigest().upper() != digest:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
actual_paths.append(name)
|
entries.append({"path": name, "bytes": size, "sha256": digest})
|
if sorted(actual_paths) != expected_paths or _projection_tree(entries) != contract["tree_sha256"]:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
manifest = strict_json_loads((projection_root / "manifest.json").read_bytes())
|
if (
|
not isinstance(manifest, dict) or manifest.get("manifest_version") != 3
|
or manifest.get("version") != "1.2.25"
|
or manifest.get("version_name") != "1.2.25+20260829.generic.v027"
|
or _extension_id_from_manifest_key(manifest.get("key")) != EXPECTED_EXTENSION_ID
|
or EXTENSION_BUILD.encode("ascii") not in (projection_root / "background.js").read_bytes()
|
):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
|
|
def _validate_reload_applied(path: Path) -> None:
|
payload = _read_strict_payload(path, 4 * 1024 * 1024, "E_DEPLOYMENT_NOT_READY")
|
if not payload.endswith(b"\n"):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
matching: list[dict[str, Any]] = []
|
for line in payload.splitlines():
|
value = strict_json_loads(line)
|
if not isinstance(value, dict) or set(value) != {
|
"schema", "generation", "event", "token", "from_build", "to_build", "at_unix_ms"
|
}:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
if value.get("generation") == RELOAD_GENERATION:
|
matching.append(value)
|
events = [value.get("event") for value in matching]
|
if events not in (["APPLIED"], ["OFFERED", "BEGIN", "APPLIED"]):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
last = matching[-1]
|
if last.get("from_build") != EXTENSION_BUILD or last.get("to_build") != EXTENSION_BUILD:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
if (
|
any(value.get("schema") != 1 for value in matching)
|
or any(not isinstance(value.get("token"), str) or not re.fullmatch(r"[0-9a-f]{32}", value["token"]) for value in matching)
|
or len({value["token"] for value in matching}) != 1
|
or any(value.get("to_build") != EXTENSION_BUILD for value in matching)
|
or any(isinstance(value.get("at_unix_ms"), bool) or not isinstance(value.get("at_unix_ms"), int) or value["at_unix_ms"] <= 0 for value in matching)
|
):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
|
|
def _deployment_identity_branch(root: Path, deployment: dict[str, Any]) -> str:
|
code = "E_DEPLOYMENT_NOT_READY"
|
identities = {
|
name: _plain_mapping(deployment[name], _FILE_ID_KEYS, code)["relative_path"]
|
for name in ("build_approval", "install_approval", "install_receipt")
|
}
|
for relative in identities.values():
|
_relative_path(relative, ("ai-inforev/worklog/", "ai-infoadmin/worklog/"), code)
|
build_parent = PurePosixPath(identities["build_approval"]).parent.as_posix()
|
install_parent = PurePosixPath(identities["install_approval"]).parent.as_posix()
|
receipt_parent = PurePosixPath(identities["install_receipt"]).parent.as_posix()
|
if (
|
build_parent == "ai-inforev/worklog"
|
and install_parent == "ai-inforev/worklog"
|
and receipt_parent == "ai-infoadmin/worklog"
|
):
|
return "legacy-reviewer"
|
if (
|
build_parent == "ai-infoadmin/worklog"
|
and install_parent == "ai-infoadmin/worklog"
|
and receipt_parent == "ai-infoadmin/worklog"
|
and PurePosixPath(identities["build_approval"]).name.endswith("-fast-path-source-receipt.json")
|
and PurePosixPath(identities["install_approval"]).name.endswith("-fast-path-build-receipt.json")
|
and PurePosixPath(identities["install_receipt"]).name.endswith("-fast-path-install-readiness-receipt.json")
|
):
|
return "project-admin-fast-path"
|
raise ProtocolError(code)
|
|
|
def _validate_fast_path_host_source(
|
root: Path,
|
receipt: dict[str, Any],
|
current_source: dict[str, Any],
|
current_source_payload: bytes,
|
build_receipt: dict[str, Any],
|
install_approval: dict[str, Any],
|
) -> None:
|
code = "E_DEPLOYMENT_NOT_READY"
|
path, payload = _file_identity(
|
root, receipt["host_build_source_manifest"], ("ai-infoadmin/worklog/",), code,
|
)
|
if (
|
path.parent != root / "ai-infoadmin" / "worklog"
|
or not path.name.endswith("-fast-path-host-build-source-manifest.json")
|
):
|
raise ProtocolError(code)
|
host_source = _plain_mapping(strict_json_loads(payload), _SOURCE_MANIFEST_KEYS, code)
|
for name in _SOURCE_MANIFEST_KEYS - {"files"}:
|
if host_source[name] != current_source[name]:
|
raise ProtocolError(code)
|
current_files = current_source["files"]
|
host_files = host_source["files"]
|
if not isinstance(current_files, list) or not isinstance(host_files, list) or len(host_files) != 21:
|
raise ProtocolError(code)
|
current_by_path: dict[str, dict[str, Any]] = {}
|
host_by_path: dict[str, dict[str, Any]] = {}
|
for target, values in ((current_by_path, current_files), (host_by_path, host_files)):
|
for raw in values:
|
item = _plain_mapping(raw, _ARTIFACT_FILE_KEYS, code)
|
relative = _relative_path(item["path"], ("",), code)
|
if relative in target:
|
raise ProtocolError(code)
|
_positive_integer(item["bytes"], code)
|
_upper_sha(item["sha256"], code)
|
target[relative] = item
|
if set(current_by_path) != set(host_by_path):
|
raise ProtocolError(code)
|
changed = sorted(
|
relative for relative in current_by_path
|
if current_by_path[relative] != host_by_path[relative]
|
)
|
mode = receipt["host_source_binding_mode"]
|
if mode == "SAME_SOURCE_HOST_BUILD":
|
expected_changed: list[str] = []
|
if payload != current_source_payload:
|
raise ProtocolError(code)
|
elif mode == "PRODUCER_ONLY_DELTA":
|
expected_changed = ["queue_producer.py"]
|
if payload == current_source_payload:
|
raise ProtocolError(code)
|
else:
|
raise ProtocolError(code)
|
host_manifest_hash = hashlib.sha256(payload).hexdigest().upper()
|
if (
|
changed != expected_changed
|
or receipt["producer_only_changed_files"] != expected_changed
|
or receipt["host_archive_excluded_modules"] != ["bili_authenticated_extension.queue_producer"]
|
or "bili_authenticated_extension.queue_producer"
|
in build_receipt["archive_verification"]["required_modules"]
|
or build_receipt["source_artifact_manifest_bytes"] != len(payload)
|
or build_receipt["source_artifact_manifest_sha256"] != host_manifest_hash
|
or install_approval["source_artifact_manifest_sha256"] != host_manifest_hash
|
):
|
raise ProtocolError(code)
|
|
|
def _validate_fast_path_receipt(
|
root: Path,
|
config: dict[str, Any],
|
approval: dict[str, Any],
|
authorization: dict[str, Any],
|
deployment: dict[str, Any],
|
source: dict[str, Any],
|
build_receipt: dict[str, Any],
|
install_approval: dict[str, Any],
|
source_payload: bytes,
|
source_receipt_payload: bytes,
|
build_receipt_payload: bytes,
|
build_validation_receipt_payload: bytes,
|
exe_payload: bytes,
|
receipt: dict[str, Any],
|
actual: dict[str, tuple[Path, bytes]],
|
) -> None:
|
code = "E_DEPLOYMENT_NOT_READY"
|
owner_ai_id, owner_thread_id, owner_role = _FAST_PATH_OWNER
|
authorization_spec = _plain_mapping(approval["authorization_file"], _FILE_ID_KEYS, code)
|
review = _plain_mapping(approval["implementation_review"], _REVIEW_KEYS, code)
|
source_receipt_hash = hashlib.sha256(source_receipt_payload).hexdigest().upper()
|
build_validation_receipt_hash = hashlib.sha256(build_validation_receipt_payload).hexdigest().upper()
|
build_receipt_hash = hashlib.sha256(build_receipt_payload).hexdigest().upper()
|
exe_hash = hashlib.sha256(exe_payload).hexdigest().upper()
|
if (
|
receipt["schema"] != 2
|
or receipt["task_id"] != _TASK_ID
|
or receipt["validation_scope"] != "continuous-fast-path-installed-readiness-v1"
|
or receipt["validated_by_role"] != "project.admin"
|
or receipt["status"] != "VALIDATED"
|
or receipt["continuous_authorization_handoff_id"] != _CONTINUOUS_FAST_PATH_HANDOFF_ID
|
or receipt["owner_ai_id"] != owner_ai_id
|
or receipt["owner_thread_id"] != owner_thread_id
|
or receipt["owner_role_instance_id"] != owner_role
|
or receipt["authorization_file_relative_path"] != authorization_spec["relative_path"]
|
or receipt["authorization_file_bytes"] != authorization_spec["bytes"]
|
or receipt["authorization_file_sha256"] != authorization_spec["sha256"]
|
or receipt["successor_scope_sha256"] != _successor_scope_sha256(authorization["successors"])
|
or receipt["implementation_review_audit_id"] != review["audit_id"]
|
or receipt["implementation_review_audit_bytes"] != review["audit_bytes"]
|
or receipt["implementation_review_audit_sha256"] != review["audit_sha256"]
|
or receipt["source_artifact_manifest_bytes"] != len(source_payload)
|
or receipt["source_artifact_manifest_sha256"] != deployment["source_manifest_sha256"]
|
or receipt["source_receipt_bytes"] != len(source_receipt_payload)
|
or receipt["source_receipt_sha256"] != source_receipt_hash
|
or receipt["build_artifact_manifest_bytes"] != len(build_receipt_payload)
|
or receipt["build_artifact_manifest_sha256"] != build_receipt_hash
|
or receipt["build_receipt_bytes"] != len(build_validation_receipt_payload)
|
or receipt["build_receipt_sha256"] != build_validation_receipt_hash
|
or receipt["host_executable_bytes"] != len(exe_payload)
|
or receipt["host_executable_sha256"] != exe_hash
|
or receipt["native_messaging_host_manifest"] != str(actual["native-host-manifest.json"][0])
|
or receipt["projection_contract_sha256"] != deployment["projection_contract_sha256"]
|
or receipt["projection_tree_sha256"] != deployment["projection_tree_sha256"]
|
or receipt["secret_field_count"] != 0
|
):
|
raise ProtocolError(code)
|
for name in (
|
"authorization_file_sha256", "successor_scope_sha256",
|
"implementation_review_audit_sha256", "source_artifact_manifest_sha256",
|
"source_receipt_sha256", "build_artifact_manifest_sha256", "build_receipt_sha256",
|
"host_executable_sha256", "projection_contract_sha256", "projection_tree_sha256",
|
):
|
_upper_sha(receipt[name], code)
|
for name in (
|
"authorization_file_bytes", "implementation_review_audit_bytes",
|
"source_artifact_manifest_bytes", "source_receipt_bytes",
|
"build_artifact_manifest_bytes", "build_receipt_bytes", "host_executable_bytes",
|
):
|
_positive_integer(receipt[name], code)
|
_reject_secret_keys(receipt)
|
_validate_fast_path_host_source(
|
root, receipt, source, source_payload, build_receipt, install_approval,
|
)
|
canonical = _fast_path_canonical_paths()
|
if (
|
config["queue_paths"]["queue_path"] != canonical["queue"]
|
or config["queue_paths"]["queue_state_path"] != canonical["queue_state"]
|
or config["queue_paths"]["reload_state_path"] != canonical["reload_state"]
|
):
|
raise ProtocolError(code)
|
_validate_prefix_identity(canonical["reload_state"], receipt, "reload_state", exact=True)
|
_validate_prefix_identity(canonical["queue"], receipt, "queue", exact=False)
|
_validate_prefix_identity(canonical["queue_state"], receipt, "queue_state", exact=False)
|
_validate_prefix_identity(canonical["formal_manifest"], receipt, "formal_manifest", exact=False)
|
if Path(_read_native_host_registry_exact()) != actual["native-host-manifest.json"][0]:
|
raise ProtocolError(code)
|
|
|
def _load_release_approval(config: dict[str, Any]) -> ReleaseApproval:
|
if config.get("schema") != 2:
|
raise ProtocolError("E_AUTH_TRUST")
|
authorization_message_id = config.get("successor_authorization_message_id")
|
if not isinstance(authorization_message_id, str) or not MESSAGE_ID_RE.fullmatch(authorization_message_id):
|
raise ProtocolError("E_AUTH_TRUST")
|
root = _project_root_from_source()
|
if config["project_root"] != root:
|
raise ProtocolError("E_AUTH_TRUST")
|
approval_path = root / "ai-infoadmin" / "worklog" / f"bili-auth-successor-release-approval-{authorization_message_id}.json"
|
approval_payload = _read_strict_payload(approval_path, 256 * 1024, "E_AUTH_TRUST")
|
approval = strict_json_loads(approval_payload)
|
_plain_mapping(approval, _RELEASE_KEYS, "E_AUTH_TRUST")
|
if (
|
approval["schema"] != 1 or approval["scope"] != "bili-auth-successor-exact-release-v1"
|
or approval["project_id"] != _PROJECT_ID or approval["task_id"] != _TASK_ID
|
or approval["approved_by_role"] != "project.admin"
|
or approval["authorization_message_id"] != authorization_message_id
|
or not isinstance(approval["authorization_handoff_id"], str)
|
or not approval["authorization_handoff_id"].startswith("HANDOFF-")
|
):
|
raise ProtocolError("E_AUTH_TRUST")
|
implementation_review = _validate_review(approval["implementation_review"], "E_AUTH_TRUST")
|
_validate_audit_prefix(root, implementation_review, "E_AUTH_TRUST")
|
authorization_path, authorization_payload = _file_identity(
|
root, approval["authorization_file"], ("ai-infoadmin/worklog/",), "E_AUTH_TRUST"
|
)
|
if authorization_path.parent != approval_path.parent:
|
raise ProtocolError("E_AUTH_TRUST")
|
authorization = strict_json_loads(authorization_payload)
|
_plain_mapping(authorization, _AUTH_KEYS, "E_AUTH_TRUST")
|
if (
|
authorization["schema"] != 1 or authorization["scope"] != "bili-auth-successor-lineage-v1"
|
or authorization["task_id"] != _TASK_ID or authorization["authorized_by_role"] != "project.admin"
|
or authorization["authorization_message_id"] != authorization_message_id
|
or authorization["authorization_handoff_id"] != approval["authorization_handoff_id"]
|
):
|
raise ProtocolError("E_AUTH_TRUST")
|
repair = _validate_review({
|
"result_message_id": authorization["repair"].get("review_result_message_id") if isinstance(authorization.get("repair"), dict) else None,
|
"audit_id": authorization["repair"].get("audit_id") if isinstance(authorization.get("repair"), dict) else None,
|
"audit_bytes": authorization["repair"].get("audit_bytes") if isinstance(authorization.get("repair"), dict) else None,
|
"audit_sha256": authorization["repair"].get("audit_sha256") if isinstance(authorization.get("repair"), dict) else None,
|
"verdict": authorization["repair"].get("verdict") if isinstance(authorization.get("repair"), dict) else None,
|
"blocking_findings": authorization["repair"].get("blocking_findings") if isinstance(authorization.get("repair"), dict) else None,
|
}, "E_AUTH_TRUST")
|
if not isinstance(authorization["repair"], dict) or set(authorization["repair"]) != _AUTH_REPAIR_KEYS:
|
raise ProtocolError("E_AUTH_TRUST")
|
_validate_audit_prefix(root, repair, "E_AUTH_TRUST")
|
raw_successors = authorization["successors"]
|
if not isinstance(raw_successors, list) or not 1 <= len(raw_successors) <= 100:
|
raise ProtocolError("E_AUTH_TRUST")
|
successors: list[AuthorizedSuccessor] = []
|
for raw in raw_successors:
|
_plain_mapping(raw, _AUTH_SUCCESSOR_KEYS, "E_AUTH_TRUST")
|
successors.append(AuthorizedSuccessor(
|
creator_uid=raw["creator_uid"], bvid=raw["bvid"],
|
predecessor_job_id=raw["predecessor_job_id"],
|
retry_generation=raw["retry_generation"], terminal_error_code=raw["terminal_error_code"],
|
))
|
release = ReleaseApproval(
|
authorization_message_id=authorization_message_id,
|
authorization_handoff_id=approval["authorization_handoff_id"],
|
authorization_sha256=hashlib.sha256(authorization_payload).hexdigest().upper(),
|
repair_review_result_message_id=repair["result_message_id"],
|
repair_audit_id=repair["audit_id"], repair_audit_bytes=repair["audit_bytes"],
|
repair_audit_sha256=repair["audit_sha256"], successors=tuple(successors),
|
)
|
QueueStore._validate_release(release)
|
_validate_deployment(root, config, approval, authorization)
|
return release
|
|
|
def _validate_deployment(
|
root: Path, config: dict[str, Any], approval: dict[str, Any], authorization: dict[str, Any]
|
) -> None:
|
deployment = _plain_mapping(approval["deployment"], _DEPLOYMENT_KEYS, "E_DEPLOYMENT_NOT_READY")
|
if (
|
deployment["extension_build"] != EXTENSION_BUILD
|
or deployment["host_build"] != HOST_BUILD
|
or deployment["reload_generation"] != RELOAD_GENERATION
|
or deployment["extension_id"] != EXPECTED_EXTENSION_ID
|
):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
for name in (
|
"source_manifest_sha256", "installed_config_sha256", "installed_manifest_sha256",
|
"installed_exe_sha256", "projection_contract_sha256", "projection_tree_sha256",
|
):
|
_upper_sha(deployment[name], "E_DEPLOYMENT_NOT_READY")
|
source_manifest_path = root / "dev" / "project-dev" / "bili_authenticated_extension" / "source-artifact-manifest.json"
|
source_payload = _read_strict_payload(source_manifest_path, 256 * 1024, "E_DEPLOYMENT_NOT_READY")
|
if hashlib.sha256(source_payload).hexdigest().upper() != deployment["source_manifest_sha256"]:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
source = _validate_source_tree(root, source_payload, deployment)
|
branch = _deployment_identity_branch(root, deployment)
|
approval_prefix = ("ai-inforev/worklog/",) if branch == "legacy-reviewer" else ("ai-infoadmin/worklog/",)
|
_, build_approval_payload = _file_identity(root, deployment["build_approval"], approval_prefix, "E_DEPLOYMENT_NOT_READY")
|
_, build_receipt_payload = _file_identity(root, deployment["build_receipt"], ("dev/tmp/",), "E_DEPLOYMENT_NOT_READY")
|
_, exe_payload = _file_identity(root, deployment["exe"], ("dev/tmp/",), "E_DEPLOYMENT_NOT_READY")
|
_, install_approval_payload = _file_identity(root, deployment["install_approval"], approval_prefix, "E_DEPLOYMENT_NOT_READY")
|
_, install_receipt_payload = _file_identity(root, deployment["install_receipt"], ("ai-infoadmin/worklog/",), "E_DEPLOYMENT_NOT_READY")
|
build_approval = strict_json_loads(build_approval_payload)
|
build_receipt = strict_json_loads(build_receipt_payload)
|
install_approval = strict_json_loads(install_approval_payload)
|
install_receipt = strict_json_loads(install_receipt_payload)
|
build_approval = _plain_mapping(build_approval, _BUILD_APPROVAL_KEYS, "E_DEPLOYMENT_NOT_READY")
|
build_receipt = _plain_mapping(build_receipt, _BUILD_RECEIPT_KEYS, "E_DEPLOYMENT_NOT_READY")
|
install_approval = _plain_mapping(install_approval, _INSTALL_APPROVAL_KEYS, "E_DEPLOYMENT_NOT_READY")
|
install_receipt = _plain_mapping(
|
install_receipt,
|
_INSTALL_RECEIPT_KEYS if branch == "legacy-reviewer" else _FAST_PATH_INSTALL_RECEIPT_KEYS,
|
"E_DEPLOYMENT_NOT_READY",
|
)
|
exe_hash = hashlib.sha256(exe_payload).hexdigest().upper()
|
build_receipt_hash = hashlib.sha256(build_receipt_payload).hexdigest().upper()
|
expected_role = "dev.reviewer.project" if branch == "legacy-reviewer" else "project.admin"
|
if (
|
build_approval["schema"] != 1 or build_approval["task_id"] != _TASK_ID
|
or build_approval["approval_scope"] != "controlled-build-source-manifest"
|
or build_approval["approved_by_role"] != expected_role or build_approval["status"] != "APPROVED"
|
or build_approval["source_artifact_manifest_bytes"] != len(source_payload)
|
or build_approval["source_artifact_manifest_sha256"] != deployment["source_manifest_sha256"]
|
or build_receipt["schema"] != 2 or build_receipt["scope"] != "generic-bilibili-queue"
|
or build_receipt["extension_id"] != EXPECTED_EXTENSION_ID
|
or build_receipt["extension_build"] != EXTENSION_BUILD or build_receipt["host_build"] != HOST_BUILD
|
or build_receipt["packaging"] != "pyinstaller-onefile"
|
or build_receipt["pyinstaller_version"] != "6.15.0" or build_receipt["yt_dlp_version"] != "2026.7.4"
|
or (
|
branch == "legacy-reviewer"
|
and (
|
build_receipt["source_artifact_manifest_bytes"] != len(source_payload)
|
or build_receipt["source_artifact_manifest_sha256"] != deployment["source_manifest_sha256"]
|
)
|
)
|
or build_receipt["dependency_artifact_manifest_bytes"] != source["dependency_artifact_manifest_bytes"]
|
or build_receipt["dependency_artifact_manifest_sha256"] != source["dependency_artifact_manifest_sha256"]
|
or install_approval["schema"] != 1 or install_approval["task_id"] != _TASK_ID
|
or install_approval["approval_scope"] != "install-exact-build"
|
or install_approval["approved_by_role"] != expected_role or install_approval["status"] != "APPROVED"
|
or (
|
branch == "legacy-reviewer"
|
and install_approval["source_artifact_manifest_sha256"] != deployment["source_manifest_sha256"]
|
)
|
or install_approval["build_artifact_manifest_bytes"] != len(build_receipt_payload)
|
or install_approval["build_artifact_manifest_sha256"] != build_receipt_hash
|
or install_approval["host_executable_bytes"] != len(exe_payload)
|
or install_approval["host_executable_sha256"] != exe_hash
|
or install_receipt["schema"] != (1 if branch == "legacy-reviewer" else 2)
|
or install_receipt["task_id"] != _TASK_ID
|
or install_receipt["host_build"] != HOST_BUILD or install_receipt["required_extension_build"] != EXTENSION_BUILD
|
or install_receipt["extension_id"] != EXPECTED_EXTENSION_ID
|
or install_receipt["host_name"] != "com.project_info.bili_auth_ingress"
|
):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
for name in (
|
"builder_python_sha256", "pyinstaller_executable_sha256",
|
"builder_provision_receipt_sha256", "build_script_sha256",
|
"yt_dlp_wheel_sha256",
|
):
|
_upper_sha(build_receipt[name], "E_DEPLOYMENT_NOT_READY")
|
for name in (
|
"pyinstaller_executable_bytes", "builder_provision_receipt_bytes",
|
):
|
_positive_integer(build_receipt[name], "E_DEPLOYMENT_NOT_READY")
|
archive = _plain_mapping(build_receipt["archive_verification"], _ARCHIVE_VERIFICATION_KEYS, "E_DEPLOYMENT_NOT_READY")
|
if (
|
archive["status"] != "PASS"
|
or not isinstance(archive["method"], str) or not archive["method"] or _CONTROL.search(archive["method"])
|
or archive["required_modules"] != [
|
"bili_authenticated_extension.worker", "yt_dlp", "yt_dlp.downloader",
|
"yt_dlp.globals", "yt_dlp.plugins", "yt_dlp.version",
|
]
|
or archive["metadata_entry"] != "yt_dlp-2026.7.4.dist-info/METADATA"
|
or archive["metadata_files"] != 7
|
or not isinstance(archive["metadata_type_codes"], list)
|
or not archive["metadata_type_codes"]
|
or any(value not in {"b", "x"} for value in archive["metadata_type_codes"])
|
or len(set(archive["metadata_type_codes"])) != len(archive["metadata_type_codes"])
|
or archive["metadata_tree_sha256"] != source["archive_metadata_contract"]["canonical_tree_sha256"]
|
):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
files = build_receipt.get("files")
|
if not isinstance(files, list) or len(files) != 1:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
build_file = _plain_mapping(files[0], _ARTIFACT_FILE_KEYS, "E_DEPLOYMENT_NOT_READY")
|
if (
|
build_file["path"] != "project-info-bili-auth-native-host.exe"
|
or build_file["bytes"] != len(exe_payload) or build_file["sha256"] != exe_hash
|
):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
installed_root = _absolute_local_path(install_receipt.get("installed_root"), "E_DEPLOYMENT_NOT_READY")
|
installed_files = install_receipt.get("installed_files")
|
if not isinstance(installed_files, list) or {item.get("path") for item in installed_files if isinstance(item, dict)} != {
|
"project-info-bili-auth-native-host.exe", "config.json", "native-host-manifest.json"
|
}:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
actual: dict[str, tuple[Path, bytes]] = {}
|
for item in installed_files:
|
item = _plain_mapping(item, _ARTIFACT_FILE_KEYS, "E_DEPLOYMENT_NOT_READY")
|
size = _positive_integer(item["bytes"], "E_DEPLOYMENT_NOT_READY")
|
digest = _upper_sha(item["sha256"], "E_DEPLOYMENT_NOT_READY")
|
candidate = installed_root / item["path"]
|
payload = _read_strict_payload(candidate, size + 1, "E_DEPLOYMENT_NOT_READY")
|
if len(payload) != size or hashlib.sha256(payload).hexdigest().upper() != digest:
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
actual[item["path"]] = (candidate, payload)
|
if (
|
hashlib.sha256(actual["config.json"][1]).hexdigest().upper() != config.get("_host_config_sha256")
|
or hashlib.sha256(actual["config.json"][1]).hexdigest().upper() != deployment["installed_config_sha256"]
|
or hashlib.sha256(actual["native-host-manifest.json"][1]).hexdigest().upper() != deployment["installed_manifest_sha256"]
|
or hashlib.sha256(actual["project-info-bili-auth-native-host.exe"][1]).hexdigest().upper() != deployment["installed_exe_sha256"]
|
or actual["project-info-bili-auth-native-host.exe"][1] != exe_payload
|
or deployment["installed_exe_sha256"] != exe_hash
|
or actual["config.json"][0] != config["host_config_path"]
|
):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
installed_config = _plain_mapping(
|
strict_json_loads(actual["config.json"][1]), _HOST_KEYS, "E_DEPLOYMENT_NOT_READY"
|
)
|
if (
|
installed_config["schema"] != 2
|
or installed_config["required_extension_build"] != EXTENSION_BUILD
|
or installed_config["reload_generation"] != RELOAD_GENERATION
|
or installed_config["creator_allowlist"] != sorted(set(installed_config["creator_allowlist"]))
|
or config.get("creator_uid") not in installed_config["creator_allowlist"]
|
or any(
|
installed_config[name] != str(config["queue_paths"][name])
|
for name in ("queue_path", "queue_state_path", "queue_lock_path", "reload_state_path")
|
)
|
):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
native_manifest = strict_json_loads(actual["native-host-manifest.json"][1])
|
if (
|
not isinstance(native_manifest, dict) or native_manifest.get("name") != "com.project_info.bili_auth_ingress"
|
or native_manifest.get("type") != "stdio" or native_manifest.get("allowed_origins") != [EXPECTED_ORIGIN]
|
or Path(native_manifest.get("path", "")) != actual["project-info-bili-auth-native-host.exe"][0]
|
or Path(_read_native_host_registry_default()) != actual["native-host-manifest.json"][0]
|
):
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
_validate_projection(root, deployment)
|
_validate_reload_applied(config["queue_paths"]["reload_state_path"])
|
if branch == "project-admin-fast-path":
|
_validate_fast_path_receipt(
|
root, config, approval, authorization, deployment, source, build_receipt,
|
install_approval, source_payload,
|
build_approval_payload, build_receipt_payload, install_approval_payload,
|
exe_payload, install_receipt, actual,
|
)
|
|
|
def _formal_record(event: dict[str, Any], config: dict[str, Any]) -> tuple[str, dict[str, Any] | None, bool] | None:
|
schema = event.get("schema_version")
|
if schema == 1:
|
if event.get("creator") != config["creator_name"]:
|
raise ProtocolError("E_CATALOG")
|
uid = event.get("creator_uid")
|
if uid is not None and str(uid) != config["creator_uid"]:
|
raise ProtocolError("E_ALLOWLIST")
|
if event.get("item_type") != "video":
|
return None
|
bvid = event.get("stable_id")
|
status = event.get("status")
|
if not isinstance(status, str) or not status.startswith("VIDEO_"):
|
raise ProtocolError("E_CATALOG")
|
terminal = status in _TERMINAL_VIDEO_STATUSES or event.get("video_path") is not None
|
duration = event.get("expected_duration_seconds")
|
elif schema == 2:
|
if event.get("creator") != config["creator_name"] or str(event.get("creator_uid")) != config["creator_uid"]:
|
raise ProtocolError("E_ALLOWLIST")
|
if event.get("event_type") != "DYNAMIC_CONTENT_SAVED" or event.get("status") != "SAVED":
|
raise ProtocolError("E_CATALOG")
|
if event.get("content_type") != "video":
|
return None
|
bvid = event.get("bvid")
|
terminal = False
|
duration = event.get("duration_seconds")
|
else:
|
raise ProtocolError("E_CATALOG")
|
try:
|
bvid = validate_bvid(bvid)
|
except ValueError as exc:
|
raise ProtocolError("E_CATALOG") from exc
|
allowlist = config["bvid_allowlist"]
|
if allowlist is not None and bvid not in allowlist:
|
return bvid, None, terminal
|
if event.get("source_url") != canonical_url(bvid):
|
raise ProtocolError("E_CATALOG")
|
if terminal:
|
return bvid, None, True
|
required = (event.get("title"), event.get("published_at"), event.get("collected_at"), duration)
|
if any(value is None for value in required):
|
return bvid, None, terminal
|
record = {
|
"schema": 1,
|
"bvid": bvid,
|
"creator_uid": config["creator_uid"],
|
"expected_duration_ms": _duration_ms(duration),
|
"discovered_at_unix_ms": _time_ms(event["collected_at"]),
|
"published_at": _safe_text(event["published_at"], 64),
|
"title": _safe_text(event["title"]),
|
}
|
validate_ingress_record(record, frozenset({config["creator_uid"]}))
|
return bvid, record, terminal
|
|
|
def load_dynamic_manifest(config: dict[str, Any]) -> tuple[dict[str, dict[str, Any]], set[str], str]:
|
path = config["dynamic_manifest_path"]
|
payload = path.read_bytes()
|
if len(payload) > 16 * 1024 * 1024 or (payload and not payload.endswith(b"\n")):
|
raise ProtocolError("E_CATALOG")
|
records: dict[str, dict[str, Any]] = {}
|
terminal: set[str] = set()
|
seen: set[str] = set()
|
for line in payload.splitlines():
|
if not line or len(line) > 1024 * 1024:
|
raise ProtocolError("E_CATALOG")
|
event = strict_json_loads(line)
|
if not isinstance(event, dict):
|
raise ProtocolError("E_CATALOG")
|
_reject_secret_keys(event)
|
normalized = _formal_record(event, config)
|
if normalized is None:
|
continue
|
bvid, record, is_terminal = normalized
|
seen.add(bvid)
|
if is_terminal:
|
terminal.add(bvid)
|
if record is None:
|
continue
|
previous = records.get(bvid)
|
if previous is None:
|
records[bvid] = record
|
continue
|
left = {key: value for key, value in previous.items() if key != "discovered_at_unix_ms"}
|
right = {key: value for key, value in record.items() if key != "discovered_at_unix_ms"}
|
if left != right:
|
raise ProtocolError("E_CATALOG_CONFLICT")
|
previous["discovered_at_unix_ms"] = min(previous["discovered_at_unix_ms"], record["discovered_at_unix_ms"])
|
return records, terminal, hashlib.sha256(payload).hexdigest().upper()
|
|
|
def load_registered_catalog(config: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
payload = config["_registered_catalog_payload"]
|
if payload is None:
|
return {}
|
raw = strict_json_loads(payload)
|
if not isinstance(raw, dict) or set(raw) != _REGISTERED_KEYS or raw.get("schema") != 1 or raw.get("source") != "bili-dynamic-collector-registered-v1":
|
raise ProtocolError("E_CATALOG")
|
_reject_secret_keys(raw)
|
if raw.get("creator_name") != config["creator_name"] or raw.get("creator_uid") != config["creator_uid"]:
|
raise ProtocolError("E_ALLOWLIST")
|
items = raw.get("items")
|
if not isinstance(items, list) or not 1 <= len(items) <= 10_000:
|
raise ProtocolError("E_CATALOG")
|
records: dict[str, dict[str, Any]] = {}
|
for value in items:
|
if not isinstance(value, dict) or set(value) != _INGRESS_KEYS:
|
raise ProtocolError("E_CATALOG")
|
job = validate_ingress_record(value, frozenset({config["creator_uid"]}))
|
if job["creator_uid"] != config["creator_uid"]:
|
raise ProtocolError("E_ALLOWLIST")
|
allowlist = config["bvid_allowlist"]
|
if allowlist is not None and job["bvid"] not in allowlist:
|
continue
|
previous = records.get(job["bvid"])
|
if previous is not None and previous != value:
|
raise ProtocolError("E_CATALOG_CONFLICT")
|
records.setdefault(job["bvid"], value)
|
return records
|
|
|
def collect_jobs(config: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
records, terminal, dynamic_hash = load_dynamic_manifest(config)
|
registered = load_registered_catalog(config)
|
for bvid, record in registered.items():
|
previous = records.get(bvid)
|
if previous is not None and previous != record:
|
raise ProtocolError("E_CATALOG_CONFLICT")
|
records.setdefault(bvid, record)
|
allowlist = config["bvid_allowlist"]
|
if allowlist is not None:
|
missing = set(allowlist).difference(records, terminal)
|
if missing:
|
raise ProtocolError("E_CATALOG_MISSING")
|
jobs = [records[bvid] for bvid in sorted(records) if bvid not in terminal]
|
if not jobs:
|
raise ProtocolError("E_NO_JOBS")
|
return jobs, {
|
"dynamic_manifest_sha256": dynamic_hash,
|
"registered_catalog_sha256": config["_registered_catalog_sha256"],
|
"catalog_jobs": len(records),
|
"terminal_skipped": len(set(records).intersection(terminal)),
|
}
|
|
|
def run(config_path: Path, *, append: bool, append_successors: bool = False) -> dict[str, Any]:
|
if append and append_successors:
|
raise ProtocolError("E_CONFIG")
|
config = load_producer_configuration(config_path)
|
jobs, evidence = collect_jobs(config)
|
paths = config["queue_paths"]
|
result: dict[str, int] = {"appended": 0, "unchanged": 0}
|
if append or append_successors:
|
runtime_root = paths["queue_path"].parent
|
if append_successors:
|
_check_existing_chain(runtime_root, require_file=False)
|
_check_existing_chain(paths["queue_lock_path"], require_file=True)
|
if not runtime_root.is_dir():
|
raise ProtocolError("E_DEPLOYMENT_NOT_READY")
|
else:
|
runtime_root.mkdir(parents=True, exist_ok=True)
|
_check_existing_chain(runtime_root, require_file=False)
|
store = QueueStore(
|
paths["queue_path"], paths["queue_state_path"], paths["queue_lock_path"],
|
frozenset({config["creator_uid"]}),
|
)
|
if append_successors:
|
result = store.append_authorized_successors(lambda: _load_release_approval(config), jobs)
|
else:
|
result = store.append_ingress_jobs(jobs)
|
return {
|
"schema": 2 if append_successors else 1,
|
"producer_id": PRODUCER_ID,
|
"status": (
|
"SUCCESSOR_APPENDED" if append_successors and result["appended"]
|
else "NO_CHANGE" if append_successors
|
else "APPENDED" if append and result["appended"]
|
else "NO_CHANGE" if append
|
else "VALIDATION_PASS_ONLY"
|
),
|
"job_count": len(jobs),
|
**result,
|
**evidence,
|
}
|
|
|
def _parser() -> argparse.ArgumentParser:
|
parser = argparse.ArgumentParser(description="Produce governed Bilibili authenticated queue records.")
|
parser.add_argument("--config", required=True, type=Path)
|
action = parser.add_mutually_exclusive_group()
|
action.add_argument("--append", action="store_true", help="Append initial schema-1 jobs after validation")
|
action.add_argument("--append-successors", action="store_true", help="Append one admin-approved schema-2 successor block")
|
return parser
|
|
|
def main(argv: list[str] | None = None) -> int:
|
try:
|
arguments = _parser().parse_args(argv)
|
result = run(
|
arguments.config, append=arguments.append,
|
append_successors=arguments.append_successors,
|
)
|
except ProtocolError as exc:
|
print(json.dumps({"schema": 1, "status": "FAILED", "error_code": exc.code}, sort_keys=True, separators=(",", ":")))
|
return 3
|
except (OSError, KeyError, TypeError, ValueError):
|
print(json.dumps({"schema": 1, "status": "FAILED", "error_code": "E_PRODUCER"}, sort_keys=True, separators=(",", ":")))
|
return 3
|
print(json.dumps(result, ensure_ascii=False, sort_keys=True, separators=(",", ":")))
|
return 0
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
