#!/usr/bin/env python3
|
"""Minimal native-messaging boundary for the generic article/image collector."""
|
|
from __future__ import annotations
|
|
import json
|
import struct
|
import sys
|
from pathlib import Path
|
from typing import Any, BinaryIO, Mapping
|
|
import bili_article_image_collector as collector
|
|
|
MAX_REQUEST_BYTES = 16 * 1024 * 1024
|
MAX_RESPONSE_BYTES = 1024 * 1024
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
|
|
def _exact_keys(value: Any, expected: set[str]) -> Mapping[str, Any]:
|
if not isinstance(value, Mapping) or set(value) != expected:
|
raise collector.CollectorError("E_HOST_SCHEMA", "Native Host request keys differ.", safety=True)
|
collector._reject_secrets(value)
|
return value
|
|
|
def _project_path(value: Any, *, required: bool) -> Path | None:
|
if value is None and not required:
|
return None
|
if not isinstance(value, str) or not value.strip():
|
raise collector.CollectorError("E_HOST_PATH", "Native Host path is invalid.", safety=True)
|
path = Path(value)
|
path = (path if path.is_absolute() else PROJECT_ROOT / path).resolve(strict=False)
|
if not collector._within(path, PROJECT_ROOT):
|
raise collector.CollectorError("E_HOST_PATH", "Native Host path escapes the project root.", safety=True)
|
collector._safe_existing_chain(path, allow_missing_leaf=not required)
|
if required:
|
collector._ordinary_file(path)
|
return path
|
|
|
def run_request(raw: Any) -> tuple[int, dict[str, Any]]:
|
try:
|
request = _exact_keys(raw, {"schema_version", "action", "config_path", "capture_path", "terminal_path"})
|
if request["schema_version"] != 1:
|
raise collector.CollectorError("E_HOST_SCHEMA", "Native Host schema_version differs.", safety=True)
|
action = request["action"]
|
if action not in {"validate_capture", "collect", "verify"}:
|
raise collector.CollectorError("E_HOST_ACTION", "Native Host action is unsupported.", safety=True)
|
config_path = _project_path(request["config_path"], required=True)
|
capture_required = action != "verify"
|
capture_path = _project_path(request["capture_path"], required=capture_required)
|
terminal_path = _project_path(request["terminal_path"], required=False)
|
if not capture_required and capture_path is not None:
|
raise collector.CollectorError("E_HOST_SCHEMA", "verify forbids capture_path.", safety=True)
|
argv = ["--config", str(config_path)]
|
if action == "validate_capture":
|
if terminal_path is not None:
|
raise collector.CollectorError("E_HOST_SCHEMA", "validate_capture forbids terminal_path.", safety=True)
|
argv.extend(["validate-capture", "--capture", str(capture_path)])
|
elif action == "collect":
|
argv.extend(["collect", "--capture", str(capture_path)])
|
if terminal_path is not None:
|
argv.extend(["--terminal", str(terminal_path)])
|
else:
|
argv.append("verify")
|
if terminal_path is not None:
|
argv.extend(["--terminal", str(terminal_path)])
|
return collector.run(argv)
|
except collector.CollectorError as exc:
|
return (3 if exc.safety else 2), {
|
"schema_version": 1,
|
"status": "SAFETY_STOP" if exc.safety else "INPUT_ERROR",
|
"error_code": exc.code,
|
"message": exc.message,
|
"mutation_count": 0,
|
}
|
except Exception:
|
return 1, {"schema_version": 1, "status": "INTERNAL_ERROR", "error_code": "E_INTERNAL", "mutation_count": 0}
|
|
|
def _read_frame(stream: BinaryIO) -> Any:
|
header = stream.read(4)
|
if len(header) != 4:
|
raise EOFError
|
size = struct.unpack("<I", header)[0]
|
if size < 2 or size > MAX_REQUEST_BYTES:
|
raise collector.CollectorError("E_HOST_FRAME", "Native Host request frame size is invalid.", safety=True)
|
payload = stream.read(size)
|
if len(payload) != size:
|
raise collector.CollectorError("E_HOST_FRAME", "Native Host request frame is truncated.", safety=True)
|
return json.loads(payload.decode("utf-8", errors="strict"))
|
|
|
def _write_frame(stream: BinaryIO, value: Mapping[str, Any]) -> None:
|
payload = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
if len(payload) > MAX_RESPONSE_BYTES:
|
payload = b'{"error_code":"E_HOST_RESPONSE_LIMIT","mutation_count":0,"schema_version":1,"status":"INTERNAL_ERROR"}'
|
stream.write(struct.pack("<I", len(payload)))
|
stream.write(payload)
|
stream.flush()
|
|
|
def main() -> int:
|
try:
|
request = _read_frame(sys.stdin.buffer)
|
code, result = run_request(request)
|
except EOFError:
|
return 0
|
except Exception:
|
code, result = 3, {"schema_version": 1, "status": "SAFETY_STOP", "error_code": "E_HOST_FRAME", "mutation_count": 0}
|
_write_frame(sys.stdout.buffer, result)
|
return code
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
