#!/usr/bin/env python3
|
"""Local Bilibili dynamic collection coordinator.
|
|
This tool deliberately has no network or browser integration. It consumes
|
metadata exported by a person/browser extension and coordinates local files.
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
import hashlib
|
import json
|
import os
|
import re
|
import sys
|
import uuid
|
from dataclasses import dataclass
|
from datetime import datetime, timedelta, timezone
|
from pathlib import Path
|
from typing import Any, Iterable, Mapping, Sequence
|
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
# Keep lazy refresh imports on the same module identity when this file is run as
|
# a script, so CollectorError and dataclass contracts are not duplicated.
|
if __name__ == "__main__":
|
sys.modules.setdefault("bili_dynamic_collector", sys.modules[__name__])
|
|
|
SCHEMA_VERSION = 1
|
MANIFEST_SCHEMA_VERSION = 1
|
SECRET_KEY_PATTERN = re.compile(
|
r"(?:password|passwd|cookie|token|secret|authorization|captcha|session|"
|
r"口令|密码|令牌|验证码|会话)",
|
re.IGNORECASE,
|
)
|
WINDOWS_INVALID_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
|
WINDOWS_RESERVED_NAMES = {
|
"CON",
|
"PRN",
|
"AUX",
|
"NUL",
|
*(f"COM{i}" for i in range(1, 10)),
|
*(f"LPT{i}" for i in range(1, 10)),
|
}
|
CONTENT_TYPE_LABELS = {
|
"text": "文字",
|
"article": "专栏",
|
"image": "图片",
|
"video": "视频",
|
}
|
TODO_ACTIONS = {
|
"text": "EXPORT_FULL_TEXT_AS_UTF8_TXT",
|
"article": "EXPORT_FULL_ARTICLE_AS_UTF8_TXT",
|
"image": "DOWNLOAD_ORIGINAL_IMAGES",
|
"video": "OPEN_PLAY_PAGE_AND_USE_INSTALLED_EXTENSION",
|
}
|
ACTIVE_OR_SUCCESS_STATUSES = {
|
"TODO_QUEUED",
|
"VIDEO_MOVED",
|
"VIDEO_MOVED_SOURCE_RETAINED",
|
"PROCESSING_HANDOFF_CONFIRMED",
|
"PROCESSING",
|
"COMPLETE",
|
"CONTENT_SAVED",
|
}
|
RETRYABLE_STATUSES = {
|
"QUEUE_FAILED",
|
"MOVE_FAILED",
|
"PROCESSING_FAILED",
|
}
|
TEMP_DOWNLOAD_SUFFIXES = {
|
".crdownload",
|
".part",
|
".partial",
|
".tmp",
|
".download",
|
}
|
DEFAULT_VIDEO_EXTENSIONS = {".mp4", ".mkv", ".mov", ".webm"}
|
FILE_ATTRIBUTE_REPARSE_POINT = 0x400
|
THREAD_ID_PATTERN = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}")
|
LOWER_SHA256_PATTERN = re.compile(r"[0-9a-f]{64}")
|
INTERNAL_ENTITY_ID_PATTERN = re.compile(r"[0-9a-f]{24}")
|
CONTROL_CHARACTER_PATTERN = re.compile(r"[\x00-\x1f\x7f]")
|
|
|
class CollectorError(Exception):
|
"""Expected contract or safety failure."""
|
|
def __init__(
|
self,
|
code: str,
|
message: str,
|
*,
|
details: Mapping[str, Any] | None = None,
|
safety: bool = False,
|
) -> None:
|
super().__init__(message)
|
self.code = code
|
self.message = message
|
self.details = dict(details or {})
|
self.safety = safety
|
|
|
@dataclass(frozen=True)
|
class NativeHandoffRoute:
|
project_id: str
|
source_ai_id: str
|
source_thread_id: str
|
source_role_instance_id: str
|
target_ai_id: str
|
target_thread_id: str
|
target_role_instance_id: str
|
reply_thread_id: str
|
|
def as_dict(self) -> dict[str, str]:
|
return {
|
"project_id": self.project_id,
|
"source_ai_id": self.source_ai_id,
|
"source_thread_id": self.source_thread_id,
|
"source_role_instance_id": self.source_role_instance_id,
|
"target_ai_id": self.target_ai_id,
|
"target_thread_id": self.target_thread_id,
|
"target_role_instance_id": self.target_role_instance_id,
|
"reply_thread_id": self.reply_thread_id,
|
}
|
|
|
@dataclass(frozen=True)
|
class RefreshConfig:
|
archive_dir: Path
|
formal_manifest: Path
|
intake_dir: Path
|
overall_deadline_seconds: int
|
refresh_action_timeout_seconds: int
|
observation_timeout_seconds: int
|
page_internal_settle_timeout_seconds: int
|
max_refresh_count: int
|
run_history_slots: int
|
max_items: int
|
max_images_per_item: int
|
max_image_bytes: int
|
max_text_bytes: int
|
|
|
@dataclass(frozen=True)
|
class CollectorConfig:
|
creator_name: str
|
creator_uid: str | None
|
creator_dynamic_url: str
|
timezone_name: str
|
window_hours: int
|
state_dir: Path
|
download_dir: Path
|
video_dir: Path
|
minimum_complete_age_seconds: int
|
title_max_length: int
|
allowed_source_hosts: frozenset[str]
|
allowed_video_extensions: frozenset[str]
|
native_handoff: NativeHandoffRoute
|
refresh: RefreshConfig | None = None
|
|
@property
|
def manifest_path(self) -> Path:
|
return self.state_dir / "manifest.jsonl"
|
|
@property
|
def queues_dir(self) -> Path:
|
return self.state_dir / "queues"
|
|
@property
|
def handoffs_dir(self) -> Path:
|
return self.state_dir / "handoffs"
|
|
@property
|
def lock_path(self) -> Path:
|
return self.state_dir / ".collector.lock"
|
|
@property
|
def tz(self) -> ZoneInfo:
|
return ZoneInfo(self.timezone_name)
|
|
|
class StateLock:
|
"""Kernel-backed fail-fast lock; a dead process cannot strand ownership."""
|
|
def __init__(self, path: Path) -> None:
|
self.path = path
|
self.fd: int | None = None
|
self.owner_path = path.parent / f"{path.name}.owner.json"
|
|
def _try_kernel_lock(self) -> None:
|
assert self.fd is not None
|
if os.name == "nt":
|
import msvcrt
|
|
os.lseek(self.fd, 0, os.SEEK_SET)
|
try:
|
msvcrt.locking(self.fd, msvcrt.LK_NBLCK, 1)
|
except OSError as exc:
|
raise CollectorError("E_STATE_LOCKED", "Another collector invocation is active.", safety=True) from exc
|
else: # pragma: no cover - exercised by non-Windows CI only
|
import fcntl
|
|
try:
|
fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
except OSError as exc:
|
raise CollectorError("E_STATE_LOCKED", "Another collector invocation is active.", safety=True) from exc
|
|
def _unlock(self) -> None:
|
assert self.fd is not None
|
if os.name == "nt":
|
import msvcrt
|
|
os.lseek(self.fd, 0, os.SEEK_SET)
|
msvcrt.locking(self.fd, msvcrt.LK_UNLCK, 1)
|
else: # pragma: no cover
|
import fcntl
|
|
fcntl.flock(self.fd, fcntl.LOCK_UN)
|
|
def __enter__(self) -> "StateLock":
|
ensure_directory(self.path.parent, create=True)
|
lexical_lstat_chain(self.path, allow_missing_leaf=True)
|
if self.path.exists():
|
info = os.lstat(self.path)
|
if not self.path.is_file() or is_reparse(info):
|
raise CollectorError("E_STATE_LOCK", "State lock path is not a regular file.", safety=True)
|
try:
|
lock_bytes = self.path.read_bytes()
|
except PermissionError as exc:
|
raise CollectorError("E_STATE_LOCKED", "Another collector invocation is active.", safety=True) from exc
|
if lock_bytes not in (b"", b"\0"):
|
raise CollectorError("E_STATE_LOCK", "Legacy or damaged state lock requires review.", safety=True)
|
try:
|
self.fd = os.open(self.path, os.O_CREAT | os.O_RDWR, 0o600)
|
except PermissionError as exc:
|
raise CollectorError("E_STATE_LOCKED", "Another collector invocation is active.", safety=True) from exc
|
try:
|
if os.fstat(self.fd).st_size == 0:
|
os.write(self.fd, b"\0")
|
os.fsync(self.fd)
|
self._try_kernel_lock()
|
if self.owner_path.exists():
|
lexical_lstat_chain(self.owner_path, allow_missing_leaf=False)
|
try:
|
prior = json.loads(self.owner_path.read_text(encoding="utf-8"))
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
raise CollectorError("E_STATE_LOCK", "State lock owner metadata is damaged.", safety=True) from exc
|
if not isinstance(prior, dict) or set(prior) != {"pid", "process_created_at", "run_id", "acquired_at"}:
|
raise CollectorError("E_STATE_LOCK", "State lock owner metadata has an invalid schema.", safety=True)
|
prior_pid = prior.get("pid")
|
prior_created = prior.get("process_created_at")
|
if not isinstance(prior_pid, int) or not isinstance(prior_created, str):
|
raise CollectorError("E_STATE_LOCK", "State lock owner identity is invalid.", safety=True)
|
try:
|
actual_created = process_created_at(prior_pid)
|
except ProcessLookupError:
|
actual_created = None
|
except OSError as exc:
|
raise CollectorError("E_STATE_LOCK", "State lock owner identity is unprovable.", safety=True) from exc
|
if actual_created is not None:
|
code = "E_STATE_LOCKED" if actual_created == prior_created else "E_STATE_LOCK_PID_REUSE"
|
raise CollectorError(code, "State lock owner metadata refers to a live process.", safety=True)
|
quarantine = self.path.parent / f".{self.owner_path.name}.{hashlib.sha256(self.owner_path.read_bytes()).hexdigest()}.stale"
|
if quarantine.exists():
|
raise CollectorError("E_STATE_LOCK", "State lock owner quarantine already exists.", safety=True)
|
os.replace(self.owner_path, quarantine)
|
else:
|
quarantine = None
|
payload = canonical_json_bytes(
|
{
|
"pid": os.getpid(),
|
"process_created_at": process_created_at(os.getpid()),
|
"run_id": None,
|
"acquired_at": canonical_datetime(utc_now()),
|
}
|
)
|
atomic_replace_bytes(self.owner_path, payload)
|
if quarantine is not None:
|
quarantine.unlink()
|
except BaseException:
|
try:
|
self._unlock()
|
except BaseException:
|
pass
|
os.close(self.fd)
|
self.fd = None
|
raise
|
return self
|
|
def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
|
if self.fd is not None:
|
try:
|
if self.owner_path.exists():
|
self.owner_path.unlink()
|
self._unlock()
|
finally:
|
os.close(self.fd)
|
self.fd = None
|
|
|
def process_created_at(pid: int) -> str:
|
"""Return a stable process creation identity without exposing command lines."""
|
if os.name == "nt":
|
import ctypes
|
from ctypes import wintypes
|
|
PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
|
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
kernel32.OpenProcess.argtypes = (wintypes.DWORD, wintypes.BOOL, wintypes.DWORD)
|
kernel32.OpenProcess.restype = wintypes.HANDLE
|
kernel32.GetProcessTimes.argtypes = (
|
wintypes.HANDLE,
|
ctypes.POINTER(wintypes.FILETIME),
|
ctypes.POINTER(wintypes.FILETIME),
|
ctypes.POINTER(wintypes.FILETIME),
|
ctypes.POINTER(wintypes.FILETIME),
|
)
|
kernel32.GetProcessTimes.restype = wintypes.BOOL
|
kernel32.CloseHandle.argtypes = (wintypes.HANDLE,)
|
kernel32.CloseHandle.restype = wintypes.BOOL
|
handle = kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
|
if not handle:
|
error = ctypes.get_last_error()
|
if error in (87, 1168):
|
raise ProcessLookupError(pid)
|
raise OSError(error, "Could not query process identity")
|
try:
|
created = wintypes.FILETIME()
|
exited = wintypes.FILETIME()
|
kernel = wintypes.FILETIME()
|
user = wintypes.FILETIME()
|
if not kernel32.GetProcessTimes(handle, ctypes.byref(created), ctypes.byref(exited), ctypes.byref(kernel), ctypes.byref(user)):
|
raise OSError(ctypes.get_last_error(), "Could not read process creation time")
|
ticks = (created.dwHighDateTime << 32) | created.dwLowDateTime
|
return str(ticks)
|
finally:
|
kernel32.CloseHandle(handle)
|
stat = Path(f"/proc/{pid}/stat") # pragma: no cover
|
if not stat.exists():
|
raise ProcessLookupError(pid)
|
return stat.read_text(encoding="ascii").split()[21]
|
|
|
def utc_now() -> datetime:
|
return datetime.now(timezone.utc)
|
|
|
def parse_datetime(value: Any, field: str) -> datetime:
|
if not isinstance(value, str) or not value.strip():
|
raise CollectorError("E_INPUT_SCHEMA", f"{field} must be a non-empty ISO-8601 string.")
|
text = value.strip()
|
if text.endswith("Z"):
|
text = text[:-1] + "+00:00"
|
try:
|
parsed = datetime.fromisoformat(text)
|
except ValueError as exc:
|
raise CollectorError("E_INPUT_SCHEMA", f"{field} is not valid ISO-8601: {value!r}.") from exc
|
if parsed.tzinfo is None or parsed.utcoffset() is None:
|
raise CollectorError("E_INPUT_SCHEMA", f"{field} must include a UTC offset or Z.")
|
return parsed
|
|
|
def canonical_datetime(value: datetime) -> str:
|
return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
|
def reject_secret_keys(value: Any, path: str = "$") -> None:
|
if isinstance(value, Mapping):
|
for key, child in value.items():
|
key_text = str(key)
|
if SECRET_KEY_PATTERN.search(key_text):
|
raise CollectorError(
|
"E_SECRET_FIELD",
|
"Authentication or session fields are forbidden.",
|
details={"field": f"{path}.{key_text}"},
|
safety=True,
|
)
|
reject_secret_keys(child, f"{path}.{key_text}")
|
elif isinstance(value, list):
|
for index, child in enumerate(value):
|
reject_secret_keys(child, f"{path}[{index}]")
|
|
|
def load_json(path: Path, description: str) -> Any:
|
lexical_lstat_chain(path, allow_missing_leaf=False)
|
if not path.is_file():
|
raise CollectorError("E_INPUT_PATH", f"{description} is not a regular file: {path}")
|
try:
|
raw = path.read_text(encoding="utf-8")
|
except UnicodeDecodeError as exc:
|
raise CollectorError("E_INPUT_ENCODING", f"{description} must be strict UTF-8: {path}") from exc
|
try:
|
value = json.loads(raw)
|
except json.JSONDecodeError as exc:
|
raise CollectorError(
|
"E_INPUT_JSON",
|
f"{description} is not valid JSON: {path}",
|
details={"line": exc.lineno, "column": exc.colno},
|
) from exc
|
reject_secret_keys(value)
|
return value
|
|
|
def absolute_lexical(path: Path) -> Path:
|
return Path(os.path.abspath(os.fspath(path)))
|
|
|
def is_reparse(stat_result: os.stat_result) -> bool:
|
return bool(getattr(stat_result, "st_file_attributes", 0) & FILE_ATTRIBUTE_REPARSE_POINT)
|
|
|
def lexical_lstat_chain(path: Path, *, allow_missing_leaf: bool) -> None:
|
absolute = absolute_lexical(path)
|
parts = absolute.parts
|
if not parts:
|
raise CollectorError("E_PATH", "Empty path is not allowed.", safety=True)
|
current = Path(parts[0])
|
for index, part in enumerate(parts[1:], start=1):
|
current = current / part
|
try:
|
info = os.lstat(current)
|
except FileNotFoundError:
|
if allow_missing_leaf:
|
return
|
raise CollectorError(
|
"E_PATH_MISSING",
|
f"Required path does not exist: {current}",
|
safety=True,
|
)
|
if os.path.islink(current) or is_reparse(info):
|
raise CollectorError(
|
"E_PATH_REPARSE",
|
f"Symlink, junction, or reparse path is not allowed: {current}",
|
safety=True,
|
)
|
if index < len(parts) - 1 and not current.is_dir():
|
raise CollectorError(
|
"E_PATH_PARENT",
|
f"Path parent is not a directory: {current}",
|
safety=True,
|
)
|
|
|
def ensure_directory(path: Path, *, create: bool) -> Path:
|
absolute = absolute_lexical(path)
|
lexical_lstat_chain(absolute, allow_missing_leaf=create)
|
if not absolute.exists():
|
if not create:
|
raise CollectorError("E_PATH_MISSING", f"Directory does not exist: {absolute}", safety=True)
|
missing: list[Path] = []
|
cursor = absolute
|
while not cursor.exists():
|
missing.append(cursor)
|
cursor = cursor.parent
|
lexical_lstat_chain(cursor, allow_missing_leaf=False)
|
for candidate in reversed(missing):
|
candidate.mkdir()
|
lexical_lstat_chain(candidate, allow_missing_leaf=False)
|
if not absolute.is_dir():
|
raise CollectorError("E_PATH_TYPE", f"Expected directory: {absolute}", safety=True)
|
return absolute
|
|
|
def path_within(child: Path, parent: Path) -> bool:
|
try:
|
return os.path.commonpath((str(child), str(parent))) == str(parent)
|
except ValueError:
|
return False
|
|
|
def config_path(base: Path, value: Any, field: str) -> Path:
|
if not isinstance(value, str) or not value.strip():
|
raise CollectorError("E_CONFIG", f"{field} must be a non-empty path string.")
|
candidate = Path(value)
|
if not candidate.is_absolute():
|
candidate = base / candidate
|
return absolute_lexical(candidate)
|
|
|
def validate_url(value: Any, field: str, allowed_hosts: Iterable[str]) -> str:
|
if not isinstance(value, str) or not value.strip():
|
raise CollectorError("E_INPUT_SCHEMA", f"{field} must be a non-empty HTTPS URL.")
|
parts = urlsplit(value.strip())
|
host = (parts.hostname or "").lower()
|
if parts.scheme.lower() != "https" or not host or parts.username or parts.password:
|
raise CollectorError("E_SOURCE_URL", f"{field} must be an HTTPS URL without credentials.", safety=True)
|
if host not in set(allowed_hosts):
|
raise CollectorError(
|
"E_SOURCE_HOST",
|
f"{field} host is not registered: {host}",
|
details={"allowed_hosts": sorted(allowed_hosts)},
|
safety=True,
|
)
|
return canonical_url(value)
|
|
|
def canonical_url(value: str) -> str:
|
parts = urlsplit(value.strip())
|
host = (parts.hostname or "").lower()
|
port = f":{parts.port}" if parts.port else ""
|
path = parts.path.rstrip("/") or "/"
|
query = urlencode(sorted(parse_qsl(parts.query, keep_blank_values=True)), doseq=True)
|
return urlunsplit((parts.scheme.lower(), host + port, path, query, ""))
|
|
|
def load_config(path: Path) -> CollectorConfig:
|
value = load_json(path, "config")
|
if not isinstance(value, Mapping) or value.get("schema_version") != SCHEMA_VERSION:
|
raise CollectorError("E_CONFIG", f"config.schema_version must equal {SCHEMA_VERSION}.")
|
creator = value.get("creator")
|
paths = value.get("paths")
|
native_handoff = value.get("native_handoff")
|
if not isinstance(creator, Mapping) or not isinstance(paths, Mapping):
|
raise CollectorError("E_CONFIG", "config.creator and config.paths must be objects.")
|
if not isinstance(native_handoff, Mapping):
|
raise CollectorError("E_CONFIG", "config.native_handoff must be an object.")
|
route_fields = (
|
"project_id",
|
"source_ai_id",
|
"source_thread_id",
|
"source_role_instance_id",
|
"target_ai_id",
|
"target_thread_id",
|
"target_role_instance_id",
|
"reply_thread_id",
|
)
|
route_values: dict[str, str] = {}
|
for field in route_fields:
|
raw = native_handoff.get(field)
|
if not isinstance(raw, str) or not raw.strip():
|
raise CollectorError("E_CONFIG", f"native_handoff.{field} must be a non-empty string.")
|
route_values[field] = raw.strip()
|
for field in (
|
"project_id",
|
"source_ai_id",
|
"source_role_instance_id",
|
"target_ai_id",
|
"target_role_instance_id",
|
):
|
if re.fullmatch(r"[A-Za-z0-9._-]{1,128}", route_values[field]) is None:
|
raise CollectorError("E_CONFIG", f"native_handoff.{field} contains unsafe characters.")
|
for field in ("source_thread_id", "target_thread_id", "reply_thread_id"):
|
if THREAD_ID_PATTERN.fullmatch(route_values[field]) is None:
|
raise CollectorError("E_CONFIG", f"native_handoff.{field} must be a lowercase UUID.")
|
creator_name = creator.get("name")
|
if not isinstance(creator_name, str) or not creator_name.strip():
|
raise CollectorError("E_CONFIG", "creator.name must be a non-empty string.")
|
timezone_name = value.get("timezone", "Asia/Shanghai")
|
if not isinstance(timezone_name, str):
|
raise CollectorError("E_CONFIG", "timezone must be a zoneinfo name.")
|
try:
|
ZoneInfo(timezone_name)
|
except ZoneInfoNotFoundError as exc:
|
raise CollectorError("E_CONFIG", f"Unknown timezone: {timezone_name}") from exc
|
allowed_hosts_raw = value.get(
|
"allowed_source_hosts",
|
["space.bilibili.com", "www.bilibili.com", "t.bilibili.com", "b23.tv"],
|
)
|
if not isinstance(allowed_hosts_raw, list) or not allowed_hosts_raw:
|
raise CollectorError("E_CONFIG", "allowed_source_hosts must be a non-empty list.")
|
allowed_hosts = frozenset(str(item).strip().lower() for item in allowed_hosts_raw)
|
if any(not host or "/" in host or ":" in host for host in allowed_hosts):
|
raise CollectorError("E_CONFIG", "allowed_source_hosts contains an invalid hostname.")
|
dynamic_url = validate_url(creator.get("dynamic_url"), "creator.dynamic_url", allowed_hosts)
|
creator_uid_raw = creator.get("uid")
|
if creator_uid_raw is None:
|
creator_uid = None
|
elif isinstance(creator_uid_raw, (str, int)) and re.fullmatch(r"[1-9][0-9]{0,19}", str(creator_uid_raw)):
|
creator_uid = str(creator_uid_raw)
|
else:
|
raise CollectorError("E_CONFIG", "creator.uid must be a positive decimal identifier.")
|
window_hours = value.get("window_hours", 72)
|
minimum_age = value.get("minimum_complete_age_seconds", 30)
|
title_max = value.get("title_max_length", 48)
|
if not isinstance(window_hours, int) or not 1 <= window_hours <= 24 * 31:
|
raise CollectorError("E_CONFIG", "window_hours must be an integer from 1 to 744.")
|
if not isinstance(minimum_age, int) or not 0 <= minimum_age <= 3600:
|
raise CollectorError("E_CONFIG", "minimum_complete_age_seconds must be 0..3600.")
|
if not isinstance(title_max, int) or not 8 <= title_max <= 96:
|
raise CollectorError("E_CONFIG", "title_max_length must be 8..96.")
|
extensions_raw = value.get("allowed_video_extensions", sorted(DEFAULT_VIDEO_EXTENSIONS))
|
if not isinstance(extensions_raw, list) or not extensions_raw:
|
raise CollectorError("E_CONFIG", "allowed_video_extensions must be a non-empty list.")
|
extensions = frozenset(str(item).lower() for item in extensions_raw)
|
if any(not re.fullmatch(r"\.[a-z0-9]{1,8}", item) for item in extensions):
|
raise CollectorError("E_CONFIG", "allowed_video_extensions contains an invalid suffix.")
|
base = path.parent
|
refresh_raw = value.get("refresh")
|
refresh: RefreshConfig | None = None
|
if refresh_raw is not None:
|
if not isinstance(refresh_raw, Mapping):
|
raise CollectorError("E_CONFIG", "config.refresh must be an object.")
|
if (
|
creator_name.strip() != "青枫浦上Q"
|
or creator_uid != "1420210197"
|
or dynamic_url != "https://space.bilibili.com/1420210197/dynamic"
|
):
|
raise CollectorError("E_CONFIG", "refresh supports only the registered creator identity.", safety=True)
|
|
def bounded_int(field: str, default: int, lower: int, upper: int) -> int:
|
raw = refresh_raw.get(field, default)
|
if not isinstance(raw, int) or isinstance(raw, bool) or not lower <= raw <= upper:
|
raise CollectorError("E_CONFIG", f"refresh.{field} must be {lower}..{upper}.")
|
return raw
|
|
legacy_timeout_fields = {
|
"timeout_seconds", "page_ready_timeout_seconds", "dom_read_timeout_seconds"
|
}
|
if legacy_timeout_fields.intersection(refresh_raw):
|
raise CollectorError("E_CONFIG", "Legacy refresh timeout fields are not valid for runtime-v2.")
|
|
def frozen_int(field: str, expected: int) -> int:
|
raw = refresh_raw.get(field)
|
if not isinstance(raw, int) or isinstance(raw, bool) or raw != expected:
|
raise CollectorError("E_CONFIG", f"refresh.{field} must equal {expected}.")
|
return raw
|
|
refresh = RefreshConfig(
|
archive_dir=config_path(base, refresh_raw.get("archive_dir"), "refresh.archive_dir"),
|
formal_manifest=config_path(base, refresh_raw.get("formal_manifest"), "refresh.formal_manifest"),
|
intake_dir=config_path(base, refresh_raw.get("intake_dir"), "refresh.intake_dir"),
|
overall_deadline_seconds=frozen_int("overall_deadline_seconds", 120),
|
refresh_action_timeout_seconds=frozen_int("refresh_action_timeout_seconds", 35),
|
observation_timeout_seconds=frozen_int("observation_timeout_seconds", 45),
|
page_internal_settle_timeout_seconds=frozen_int("page_internal_settle_timeout_seconds", 15),
|
max_refresh_count=bounded_int("max_refresh_count", 1, 1, 1),
|
run_history_slots=bounded_int("run_history_slots", 168, 168, 168),
|
max_items=bounded_int("max_items", 200, 1, 200),
|
max_images_per_item=bounded_int("max_images_per_item", 20, 1, 20),
|
max_image_bytes=bounded_int("max_image_bytes", 20 * 1024 * 1024, 1, 20 * 1024 * 1024),
|
max_text_bytes=bounded_int("max_text_bytes", 2 * 1024 * 1024, 1, 2 * 1024 * 1024),
|
)
|
return CollectorConfig(
|
creator_name=creator_name.strip(),
|
creator_uid=creator_uid,
|
creator_dynamic_url=dynamic_url,
|
timezone_name=timezone_name,
|
window_hours=window_hours,
|
state_dir=config_path(base, paths.get("state_dir"), "paths.state_dir"),
|
download_dir=config_path(base, paths.get("download_dir"), "paths.download_dir"),
|
video_dir=config_path(base, paths.get("video_dir"), "paths.video_dir"),
|
minimum_complete_age_seconds=minimum_age,
|
title_max_length=title_max,
|
allowed_source_hosts=allowed_hosts,
|
allowed_video_extensions=extensions,
|
native_handoff=NativeHandoffRoute(**route_values),
|
refresh=refresh,
|
)
|
|
|
def clean_identifier(value: Any, field: str) -> str | None:
|
if value is None or value == "":
|
return None
|
if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,128}", value.strip()):
|
raise CollectorError("E_INPUT_SCHEMA", f"{field} has an invalid identifier.")
|
return value.strip()
|
|
|
def normalize_item(raw: Any, config: CollectorConfig, index: int) -> dict[str, Any]:
|
if not isinstance(raw, Mapping):
|
raise CollectorError("E_INPUT_SCHEMA", f"items[{index}] must be an object.")
|
dynamic_id = clean_identifier(raw.get("dynamic_id"), f"items[{index}].dynamic_id")
|
opus_id = clean_identifier(raw.get("opus_id"), f"items[{index}].opus_id")
|
bvid = clean_identifier(raw.get("bvid"), f"items[{index}].bvid")
|
if bvid is not None and not re.fullmatch(r"BV[0-9A-Za-z]{10}", bvid, flags=re.IGNORECASE):
|
raise CollectorError("E_INPUT_SCHEMA", f"items[{index}].bvid is not a valid BV identifier.")
|
if not any((dynamic_id, opus_id, bvid)):
|
raise CollectorError(
|
"E_INPUT_SCHEMA",
|
f"items[{index}] needs at least one of dynamic_id, opus_id, or bvid.",
|
)
|
content_type = raw.get("content_type")
|
if content_type not in CONTENT_TYPE_LABELS:
|
raise CollectorError(
|
"E_INPUT_SCHEMA",
|
f"items[{index}].content_type must be one of {sorted(CONTENT_TYPE_LABELS)}.",
|
)
|
title = raw.get("title")
|
if not isinstance(title, str) or not title.strip():
|
raise CollectorError("E_INPUT_SCHEMA", f"items[{index}].title must be non-empty.")
|
published_at = parse_datetime(raw.get("published_at"), f"items[{index}].published_at")
|
source_url = validate_url(raw.get("source_url"), f"items[{index}].source_url", config.allowed_source_hosts)
|
if bvid is not None:
|
bvid = "BV" + bvid[2:]
|
item = {
|
"dynamic_id": dynamic_id,
|
"opus_id": opus_id,
|
"bvid": bvid,
|
"content_type": content_type,
|
"published_at": canonical_datetime(published_at),
|
"title": " ".join(title.split()),
|
"source_url": source_url,
|
}
|
item["dedupe_keys"] = dedupe_keys(item)
|
return item
|
|
|
def dedupe_keys(item: Mapping[str, Any]) -> list[str]:
|
keys: list[str] = []
|
if item.get("dynamic_id"):
|
keys.append(f"dynamic:{str(item['dynamic_id']).lower()}")
|
if item.get("opus_id"):
|
keys.append(f"opus:{str(item['opus_id']).lower()}")
|
if item.get("bvid"):
|
keys.append(f"bvid:{str(item['bvid']).lower()}")
|
if item.get("source_url"):
|
keys.append(f"url:{canonical_url(str(item['source_url']))}")
|
return sorted(set(keys))
|
|
|
def sanitize_windows_component(value: str, max_length: int) -> str:
|
cleaned = WINDOWS_INVALID_CHARS.sub("_", value)
|
cleaned = re.sub(r"\s+", " ", cleaned).strip(" .")
|
if not cleaned:
|
cleaned = "untitled"
|
if cleaned.upper().split(".", 1)[0] in WINDOWS_RESERVED_NAMES:
|
cleaned = "_" + cleaned
|
if len(cleaned) > max_length:
|
cleaned = cleaned[:max_length].rstrip(" .")
|
return cleaned or "untitled"
|
|
|
def suggested_base(item: Mapping[str, Any], config: CollectorConfig) -> str:
|
published = parse_datetime(item["published_at"], "published_at").astimezone(config.tz)
|
title = sanitize_windows_component(str(item["title"]), config.title_max_length)
|
return f"{published:%Y%m%d-%H%M%S}_{CONTENT_TYPE_LABELS[str(item['content_type'])]}_{title}"
|
|
|
def entity_id_for_keys(keys: Sequence[str]) -> str:
|
return hashlib.sha256("\n".join(sorted(keys)).encode("utf-8")).hexdigest()[:24]
|
|
|
def canonical_json_bytes(value: Any, *, newline: bool = True) -> bytes:
|
payload = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
return (payload + ("\n" if newline else "")).encode("utf-8")
|
|
|
def sha256_file(path: Path) -> str:
|
digest = hashlib.sha256()
|
with path.open("rb") as handle:
|
for block in iter(lambda: handle.read(1024 * 1024), b""):
|
digest.update(block)
|
return digest.hexdigest()
|
|
|
def atomic_replace_bytes(path: Path, payload: bytes) -> None:
|
ensure_directory(path.parent, create=True)
|
temp = path.parent / f".{path.name}.{uuid.uuid4().hex}.tmp"
|
try:
|
with temp.open("xb") as handle:
|
handle.write(payload)
|
handle.flush()
|
os.fsync(handle.fileno())
|
os.replace(temp, path)
|
finally:
|
try:
|
temp.unlink()
|
except FileNotFoundError:
|
pass
|
|
|
def atomic_write_new_or_reuse(path: Path, payload: bytes) -> str:
|
ensure_directory(path.parent, create=True)
|
if path.exists():
|
lexical_lstat_chain(path, allow_missing_leaf=False)
|
if path.is_file() and path.read_bytes() == payload:
|
return "REUSED"
|
raise CollectorError("E_OUTPUT_EXISTS", f"Output already exists and differs: {path}", safety=True)
|
temp = path.parent / f".{path.name}.{uuid.uuid4().hex}.tmp"
|
try:
|
with temp.open("xb") as handle:
|
handle.write(payload)
|
handle.flush()
|
os.fsync(handle.fileno())
|
try:
|
os.link(temp, path)
|
except FileExistsError as exc:
|
raise CollectorError("E_OUTPUT_EXISTS", f"Output appeared during commit: {path}", safety=True) from exc
|
except OSError as exc:
|
raise CollectorError(
|
"E_ATOMIC_CREATE",
|
f"Filesystem cannot perform safe no-overwrite commit for {path}: {exc}",
|
safety=True,
|
) from exc
|
return "GENERATED"
|
finally:
|
try:
|
temp.unlink()
|
except FileNotFoundError:
|
pass
|
|
|
def manifest_text(
|
event: Mapping[str, Any],
|
field: str,
|
path: str,
|
*,
|
allow_empty: bool = False,
|
) -> str:
|
value = event.get(field)
|
if not isinstance(value, str) or (not allow_empty and not value.strip()):
|
raise CollectorError(
|
"E_MANIFEST",
|
f"Manifest field {field} is missing or invalid.",
|
details={"field": f"{path}.{field}"},
|
safety=True,
|
)
|
if CONTROL_CHARACTER_PATTERN.search(value):
|
raise CollectorError(
|
"E_MANIFEST",
|
f"Manifest field {field} contains a control character.",
|
details={"field": f"{path}.{field}"},
|
safety=True,
|
)
|
return value
|
|
|
def manifest_sha256(event: Mapping[str, Any], path: str) -> str:
|
value = manifest_text(event, "sha256", path)
|
if LOWER_SHA256_PATTERN.fullmatch(value) is None:
|
raise CollectorError(
|
"E_MANIFEST",
|
"Manifest field sha256 must be 64 lowercase hexadecimal characters.",
|
details={"field": f"{path}.sha256"},
|
safety=True,
|
)
|
return value
|
|
|
def validate_terminal_manifest_evidence(event: Mapping[str, Any], path: str) -> None:
|
if event.get("content_type") != "video":
|
return
|
status = event.get("status")
|
if status not in {"VIDEO_MOVED", "VIDEO_MOVED_SOURCE_RETAINED", "COMPLETE"}:
|
return
|
manifest_text(event, "local_file", path)
|
manifest_sha256(event, path)
|
if status == "VIDEO_MOVED_SOURCE_RETAINED":
|
manifest_text(event, "failure_reason", path)
|
|
|
def validate_handoff_manifest_evidence(event: Mapping[str, Any], path: str) -> None:
|
entity_id = manifest_text(event, "entity_id", path)
|
if INTERNAL_ENTITY_ID_PATTERN.fullmatch(entity_id) is None:
|
raise CollectorError(
|
"E_MANIFEST",
|
"Manifest field entity_id has an invalid internal format.",
|
details={"field": f"{path}.entity_id"},
|
safety=True,
|
)
|
bvid = event.get("bvid")
|
if bvid not in (None, ""):
|
if not isinstance(bvid, str) or CONTROL_CHARACTER_PATTERN.search(bvid):
|
raise CollectorError(
|
"E_MANIFEST",
|
"Manifest field bvid has an invalid format.",
|
details={"field": f"{path}.bvid"},
|
safety=True,
|
)
|
if re.fullmatch(r"BV[0-9A-Za-z]{10}", bvid, flags=re.IGNORECASE) is None:
|
raise CollectorError(
|
"E_MANIFEST",
|
"Manifest field bvid has an invalid format.",
|
details={"field": f"{path}.bvid"},
|
safety=True,
|
)
|
published_at = manifest_text(event, "published_at", path)
|
try:
|
published = parse_datetime(published_at, f"{path}.published_at")
|
except CollectorError as exc:
|
raise CollectorError(
|
"E_MANIFEST",
|
"Manifest field published_at is not canonical offset-aware time.",
|
details={"field": f"{path}.published_at"},
|
safety=True,
|
) from exc
|
if published_at != canonical_datetime(published):
|
raise CollectorError(
|
"E_MANIFEST",
|
"Manifest field published_at is not canonical UTC time.",
|
details={"field": f"{path}.published_at"},
|
safety=True,
|
)
|
manifest_sha256(event, path)
|
for field in ("title", "source_url", "local_file"):
|
manifest_text(event, field, path)
|
|
|
def load_manifest(path: Path) -> list[dict[str, Any]]:
|
if not path.exists():
|
return []
|
lexical_lstat_chain(path, allow_missing_leaf=False)
|
if not path.is_file():
|
raise CollectorError("E_MANIFEST", f"Manifest is not a regular file: {path}", safety=True)
|
events: list[dict[str, Any]] = []
|
try:
|
lines = path.read_text(encoding="utf-8").splitlines()
|
except UnicodeDecodeError as exc:
|
raise CollectorError("E_MANIFEST", "Manifest is not strict UTF-8.", safety=True) from exc
|
for line_number, line in enumerate(lines, start=1):
|
if not line.strip():
|
raise CollectorError("E_MANIFEST", f"Blank manifest line at {line_number}.", safety=True)
|
try:
|
event = json.loads(line)
|
except json.JSONDecodeError as exc:
|
raise CollectorError("E_MANIFEST", f"Invalid JSON at manifest line {line_number}.", safety=True) from exc
|
reject_secret_keys(event, f"$manifest[{line_number}]")
|
if not isinstance(event, dict) or event.get("manifest_schema_version") != MANIFEST_SCHEMA_VERSION:
|
raise CollectorError("E_MANIFEST", f"Invalid schema at manifest line {line_number}.", safety=True)
|
validate_terminal_manifest_evidence(event, f"$manifest[{line_number}]")
|
events.append(event)
|
return events
|
|
|
def append_manifest(path: Path, events: Sequence[Mapping[str, Any]]) -> None:
|
if not events:
|
return
|
previous = path.read_bytes() if path.exists() else b""
|
if previous and not previous.endswith(b"\n"):
|
raise CollectorError("E_MANIFEST", "Manifest does not end with a newline.", safety=True)
|
addition = b"".join(canonical_json_bytes(event) for event in events)
|
atomic_replace_bytes(path, previous + addition)
|
|
|
def latest_entities(events: Sequence[Mapping[str, Any]]) -> tuple[dict[str, dict[str, Any]], dict[str, set[str]]]:
|
latest: dict[str, dict[str, Any]] = {}
|
token_map: dict[str, set[str]] = {}
|
for event in events:
|
entity_id = event.get("entity_id")
|
if not isinstance(entity_id, str):
|
raise CollectorError("E_MANIFEST", "Manifest event lacks entity_id.", safety=True)
|
latest[entity_id] = dict(event)
|
keys = event.get("dedupe_keys")
|
if not isinstance(keys, list) or not all(isinstance(key, str) for key in keys):
|
raise CollectorError("E_MANIFEST", "Manifest event has invalid dedupe_keys.", safety=True)
|
for key in keys:
|
token_map.setdefault(key, set()).add(entity_id)
|
return latest, token_map
|
|
|
def resolve_entity(keys: Sequence[str], token_map: Mapping[str, set[str]]) -> str | None:
|
candidates: set[str] = set()
|
for key in keys:
|
candidates.update(token_map.get(key, set()))
|
if len(candidates) > 1:
|
raise CollectorError(
|
"E_IDENTITY_CONFLICT",
|
"Stable identifiers map to multiple manifest entities.",
|
details={"dedupe_keys": list(keys), "entity_ids": sorted(candidates)},
|
safety=True,
|
)
|
return next(iter(candidates), None)
|
|
|
def manifest_event(
|
config: CollectorConfig,
|
item: Mapping[str, Any],
|
*,
|
entity_id: str,
|
status: str,
|
collected_at: datetime,
|
suggested_stem: str,
|
local_file: str | None = None,
|
sha256: str | None = None,
|
failure_reason: str | None = None,
|
video_processing_status: str | None = None,
|
text_path: str | None = None,
|
) -> dict[str, Any]:
|
if video_processing_status is None:
|
video_processing_status = "PENDING_DOWNLOAD" if item["content_type"] == "video" else "NOT_APPLICABLE"
|
return {
|
"manifest_schema_version": MANIFEST_SCHEMA_VERSION,
|
"event_id": uuid.uuid4().hex,
|
"entity_id": entity_id,
|
"creator": config.creator_name,
|
"dynamic_id": item.get("dynamic_id"),
|
"opus_id": item.get("opus_id"),
|
"bvid": item.get("bvid"),
|
"content_type": item["content_type"],
|
"published_at": item["published_at"],
|
"title": item["title"],
|
"source_url": item["source_url"],
|
"dedupe_keys": sorted(set(item["dedupe_keys"])),
|
"suggested_stem": suggested_stem,
|
"local_file": local_file,
|
"sha256": sha256,
|
"collected_at": canonical_datetime(collected_at),
|
"status": status,
|
"failure_reason": failure_reason,
|
"video_processing_status": video_processing_status,
|
"text_path": text_path,
|
}
|
|
|
def allocate_stem(base: str, entity_id: str, latest: Mapping[str, Mapping[str, Any]]) -> str:
|
for existing_id, event in latest.items():
|
if existing_id == entity_id and isinstance(event.get("suggested_stem"), str):
|
return str(event["suggested_stem"])
|
occupied = {
|
str(event["suggested_stem"]).casefold()
|
for existing_id, event in latest.items()
|
if existing_id != entity_id and isinstance(event.get("suggested_stem"), str)
|
}
|
candidate = base
|
index = 1
|
while candidate.casefold() in occupied:
|
candidate = f"{base}_{index:02d}"
|
index += 1
|
return candidate
|
|
|
def check_items(config: CollectorConfig, input_path: Path, now: datetime) -> dict[str, Any]:
|
value = load_json(input_path, "dynamic export")
|
if not isinstance(value, Mapping) or value.get("schema_version") != SCHEMA_VERSION:
|
raise CollectorError("E_INPUT_SCHEMA", f"dynamic export schema_version must equal {SCHEMA_VERSION}.")
|
if value.get("creator") not in (None, config.creator_name):
|
raise CollectorError("E_INPUT_CREATOR", "Export creator does not match config creator.", safety=True)
|
raw_items = value.get("items")
|
if not isinstance(raw_items, list):
|
raise CollectorError("E_INPUT_SCHEMA", "dynamic export items must be a list.")
|
normalized = [normalize_item(item, config, index) for index, item in enumerate(raw_items)]
|
cutoff = now - timedelta(hours=config.window_hours)
|
events = load_manifest(config.manifest_path)
|
latest, token_map = latest_entities(events)
|
new_items: list[dict[str, Any]] = []
|
skipped_duplicate = 0
|
skipped_old = 0
|
skipped_future = 0
|
seen_input_tokens: set[str] = set()
|
for item in sorted(normalized, key=lambda row: (row["published_at"], row["source_url"])):
|
published = parse_datetime(item["published_at"], "published_at").astimezone(timezone.utc)
|
if published > now:
|
skipped_future += 1
|
continue
|
if published < cutoff:
|
skipped_old += 1
|
continue
|
keys = item["dedupe_keys"]
|
if seen_input_tokens.intersection(keys):
|
skipped_duplicate += 1
|
continue
|
seen_input_tokens.update(keys)
|
entity_id = resolve_entity(keys, token_map) or entity_id_for_keys(keys)
|
previous = latest.get(entity_id)
|
if previous and previous.get("status") in ACTIVE_OR_SUCCESS_STATUSES:
|
skipped_duplicate += 1
|
continue
|
if previous and previous.get("status") not in RETRYABLE_STATUSES:
|
raise CollectorError(
|
"E_STATUS",
|
f"Unsupported latest manifest status: {previous.get('status')}",
|
details={"entity_id": entity_id},
|
safety=True,
|
)
|
stem = allocate_stem(suggested_base(item, config), entity_id, latest)
|
event = manifest_event(
|
config,
|
item,
|
entity_id=entity_id,
|
status="TODO_QUEUED",
|
collected_at=now,
|
suggested_stem=stem,
|
)
|
todo = {
|
"entity_id": entity_id,
|
"dynamic_id": item["dynamic_id"],
|
"opus_id": item["opus_id"],
|
"bvid": item["bvid"],
|
"content_type": item["content_type"],
|
"published_at": item["published_at"],
|
"title": item["title"],
|
"source_url": item["source_url"],
|
"action": TODO_ACTIONS[item["content_type"]],
|
"suggested_stem": stem,
|
"image_name_pattern": f"{stem}_{{sequence:02d}}" if item["content_type"] == "image" else None,
|
}
|
new_items.append({"event": event, "todo": todo})
|
latest[entity_id] = event
|
for key in keys:
|
token_map.setdefault(key, set()).add(entity_id)
|
queue_path: Path | None = None
|
queue_state: str | None = None
|
if new_items:
|
queue = {
|
"schema_version": SCHEMA_VERSION,
|
"creator": config.creator_name,
|
"creator_dynamic_url": config.creator_dynamic_url,
|
"generated_at": canonical_datetime(now),
|
"window_hours": config.window_hours,
|
"window_start": canonical_datetime(cutoff),
|
"window_end": canonical_datetime(now),
|
"items": [entry["todo"] for entry in new_items],
|
"browser_interaction_required": True,
|
"authentication_data_allowed": False,
|
}
|
payload = canonical_json_bytes(queue)
|
digest = hashlib.sha256(payload).hexdigest()[:12]
|
local_time = now.astimezone(config.tz)
|
queue_path = config.queues_dir / f"{local_time:%Y%m%d-%H%M%S}_todo_{digest}.json"
|
queue_state = atomic_write_new_or_reuse(queue_path, payload)
|
try:
|
append_manifest(config.manifest_path, [entry["event"] for entry in new_items])
|
except BaseException:
|
if queue_state == "GENERATED":
|
try:
|
queue_path.unlink()
|
except FileNotFoundError:
|
pass
|
raise
|
return {
|
"status": "TODO_GENERATED" if new_items else "NO_NEW_ITEMS",
|
"creator": config.creator_name,
|
"input_items": len(normalized),
|
"new_items": len(new_items),
|
"skipped_duplicate": skipped_duplicate,
|
"skipped_old": skipped_old,
|
"skipped_future": skipped_future,
|
"queue_path": str(queue_path) if queue_path else None,
|
"queue_write": queue_state,
|
"manifest_path": str(config.manifest_path),
|
"window_start": canonical_datetime(cutoff),
|
"window_end": canonical_datetime(now),
|
}
|
|
|
def load_mapping(path: Path, config: CollectorConfig) -> list[dict[str, Any]]:
|
value = load_json(path, "completed download mapping")
|
if not isinstance(value, Mapping) or value.get("schema_version") != SCHEMA_VERSION:
|
raise CollectorError("E_INPUT_SCHEMA", f"mapping schema_version must equal {SCHEMA_VERSION}.")
|
raw_items = value.get("items")
|
if not isinstance(raw_items, list) or not raw_items:
|
raise CollectorError("E_INPUT_SCHEMA", "mapping.items must be a non-empty list.")
|
normalized: list[dict[str, Any]] = []
|
for index, raw in enumerate(raw_items):
|
if not isinstance(raw, Mapping):
|
raise CollectorError("E_INPUT_SCHEMA", f"mapping.items[{index}] must be an object.")
|
source_file = raw.get("source_file")
|
if not isinstance(source_file, str) or not source_file.strip():
|
raise CollectorError("E_INPUT_SCHEMA", f"mapping.items[{index}].source_file is required.")
|
selectors = {
|
name: clean_identifier(raw.get(name), f"mapping.items[{index}].{name}")
|
for name in ("dynamic_id", "opus_id", "bvid")
|
}
|
source_url = raw.get("source_url")
|
if source_url is not None:
|
source_url = validate_url(
|
source_url,
|
f"mapping.items[{index}].source_url",
|
config.allowed_source_hosts,
|
)
|
if not any(selectors.values()) and not source_url:
|
raise CollectorError("E_INPUT_SCHEMA", f"mapping.items[{index}] needs one stable selector.")
|
normalized.append({"source_file": source_file.strip(), **selectors, "source_url": source_url})
|
return normalized
|
|
|
def mapping_keys(mapping: Mapping[str, Any]) -> list[str]:
|
item = {
|
"dynamic_id": mapping.get("dynamic_id"),
|
"opus_id": mapping.get("opus_id"),
|
"bvid": mapping.get("bvid"),
|
"source_url": canonical_url(mapping["source_url"]) if mapping.get("source_url") else None,
|
}
|
return dedupe_keys(item)
|
|
|
def resolve_mapping_entity(mapping: Mapping[str, Any], token_map: Mapping[str, set[str]]) -> str:
|
resolved: list[str] = []
|
for key in mapping_keys(mapping):
|
candidates = token_map.get(key, set())
|
if not candidates:
|
raise CollectorError(
|
"E_MAPPING_SELECTOR_UNKNOWN",
|
"A supplied mapping selector does not identify a manifest entity.",
|
details={"selector": key},
|
safety=True,
|
)
|
if len(candidates) != 1:
|
raise CollectorError(
|
"E_MAPPING_SELECTOR_CONFLICT",
|
"A supplied mapping selector identifies multiple manifest entities.",
|
details={"selector": key, "entity_ids": sorted(candidates)},
|
safety=True,
|
)
|
resolved.append(next(iter(candidates)))
|
if not resolved:
|
raise CollectorError("E_MAPPING_SELECTOR_UNKNOWN", "Mapping has no usable selector.", safety=True)
|
if len(set(resolved)) != 1:
|
raise CollectorError(
|
"E_MAPPING_SELECTOR_CONFLICT",
|
"Supplied mapping selectors identify different manifest entities.",
|
details={"entity_ids": sorted(set(resolved))},
|
safety=True,
|
)
|
return resolved[0]
|
|
|
def resolve_download_source(value: str, download_dir: Path) -> Path:
|
source = Path(value)
|
if not source.is_absolute():
|
source = download_dir / source
|
source = absolute_lexical(source)
|
lexical_lstat_chain(source, allow_missing_leaf=False)
|
download_resolved = download_dir.resolve(strict=True)
|
source_resolved = source.resolve(strict=True)
|
if not path_within(source_resolved, download_resolved):
|
raise CollectorError(
|
"E_SOURCE_OUTSIDE_DOWNLOAD_DIR",
|
f"Download source is outside configured download_dir: {source}",
|
safety=True,
|
)
|
if not source.is_file():
|
raise CollectorError("E_SOURCE_TYPE", f"Download source is not a regular file: {source}", safety=True)
|
if source.suffix.lower() in TEMP_DOWNLOAD_SUFFIXES:
|
raise CollectorError("E_DOWNLOAD_INCOMPLETE", f"Temporary download suffix is not complete: {source}", safety=True)
|
return source
|
|
|
def resolve_move_mapping(
|
mapping: Mapping[str, Any],
|
latest: Mapping[str, Mapping[str, Any]],
|
token_map: Mapping[str, set[str]],
|
) -> dict[str, Any]:
|
entity_id = resolve_mapping_entity(mapping, token_map)
|
if entity_id not in latest:
|
raise CollectorError("E_MAPPING_NOT_FOUND", "Mapping does not identify a queued manifest item.", safety=True)
|
event = dict(latest[entity_id])
|
if event.get("content_type") != "video":
|
raise CollectorError("E_MAPPING_TYPE", "Completed download mapping must identify a video.", safety=True)
|
return {"entity_id": entity_id, "event": event, "mapping": mapping}
|
|
|
def classify_move_mapping(resolved: Mapping[str, Any]) -> dict[str, Any]:
|
entity_id = str(resolved["entity_id"])
|
event = dict(resolved["event"])
|
mapping = resolved["mapping"]
|
status = event.get("status")
|
if status in {"VIDEO_MOVED", "COMPLETE"}:
|
validate_terminal_manifest_evidence(event, "$latest")
|
return {"entity_id": entity_id, "status": "ALREADY_MOVED", "event": event, "mapping": mapping}
|
if status == "VIDEO_MOVED_SOURCE_RETAINED":
|
validate_terminal_manifest_evidence(event, "$latest")
|
return {
|
"entity_id": entity_id,
|
"status": "VIDEO_MOVED_SOURCE_RETAINED",
|
"event": event,
|
"mapping": mapping,
|
}
|
if status not in {"TODO_QUEUED", "MOVE_FAILED"}:
|
raise CollectorError("E_STATUS", f"Video cannot move from status {status!r}.", safety=True)
|
return {"entity_id": entity_id, "status": "ACTIVE_MOVE", "event": event, "mapping": mapping}
|
|
|
def preflight_active_move(
|
config: CollectorConfig,
|
classified: Mapping[str, Any],
|
now: datetime,
|
) -> dict[str, Any]:
|
mapping = classified["mapping"]
|
entity_id = str(classified["entity_id"])
|
event = dict(classified["event"])
|
source = resolve_download_source(str(mapping["source_file"]), config.download_dir)
|
suffix = source.suffix.lower()
|
if suffix not in config.allowed_video_extensions:
|
raise CollectorError(
|
"E_VIDEO_EXTENSION",
|
f"Video extension is not allowed: {suffix}",
|
details={"allowed": sorted(config.allowed_video_extensions)},
|
safety=True,
|
)
|
stat = source.stat()
|
age = now.timestamp() - stat.st_mtime
|
if age < config.minimum_complete_age_seconds:
|
raise CollectorError(
|
"E_DOWNLOAD_TOO_NEW",
|
"Download file is too new to be considered complete.",
|
details={"age_seconds": max(age, 0), "required_seconds": config.minimum_complete_age_seconds},
|
safety=True,
|
)
|
target = config.video_dir / f"{event['suggested_stem']}{suffix}"
|
lexical_lstat_chain(target, allow_missing_leaf=True)
|
if target.exists():
|
raise CollectorError("E_TARGET_EXISTS", f"Target exists; refusing to overwrite: {target}", safety=True)
|
source_identity: tuple[Any, ...]
|
if stat.st_ino:
|
source_identity = ("inode", stat.st_dev, stat.st_ino)
|
else:
|
source_identity = ("path", os.path.normcase(str(source.resolve(strict=True))))
|
return {
|
"entity_id": entity_id,
|
"status": "READY",
|
"event": event,
|
"source": source,
|
"target": target,
|
"source_stat": (stat.st_size, stat.st_mtime_ns),
|
"source_identity": source_identity,
|
}
|
|
|
def copy_commit_no_overwrite(source: Path, target: Path, expected_stat: tuple[int, int]) -> tuple[str, int]:
|
ensure_directory(target.parent, create=False)
|
temp = target.parent / f".{target.name}.{uuid.uuid4().hex}.partial"
|
digest = hashlib.sha256()
|
total = 0
|
try:
|
with source.open("rb") as source_handle, temp.open("xb") as target_handle:
|
for block in iter(lambda: source_handle.read(1024 * 1024), b""):
|
target_handle.write(block)
|
digest.update(block)
|
total += len(block)
|
target_handle.flush()
|
os.fsync(target_handle.fileno())
|
after = source.stat()
|
if (after.st_size, after.st_mtime_ns) != expected_stat or total != after.st_size:
|
raise CollectorError("E_SOURCE_CHANGED", f"Source changed during copy: {source}", safety=True)
|
if sha256_file(temp) != digest.hexdigest():
|
raise CollectorError("E_COPY_HASH", "Copied video failed SHA-256 verification.", safety=True)
|
try:
|
os.link(temp, target)
|
except FileExistsError as exc:
|
raise CollectorError("E_TARGET_EXISTS", f"Target appeared during commit: {target}", safety=True) from exc
|
except OSError as exc:
|
raise CollectorError(
|
"E_ATOMIC_CREATE",
|
f"Filesystem cannot perform safe no-overwrite target commit: {exc}",
|
safety=True,
|
) from exc
|
return digest.hexdigest(), total
|
finally:
|
try:
|
temp.unlink()
|
except FileNotFoundError:
|
pass
|
|
|
def append_move_failure(
|
config: CollectorConfig,
|
event: Mapping[str, Any],
|
entity_id: str,
|
now: datetime,
|
error: BaseException,
|
) -> None:
|
if event.get("status") in {"VIDEO_MOVED", "VIDEO_MOVED_SOURCE_RETAINED", "COMPLETE"}:
|
return
|
failure = manifest_event(
|
config,
|
event,
|
entity_id=entity_id,
|
status="MOVE_FAILED",
|
collected_at=now,
|
suggested_stem=str(event["suggested_stem"]),
|
failure_reason=f"{type(error).__name__}: {error}",
|
video_processing_status="PENDING_DOWNLOAD",
|
)
|
append_manifest(config.manifest_path, [failure])
|
|
|
def move_completed(config: CollectorConfig, mapping_path: Path, now: datetime) -> dict[str, Any]:
|
ensure_directory(config.download_dir, create=False)
|
ensure_directory(config.video_dir, create=False)
|
mappings = load_mapping(mapping_path, config)
|
events = load_manifest(config.manifest_path)
|
latest, token_map = latest_entities(events)
|
resolved = [resolve_move_mapping(mapping, latest, token_map) for mapping in mappings]
|
entity_ids = [str(item["entity_id"]) for item in resolved]
|
if len(entity_ids) != len(set(entity_ids)):
|
raise CollectorError("E_ENTITY_DUPLICATE", "A manifest entity appears more than once in the batch.", safety=True)
|
classified = [classify_move_mapping(item) for item in resolved]
|
|
prepared: list[dict[str, Any]] = []
|
for item in classified:
|
if item["status"] == "ACTIVE_MOVE":
|
prepared.append(preflight_active_move(config, item, now))
|
else:
|
prepared.append(item)
|
source_identities = [item["source_identity"] for item in prepared if item["status"] == "READY"]
|
if len(source_identities) != len(set(source_identities)):
|
raise CollectorError("E_SOURCE_DUPLICATE", "A physical source file appears more than once in the batch.", safety=True)
|
targets = [str(item["target"]).casefold() for item in prepared if item["status"] == "READY"]
|
if len(targets) != len(set(targets)):
|
raise CollectorError("E_TARGET_CONFLICT", "Multiple mappings resolve to the same target.", safety=True)
|
results: list[dict[str, Any]] = []
|
for item in prepared:
|
if item["status"] == "ALREADY_MOVED":
|
event = item["event"]
|
results.append(
|
{
|
"entity_id": item["entity_id"],
|
"status": "ALREADY_MOVED",
|
"target": event.get("local_file"),
|
"sha256": event.get("sha256"),
|
}
|
)
|
continue
|
if item["status"] == "VIDEO_MOVED_SOURCE_RETAINED":
|
event = item["event"]
|
results.append(
|
{
|
"entity_id": item["entity_id"],
|
"status": "VIDEO_MOVED_SOURCE_RETAINED",
|
"target": event.get("local_file"),
|
"sha256": event.get("sha256"),
|
"source_delete_error": event.get("failure_reason"),
|
}
|
)
|
continue
|
source = item["source"]
|
target = item["target"]
|
event = item["event"]
|
digest: str | None = None
|
committed = False
|
try:
|
digest, size = copy_commit_no_overwrite(source, target, item["source_stat"])
|
committed = True
|
moved_event = manifest_event(
|
config,
|
event,
|
entity_id=item["entity_id"],
|
status="VIDEO_MOVED",
|
collected_at=now,
|
suggested_stem=event["suggested_stem"],
|
local_file=str(target),
|
sha256=digest,
|
video_processing_status="READY_FOR_HANDOFF",
|
)
|
append_manifest(config.manifest_path, [moved_event])
|
except BaseException as exc:
|
if committed:
|
try:
|
if target.is_file() and digest and sha256_file(target) == digest:
|
target.unlink()
|
except BaseException as rollback_error:
|
if hasattr(exc, "add_note"):
|
exc.add_note(f"Could not remove unrecorded target: {rollback_error}")
|
try:
|
append_move_failure(config, event, item["entity_id"], now, exc)
|
except BaseException as evidence_error:
|
if hasattr(exc, "add_note"):
|
exc.add_note(f"Could not append MOVE_FAILED evidence: {evidence_error}")
|
raise
|
source_delete_failed: str | None = None
|
try:
|
current = source.stat()
|
if (current.st_size, current.st_mtime_ns) != item["source_stat"]:
|
raise CollectorError("E_SOURCE_CHANGED", "Source changed before deletion; retained source.", safety=True)
|
source.unlink()
|
except BaseException as exc:
|
source_delete_failed = f"{type(exc).__name__}: {exc}"
|
retained_event = manifest_event(
|
config,
|
event,
|
entity_id=item["entity_id"],
|
status="VIDEO_MOVED_SOURCE_RETAINED",
|
collected_at=now,
|
suggested_stem=event["suggested_stem"],
|
local_file=str(target),
|
sha256=digest,
|
failure_reason=source_delete_failed,
|
video_processing_status="READY_FOR_HANDOFF",
|
)
|
append_manifest(config.manifest_path, [retained_event])
|
results.append(
|
{
|
"entity_id": item["entity_id"],
|
"status": "VIDEO_MOVED_SOURCE_RETAINED" if source_delete_failed else "VIDEO_MOVED",
|
"source": str(source),
|
"target": str(target),
|
"bytes": size,
|
"sha256": digest,
|
"source_delete_error": source_delete_failed,
|
}
|
)
|
return {
|
"status": "COMPLETE_WITH_RETAINED_SOURCE"
|
if any(row["status"] == "VIDEO_MOVED_SOURCE_RETAINED" for row in results)
|
else "COMPLETE",
|
"manifest_path": str(config.manifest_path),
|
"items": results,
|
}
|
|
|
def handoff_markdown(config: CollectorConfig, ready: Sequence[Mapping[str, Any]], _generated_at: datetime) -> str:
|
route = config.native_handoff.as_dict()
|
evidence = [
|
{
|
"entity_id": str(event["entity_id"]),
|
"bvid": str(event.get("bvid") or ""),
|
"published_at": str(event["published_at"]),
|
"title": str(event["title"]),
|
"source_url": str(event["source_url"]),
|
"local_file": str(event["local_file"]),
|
"sha256": str(event["sha256"]),
|
}
|
for event in ready
|
]
|
fingerprint = canonical_json_bytes({"route": route, "videos": evidence}, newline=False)
|
handoff_id = "HANDOFF-BILI-DYNAMIC-VIDEO-PROCESSING-" + hashlib.sha256(fingerprint).hexdigest()[:24].upper()
|
lines = [
|
"<codex_native_handoff>",
|
f"project_id={route['project_id']}",
|
"message_type=video_processing_request",
|
f"handoff_id={handoff_id}",
|
f"source_ai_id={route['source_ai_id']}",
|
f"source_thread_id={route['source_thread_id']}",
|
f"source_role_instance_id={route['source_role_instance_id']}",
|
f"target_ai_id={route['target_ai_id']}",
|
f"target_thread_id={route['target_thread_id']}",
|
f"target_role_instance_id={route['target_role_instance_id']}",
|
f"reply_thread_id={route['reply_thread_id']}",
|
"status=PROCESSING_REQUESTED",
|
"",
|
"scope:",
|
f"- creator_json={json.dumps(config.creator_name, ensure_ascii=False)}",
|
f"- video_count={len(evidence)}",
|
"- Process only the verified local video files listed below; do not collect or download content.",
|
"",
|
"evidence:",
|
]
|
for item in evidence:
|
lines.extend(
|
[
|
f"- entity_id={item['entity_id']}",
|
f" bvid={item['bvid']}",
|
f" published_at={item['published_at']}",
|
f" title_json={json.dumps(item['title'], ensure_ascii=False)}",
|
f" source_url_json={json.dumps(item['source_url'], ensure_ascii=False)}",
|
f" local_file_json={json.dumps(item['local_file'], ensure_ascii=False)}",
|
f" sha256={item['sha256']}",
|
]
|
)
|
lines.extend(
|
[
|
"",
|
"expected_action:",
|
"- Verify each local_file and sha256, then process it through the existing local media workflow.",
|
"- Return processing status and evidence to reply_thread_id; this tool does not send the handoff.",
|
"</codex_native_handoff>",
|
"",
|
]
|
)
|
return "\n".join(lines)
|
|
|
def generate_handoff(
|
config: CollectorConfig,
|
output: Path | None,
|
now: datetime,
|
) -> dict[str, Any]:
|
events = load_manifest(config.manifest_path)
|
latest, _ = latest_entities(events)
|
ready = sorted(
|
(
|
event
|
for event in latest.values()
|
if event.get("content_type") == "video"
|
and event.get("status") in {"VIDEO_MOVED", "VIDEO_MOVED_SOURCE_RETAINED"}
|
and event.get("video_processing_status") == "READY_FOR_HANDOFF"
|
and event.get("local_file")
|
and event.get("sha256")
|
),
|
key=lambda event: (str(event["published_at"]), str(event["entity_id"])),
|
)
|
if not ready:
|
return {"status": "NO_READY_VIDEOS", "video_count": 0, "handoff_path": None}
|
video_root = ensure_directory(config.video_dir, create=False).resolve(strict=True)
|
for index, event in enumerate(ready):
|
validate_handoff_manifest_evidence(event, f"$handoff[{index}]")
|
local_path = absolute_lexical(Path(str(event["local_file"])))
|
lexical_lstat_chain(local_path, allow_missing_leaf=False)
|
local_resolved = local_path.resolve(strict=True)
|
if not path_within(local_resolved, video_root) or not local_path.is_file():
|
raise CollectorError(
|
"E_HANDOFF_FILE",
|
f"Handoff video is missing or outside configured video_dir: {local_path}",
|
safety=True,
|
)
|
if sha256_file(local_path) != event["sha256"]:
|
raise CollectorError(
|
"E_HANDOFF_HASH",
|
f"Handoff video SHA-256 no longer matches manifest: {local_path}",
|
safety=True,
|
)
|
markdown = handoff_markdown(config, ready, now)
|
payload = markdown.encode("utf-8")
|
digest = hashlib.sha256(payload).hexdigest()[:12]
|
target = absolute_lexical(output) if output else config.handoffs_dir / f"video_processor_handoff_{digest}.md"
|
state = atomic_write_new_or_reuse(target, payload)
|
return {
|
"status": "HANDOFF_DRAFT_READY",
|
"video_count": len(ready),
|
"handoff_path": str(target),
|
"sha256": hashlib.sha256(payload).hexdigest(),
|
"write": state,
|
"sent": False,
|
}
|
|
|
def parse_now(value: str | None) -> datetime:
|
return parse_datetime(value, "--now").astimezone(timezone.utc) if value else utc_now()
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
parser = argparse.ArgumentParser(
|
description="Coordinate Bilibili dynamic collection from local, credential-free exports.",
|
)
|
parser.add_argument("--config", required=True, type=Path, help="UTF-8 JSON config path")
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
check = subparsers.add_parser("check", help="Filter recent items and generate a de-duplicated todo queue")
|
check.add_argument("--input", required=True, type=Path, help="Credential-free exported item list")
|
check.add_argument("--now", help="Optional offset-aware ISO-8601 clock for deterministic runs")
|
move = subparsers.add_parser("move-completed", help="Safely move mapped, completed video downloads")
|
move.add_argument("--mapping", required=True, type=Path, help="Completed download mapping JSON")
|
move.add_argument("--now", help="Optional offset-aware ISO-8601 clock for deterministic runs")
|
handoff = subparsers.add_parser("handoff", help="Generate an unsent canonical Codex-native video handoff")
|
handoff.add_argument("--output", type=Path, help="Optional no-overwrite Markdown output path")
|
handoff.add_argument("--now", help="Optional offset-aware ISO-8601 clock for deterministic runs")
|
begin = subparsers.add_parser("refresh-begin", help="Create one durable hourly browser-refresh run")
|
begin.add_argument("--now", help="Optional offset-aware ISO-8601 clock for deterministic runs")
|
commit = subparsers.add_parser("refresh-commit", help="Validate and commit one local browser evidence file")
|
commit.add_argument("--input", required=True, type=Path, help="Final browser evidence JSON from refresh-begin")
|
commit.add_argument("--now", help="Optional offset-aware ISO-8601 clock for deterministic runs")
|
return parser
|
|
|
def run(argv: Sequence[str] | None = None) -> tuple[int, dict[str, Any]]:
|
args = build_parser().parse_args(argv)
|
config_path_value = absolute_lexical(args.config)
|
config = load_config(config_path_value)
|
now = parse_now(args.now)
|
ensure_directory(config.state_dir, create=True)
|
with StateLock(config.lock_path):
|
if args.command == "check":
|
result = check_items(config, absolute_lexical(args.input), now)
|
elif args.command == "move-completed":
|
result = move_completed(config, absolute_lexical(args.mapping), now)
|
elif args.command == "handoff":
|
output = None
|
if args.output:
|
output = args.output if args.output.is_absolute() else Path.cwd() / args.output
|
result = generate_handoff(config, output, now)
|
elif args.command in {"refresh-begin", "refresh-commit"}:
|
from bili_dynamic_refresh import refresh_begin, refresh_commit
|
|
if config.refresh is None:
|
raise CollectorError("E_CONFIG", "config.refresh is required for refresh commands.")
|
if args.command == "refresh-begin":
|
result = refresh_begin(config, config_path_value, now)
|
else:
|
result = refresh_commit(config, config_path_value, absolute_lexical(args.input), now)
|
else: # pragma: no cover - argparse owns this contract
|
raise CollectorError("E_COMMAND", f"Unknown command: {args.command}")
|
result = {"schema_version": SCHEMA_VERSION, "ok": True, **result}
|
code = int(result.pop("exit_code", 4 if result.get("status") == "COMPLETE_WITH_RETAINED_SOURCE" else 0))
|
return code, result
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
try:
|
code, payload = run(argv)
|
except CollectorError as exc:
|
code = 3 if exc.safety else 2
|
payload = {
|
"schema_version": SCHEMA_VERSION,
|
"ok": False,
|
"status": "SAFETY_STOP" if exc.safety else "INPUT_ERROR",
|
"error_code": exc.code,
|
"error_message": exc.message,
|
"details": exc.details,
|
}
|
except KeyboardInterrupt:
|
payload = {
|
"schema_version": SCHEMA_VERSION,
|
"ok": False,
|
"status": "INTERRUPTED",
|
"error_code": "E_INTERRUPTED",
|
"error_message": "Interrupted by user.",
|
"details": {},
|
}
|
code = 130
|
except Exception as exc: # fail closed without exposing a traceback or secrets
|
payload = {
|
"schema_version": SCHEMA_VERSION,
|
"ok": False,
|
"status": "FAILED",
|
"error_code": "E_INTERNAL",
|
"error_message": f"{type(exc).__name__}: {exc}",
|
"details": {},
|
}
|
code = 1
|
print(json.dumps(payload, ensure_ascii=False, sort_keys=True))
|
return code
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
