#!/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 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", } 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 CollectorConfig: creator_name: str 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 @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: """Small fail-fast lock for overlapping local/scheduled invocations.""" def __init__(self, path: Path) -> None: self.path = path self.fd: int | None = None def __enter__(self) -> "StateLock": ensure_directory(self.path.parent, create=True) try: self.fd = os.open(self.path, os.O_CREAT | os.O_EXCL | os.O_WRONLY) except FileExistsError as exc: raise CollectorError( "E_STATE_LOCKED", "Another collector invocation is active, or a stale lock needs manual review.", details={"lock_path": str(self.path)}, safety=True, ) from exc payload = json.dumps( {"pid": os.getpid(), "created_at": utc_now().isoformat()}, ensure_ascii=False, sort_keys=True, ).encode("utf-8") os.write(self.fd, payload) os.fsync(self.fd) return self def __exit__(self, exc_type: object, exc: object, tb: object) -> None: if self.fd is not None: os.close(self.fd) self.fd = None try: self.path.unlink() except FileNotFoundError: pass 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) 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 return CollectorConfig( creator_name=creator_name.strip(), 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), ) 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 = [ "", 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.", "", "", ] ) 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") 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) 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 = 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())