from __future__ import annotations
|
|
import argparse
|
import csv
|
from datetime import date, datetime, timezone
|
import hashlib
|
import json
|
import os
|
from pathlib import Path
|
import re
|
import subprocess
|
import sys
|
import time
|
from typing import Any, Callable, Iterable, Mapping, Sequence
|
|
|
BATCH_SCHEMA = "HIBOR_EXACT_BATCH_V001"
|
TASK_SCHEMA = "HIBOR_FAST_TASK_SPEC_V004"
|
MAX_ITEMS = 10
|
MAX_GENERATED_PATH_CHARS = 220
|
DEFAULT_ITEM_TIMEOUT_SECONDS = 660
|
ADB_PREFLIGHT_TIMEOUT_SECONDS = 15
|
AUTO_DEVICE_SERIAL = "auto"
|
PACKAGE_NAME = "cn.com.hibor"
|
CACHE_ROOT = "/sdcard/Android/data/cn.com.hibor/files/myfile/"
|
SHORT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,23}$")
|
SOURCE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,119}$")
|
|
MANIFEST_FIELDS = (
|
"batch_id", "item_index", "item_id", "source_id", "query", "title",
|
"publisher", "publication_date", "pages", "pages_assumed", "raw_destination",
|
"status", "kernel_exit_code", "kernel_status", "kernel_stop_code", "elapsed_ms",
|
"task_path", "stdout_path", "stderr_path", "kernel_manifest_path",
|
"kernel_delivery_path", "kernel_timing_path", "kernel_terminal_path",
|
"kernel_terminal_present", "error",
|
)
|
|
|
class BatchInputError(ValueError):
|
pass
|
|
|
Runner = Callable[
|
[Sequence[str], Path, Mapping[str, str], int], subprocess.CompletedProcess[bytes]
|
]
|
|
|
def _utc_now() -> str:
|
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
|
def _json_bytes(value: Any, *, compact: bool = False) -> bytes:
|
if compact:
|
text = json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=False)
|
else:
|
text = json.dumps(value, ensure_ascii=False, indent=2, sort_keys=False)
|
return (text + "\n").encode("utf-8")
|
|
|
def _write_create_new(path: Path, data: bytes) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_BINARY", 0)
|
try:
|
descriptor = os.open(path, flags, 0o600)
|
except FileExistsError as exc:
|
raise BatchInputError(f"refusing to overwrite existing output: {path}") from exc
|
try:
|
view = memoryview(data)
|
while view:
|
written = os.write(descriptor, view)
|
if written <= 0:
|
raise OSError(f"short write: {path}")
|
view = view[written:]
|
os.fsync(descriptor)
|
finally:
|
os.close(descriptor)
|
|
|
def _read_json_file(path: Path) -> Mapping[str, Any]:
|
try:
|
raw = path.read_bytes()
|
except OSError as exc:
|
raise BatchInputError(f"cannot read batch JSON: {path}") from exc
|
if raw.startswith(b"\xef\xbb\xbf") or b"\x00" in raw:
|
raise BatchInputError("batch JSON must be UTF-8 without BOM or NUL")
|
try:
|
value = json.loads(raw.decode("utf-8"))
|
except (UnicodeError, json.JSONDecodeError) as exc:
|
raise BatchInputError("batch JSON is not valid UTF-8 JSON") from exc
|
if not isinstance(value, Mapping):
|
raise BatchInputError("batch JSON root must be an object")
|
return value
|
|
|
def _required_text(value: Mapping[str, Any], field: str, context: str) -> str:
|
result = value.get(field)
|
if not isinstance(result, str) or not result.strip():
|
raise BatchInputError(f"{context}.{field} must be a non-empty string")
|
return result.strip()
|
|
|
def _resolve_under(project_root: Path, raw: str, allowed_root: Path, field: str) -> Path:
|
candidate = Path(raw)
|
if not candidate.is_absolute():
|
candidate = project_root / candidate
|
resolved = candidate.resolve(strict=False)
|
allowed = allowed_root.resolve(strict=False)
|
try:
|
resolved.relative_to(allowed)
|
except ValueError as exc:
|
raise BatchInputError(f"{field} must stay under {allowed}") from exc
|
return resolved
|
|
|
def validate_batch(value: Mapping[str, Any], project_root: Path) -> dict[str, Any]:
|
expected_root_keys = (
|
"schema_version", "batch_id", "source_thread_id", "reply_thread_id",
|
"runtime", "items",
|
)
|
if set(value.keys()) != set(expected_root_keys):
|
raise BatchInputError(f"batch root keys must be exactly {expected_root_keys}")
|
if value["schema_version"] != BATCH_SCHEMA:
|
raise BatchInputError(f"schema_version must be {BATCH_SCHEMA}")
|
batch_id = _required_text(value, "batch_id", "batch")
|
if not SHORT_ID_RE.fullmatch(batch_id):
|
raise BatchInputError("batch_id must be 1..24 safe ASCII characters")
|
source_thread_id = _required_text(value, "source_thread_id", "batch")
|
reply_thread_id = _required_text(value, "reply_thread_id", "batch")
|
|
runtime = value.get("runtime")
|
runtime_keys = (
|
"quota_ledger", "adb_executable", "pdfinfo_executable", "device_serial",
|
"source_scope",
|
)
|
if not isinstance(runtime, Mapping) or set(runtime.keys()) != set(runtime_keys):
|
raise BatchInputError(f"runtime keys must be exactly {runtime_keys}")
|
quota_ledger = _resolve_under(
|
project_root,
|
_required_text(runtime, "quota_ledger", "runtime"),
|
project_root / "ana-data" / "tmp",
|
"runtime.quota_ledger",
|
)
|
adb_executable = _required_text(runtime, "adb_executable", "runtime")
|
pdfinfo_executable = _required_text(runtime, "pdfinfo_executable", "runtime")
|
device_serial = _required_text(runtime, "device_serial", "runtime")
|
source_scope = runtime.get("source_scope")
|
if not isinstance(source_scope, Mapping):
|
raise BatchInputError("runtime.source_scope must be an object")
|
|
items = value.get("items")
|
if not isinstance(items, list) or not 1 <= len(items) <= MAX_ITEMS:
|
raise BatchInputError(f"items must contain 1..{MAX_ITEMS} reports")
|
normalized_items: list[dict[str, Any]] = []
|
item_ids: set[str] = set()
|
for index, item in enumerate(items, 1):
|
context = f"items[{index - 1}]"
|
if not isinstance(item, Mapping):
|
raise BatchInputError(f"{context} must be an object")
|
keys = set(item)
|
required = {
|
"item_id", "query", "title", "publisher", "publication_date",
|
"source_id", "raw_destination",
|
}
|
if not required.issubset(keys) or keys.difference(required | {"pages"}):
|
missing = sorted(required.difference(keys))
|
extra = sorted(keys.difference(required | {"pages"}))
|
raise BatchInputError(f"{context} missing={missing} extra={extra}")
|
item_id = _required_text(item, "item_id", context)
|
if not SHORT_ID_RE.fullmatch(item_id) or item_id in item_ids:
|
raise BatchInputError(f"{context}.item_id must be unique safe ASCII, 1..24 chars")
|
item_ids.add(item_id)
|
source_id = _required_text(item, "source_id", context)
|
if not SOURCE_ID_RE.fullmatch(source_id):
|
raise BatchInputError(f"{context}.source_id must be safe ASCII, 1..120 chars")
|
publication_date = _required_text(item, "publication_date", context)
|
try:
|
if date.fromisoformat(publication_date).isoformat() != publication_date:
|
raise ValueError
|
except ValueError as exc:
|
raise BatchInputError(f"{context}.publication_date must be YYYY-MM-DD") from exc
|
pages = item.get("pages")
|
if pages is not None and (type(pages) is not int or pages < 1):
|
raise BatchInputError(f"{context}.pages must be a positive integer when supplied")
|
raw_destination = _resolve_under(
|
project_root,
|
_required_text(item, "raw_destination", context),
|
project_root / "ana-data" / "cases",
|
f"{context}.raw_destination",
|
)
|
normalized_items.append({
|
"item_id": item_id,
|
"query": _required_text(item, "query", context),
|
"title": _required_text(item, "title", context),
|
"publisher": _required_text(item, "publisher", context),
|
"publication_date": publication_date,
|
"pages": pages,
|
"source_id": source_id,
|
"raw_destination": raw_destination,
|
})
|
return {
|
"batch_id": batch_id,
|
"source_thread_id": source_thread_id,
|
"reply_thread_id": reply_thread_id,
|
"runtime": {
|
"quota_ledger": quota_ledger,
|
"adb_executable": adb_executable,
|
"pdfinfo_executable": pdfinfo_executable,
|
"device_serial": device_serial,
|
"source_scope": dict(source_scope),
|
},
|
"items": normalized_items,
|
}
|
|
|
def _report_identity(item: Mapping[str, Any]) -> str:
|
text = "|".join((item["title"], item["publisher"], item["publication_date"]))
|
return hashlib.sha256(text.encode("utf-8")).hexdigest()
|
|
|
def build_task_spec(
|
batch: Mapping[str, Any], item: Mapping[str, Any], item_index: int,
|
item_output_root: Path,
|
) -> dict[str, Any]:
|
batch_id = batch["batch_id"]
|
task_id = f"HB-{batch_id}-{item_index:02d}"
|
pages = item["pages"] if item["pages"] is not None else 1
|
runtime = batch["runtime"]
|
observed_at = _utc_now()
|
return {
|
"schema_version": TASK_SCHEMA,
|
"contract_version": "REPORT-COLLECTION-CAPABILITY-V1",
|
"project_id": "project-info",
|
"handoff_id": f"HIBOR-BATCH-{batch_id}-{item_index:02d}",
|
"task_id": task_id,
|
"source_role_instance_id": "case_analysis.report_collector",
|
"source_thread_id": batch["source_thread_id"],
|
"target_role_instance_id": "case_analysis.report_collector",
|
"target_thread_id": batch["source_thread_id"],
|
"reply_thread_id": batch["reply_thread_id"],
|
"requester": "case_analysis.report_collector",
|
"review_owner": "case_analysis.report_collector",
|
"mode": "collect-one",
|
"query": item["query"],
|
"quantity": 1,
|
"aliases": [item["title"]],
|
"analysts": [],
|
"institutions": [item["publisher"]],
|
"report_types": [],
|
"date_range": {"start": item["publication_date"], "end": item["publication_date"]},
|
"minimum_pages": 1,
|
"exclude": [],
|
"source_scope": runtime["source_scope"],
|
"destination": str(item["raw_destination"]),
|
"priority": "NORMAL",
|
"naming_requirement": "standard",
|
"output_root": str(item_output_root),
|
"quota_ledger": str(runtime["quota_ledger"]),
|
"adb_executable": runtime["adb_executable"],
|
"pdfinfo_executable": runtime["pdfinfo_executable"],
|
"package_name": PACKAGE_NAME,
|
"cache_root": CACHE_ROOT,
|
"device_serial": runtime["device_serial"],
|
"observed_at_utc": observed_at,
|
"total_budget_ms": 600_000,
|
"close_reserve_ms": 30_000,
|
"batch_increment_budget_ms": 240_000,
|
"min_screens": 1,
|
"normal_max_screens": 1,
|
"hard_max_screens": 20,
|
"hard_max_candidates": 100,
|
"performance_slot_id": item["item_id"],
|
"performance_plan_id": batch_id,
|
"expected_reports": [{
|
"report_identity": _report_identity(item),
|
"title": item["title"],
|
"institution": item["publisher"],
|
"report_date": item["publication_date"],
|
"page_count": pages,
|
}],
|
}
|
|
|
def _predicted_run_root(task_spec: Mapping[str, Any]) -> Path:
|
canonical = json.dumps(
|
task_spec, ensure_ascii=False, separators=(",", ":"), sort_keys=False
|
).encode("utf-8")
|
token = hashlib.sha256(b"HIBOR-RUN-V001\x00" + canonical).hexdigest()[:24]
|
task = re.sub(r"[^A-Za-z0-9_.-]", "-", str(task_spec["task_id"]))[:80]
|
return Path(task_spec["output_root"]) / f"RUN-{task}-{token}"
|
|
|
def _planned_paths(batch_root: Path, task_specs: Sequence[Mapping[str, Any]]) -> list[Path]:
|
paths = [
|
batch_root / "batch_manifest.csv",
|
batch_root / "batch_delivery.md",
|
batch_root / "batch_timing.json",
|
batch_root / "batch_terminal.json",
|
]
|
for index, task in enumerate(task_specs, 1):
|
paths.extend((
|
batch_root / "t" / f"{index:02d}.json",
|
batch_root / "l" / f"{index:02d}.stdout.log",
|
batch_root / "l" / f"{index:02d}.stderr.log",
|
_predicted_run_root(task) / "report_collection_terminal.json",
|
))
|
return paths
|
|
|
def _default_runner(
|
command: Sequence[str], cwd: Path, env: Mapping[str, str], timeout_seconds: int,
|
) -> subprocess.CompletedProcess[bytes]:
|
return subprocess.run(
|
list(command), cwd=cwd, env=dict(env), stdout=subprocess.PIPE,
|
stderr=subprocess.PIPE, check=False, timeout=timeout_seconds,
|
)
|
|
|
def _parse_adb_devices(stdout: bytes) -> dict[str, str]:
|
try:
|
text = stdout.decode("utf-8")
|
except UnicodeError as exc:
|
raise BatchInputError("adb devices output is not valid UTF-8") from exc
|
lines = [line.strip() for line in text.splitlines()]
|
try:
|
header_index = lines.index("List of devices attached")
|
except ValueError as exc:
|
raise BatchInputError("adb devices output is missing its header") from exc
|
|
devices: dict[str, str] = {}
|
for line in lines[header_index + 1:]:
|
if not line or line.startswith("*"):
|
continue
|
fields = line.split()
|
if len(fields) < 2:
|
raise BatchInputError(f"cannot parse adb device row: {line}")
|
serial, state = fields[0], fields[1]
|
if serial in devices:
|
raise BatchInputError(f"adb returned duplicate device serial: {serial}")
|
devices[serial] = state
|
return devices
|
|
|
def resolve_device_serial(
|
adb_executable: str,
|
requested_serial: str,
|
*,
|
cwd: Path,
|
environment: Mapping[str, str],
|
runner: Runner | None = None,
|
timeout_seconds: int = ADB_PREFLIGHT_TIMEOUT_SECONDS,
|
) -> str:
|
if timeout_seconds < 1:
|
raise BatchInputError("ADB preflight timeout must be positive")
|
runner = runner or _default_runner
|
try:
|
completed = runner(
|
[adb_executable, "devices", "-l"], cwd, environment, timeout_seconds
|
)
|
except subprocess.TimeoutExpired as exc:
|
raise BatchInputError(
|
f"ADB device preflight timed out after {timeout_seconds}s"
|
) from exc
|
except OSError as exc:
|
raise BatchInputError(
|
f"cannot run ADB device preflight: {type(exc).__name__}"
|
) from exc
|
if int(completed.returncode) != 0:
|
stderr = _as_bytes(completed.stderr).decode("utf-8", errors="replace").strip()
|
detail = stderr[-500:] if stderr else f"exit code {completed.returncode}"
|
raise BatchInputError(f"ADB device preflight failed: {detail}")
|
|
devices = _parse_adb_devices(_as_bytes(completed.stdout))
|
if requested_serial == AUTO_DEVICE_SERIAL:
|
if len(devices) != 1:
|
raise BatchInputError(
|
"runtime.device_serial=auto requires exactly one connected ADB device; "
|
f"found {len(devices)}"
|
)
|
serial, state = next(iter(devices.items()))
|
if state != "device":
|
raise BatchInputError(
|
f"the only connected ADB device is not ready: {serial} ({state})"
|
)
|
return serial
|
|
state = devices.get(requested_serial)
|
if state is None:
|
raise BatchInputError(
|
f"configured ADB device is not connected: {requested_serial}"
|
)
|
if state != "device":
|
raise BatchInputError(
|
f"configured ADB device is not ready: {requested_serial} ({state})"
|
)
|
return requested_serial
|
|
|
def _as_bytes(value: bytes | str | None) -> bytes:
|
if value is None:
|
return b""
|
return value if isinstance(value, bytes) else value.encode("utf-8", errors="replace")
|
|
|
def _parse_terminal(stdout: bytes) -> tuple[dict[str, Any] | None, str | None]:
|
try:
|
value = json.loads(stdout.decode("utf-8").strip())
|
except (UnicodeError, json.JSONDecodeError) as exc:
|
return None, f"invalid kernel stdout JSON: {type(exc).__name__}"
|
if not isinstance(value, dict):
|
return None, "kernel stdout JSON is not an object"
|
return value, None
|
|
|
def _stage_group(stage: str) -> str:
|
if re.fullmatch(r"item_\d+_detail", stage):
|
return "detail"
|
if re.fullmatch(r"item_\d+_cache", stage):
|
return "cache"
|
if re.fullmatch(r"item_\d+_copy", stage):
|
return "copy"
|
if re.fullmatch(r"item_\d+_validation_publish", stage):
|
return "validation_publish"
|
return stage
|
|
|
def summarize_timing_files(paths: Iterable[Path]) -> dict[str, Any]:
|
totals: dict[str, int] = {}
|
files_read = 0
|
rows_read = 0
|
for path in paths:
|
try:
|
value = json.loads(path.read_text(encoding="utf-8"))
|
except (OSError, UnicodeError, json.JSONDecodeError):
|
continue
|
rows = value.get("rows") if isinstance(value, Mapping) else None
|
if not isinstance(rows, list):
|
continue
|
files_read += 1
|
for row in rows:
|
if not isinstance(row, Mapping):
|
continue
|
stage = row.get("stage")
|
elapsed_ms = row.get("elapsed_ms")
|
if not isinstance(stage, str) or type(elapsed_ms) is not int or elapsed_ms < 0:
|
continue
|
grouped = _stage_group(stage)
|
totals[grouped] = totals.get(grouped, 0) + elapsed_ms
|
rows_read += 1
|
return {"files_read": files_read, "rows_read": rows_read, "stage_totals_ms": totals}
|
|
|
def _manifest_bytes(rows: Sequence[Mapping[str, Any]]) -> bytes:
|
import io
|
|
buffer = io.StringIO(newline="")
|
writer = csv.DictWriter(buffer, fieldnames=MANIFEST_FIELDS, lineterminator="\n")
|
writer.writeheader()
|
for row in rows:
|
writer.writerow({field: row.get(field, "") for field in MANIFEST_FIELDS})
|
return buffer.getvalue().encode("utf-8")
|
|
|
def _delivery_text(batch_id: str, rows: Sequence[Mapping[str, Any]]) -> str:
|
succeeded = sum(row["status"] == "SUCCESS" for row in rows)
|
failed = len(rows) - succeeded
|
lines = [
|
f"# 慧博精确研报批次 {batch_id}", "",
|
f"- 总数：{len(rows)}",
|
f"- 成功：{succeeded}",
|
f"- 失败：{failed}", "", "| item_id | source_id | 状态 | 正式目录 | 错误 |",
|
"|---|---|---|---|---|",
|
]
|
for row in rows:
|
error = str(row.get("error") or "").replace("|", "\\|").replace("\n", " ")
|
lines.append(
|
f"| {row['item_id']} | {row['source_id']} | {row['status']} | "
|
f"{row['raw_destination']} | {error} |"
|
)
|
return "\n".join(lines) + "\n"
|
|
|
def _path_or_none(value: Any) -> Path | None:
|
return Path(value) if isinstance(value, str) and value else None
|
|
|
def execute_batch(
|
batch_file: Path,
|
*,
|
project_root: Path | None = None,
|
kernel_root: Path | None = None,
|
python_executable: str | None = None,
|
kernel_entry: Path | None = None,
|
runner: Runner | None = None,
|
device_runner: Runner | None = None,
|
item_timeout_seconds: int = DEFAULT_ITEM_TIMEOUT_SECONDS,
|
) -> dict[str, Any]:
|
if item_timeout_seconds < 1:
|
raise BatchInputError("item timeout must be positive")
|
project_root = (project_root or Path(__file__).resolve().parents[2]).resolve()
|
kernel_root = (kernel_root or project_root / "dev" / "ana-dev").resolve()
|
python_executable = python_executable or sys.executable
|
runner = runner or _default_runner
|
batch = validate_batch(_read_json_file(batch_file.resolve()), project_root)
|
batch_root = project_root / "ana-data" / "tmp" / "hibor-runs" / batch["batch_id"]
|
if batch_root.exists() or batch_root.is_symlink():
|
raise BatchInputError(f"batch output already exists; choose a new batch_id: {batch_root}")
|
|
environment = dict(os.environ)
|
environment["PYTHONUTF8"] = "1"
|
environment["PYTHONIOENCODING"] = "utf-8"
|
batch["runtime"]["device_serial"] = resolve_device_serial(
|
batch["runtime"]["adb_executable"],
|
batch["runtime"]["device_serial"],
|
cwd=project_root,
|
environment=environment,
|
runner=device_runner,
|
)
|
|
task_specs = [
|
build_task_spec(batch, item, index, batch_root / "r" / f"{index:02d}")
|
for index, item in enumerate(batch["items"], 1)
|
]
|
planned = _planned_paths(batch_root, task_specs)
|
longest_planned = max(planned, key=lambda path: len(str(path.resolve(strict=False))))
|
if len(str(longest_planned.resolve(strict=False))) > MAX_GENERATED_PATH_CHARS:
|
raise BatchInputError(
|
f"generated evidence path would exceed {MAX_GENERATED_PATH_CHARS} chars: "
|
f"{longest_planned}"
|
)
|
|
batch_root.mkdir(parents=True, exist_ok=False)
|
started_at = _utc_now()
|
started_ns = time.monotonic_ns()
|
rows: list[dict[str, Any]] = []
|
item_timings: list[dict[str, Any]] = []
|
|
for index, (item, task_spec) in enumerate(zip(batch["items"], task_specs), 1):
|
task_path = batch_root / "t" / f"{index:02d}.json"
|
stdout_path = batch_root / "l" / f"{index:02d}.stdout.log"
|
stderr_path = batch_root / "l" / f"{index:02d}.stderr.log"
|
_write_create_new(task_path, _json_bytes(task_spec, compact=True))
|
if kernel_entry is None:
|
command = [
|
python_executable, "-m", "hibor_fast_collection", "--task", str(task_path),
|
"--execute",
|
]
|
else:
|
command = [
|
python_executable, str(kernel_entry), "--task", str(task_path), "--execute",
|
]
|
item_started = time.monotonic_ns()
|
completed: subprocess.CompletedProcess[bytes] | None = None
|
run_error: str | None = None
|
try:
|
completed = runner(command, kernel_root, environment, item_timeout_seconds)
|
stdout = _as_bytes(completed.stdout)
|
stderr = _as_bytes(completed.stderr)
|
kernel_exit_code = int(completed.returncode)
|
except subprocess.TimeoutExpired as exc:
|
stdout = _as_bytes(exc.stdout)
|
stderr = _as_bytes(exc.stderr)
|
kernel_exit_code = 124
|
run_error = f"kernel timeout after {item_timeout_seconds}s"
|
except OSError as exc:
|
stdout = b""
|
stderr = str(exc).encode("utf-8", errors="replace")
|
kernel_exit_code = 127
|
run_error = f"kernel process error: {type(exc).__name__}"
|
elapsed_ms = max(0, (time.monotonic_ns() - item_started) // 1_000_000)
|
_write_create_new(stdout_path, stdout)
|
_write_create_new(stderr_path, stderr)
|
|
terminal, parse_error = _parse_terminal(stdout) if stdout.strip() else (None, "empty kernel stdout")
|
if run_error is None:
|
run_error = parse_error
|
terminal = terminal or {}
|
kernel_terminal_path = _path_or_none(terminal.get("terminal_path"))
|
kernel_terminal_present = bool(
|
kernel_terminal_path is not None
|
and kernel_terminal_path.is_file()
|
and terminal.get("terminal_present") is True
|
)
|
kernel_status = terminal.get("status") if isinstance(terminal.get("status"), str) else None
|
success = kernel_exit_code == 0 and kernel_status == "SUCCESS" and kernel_terminal_present
|
if not success and run_error is None:
|
details = terminal.get("stop_code") or terminal.get("blocker")
|
if not details and stderr:
|
details = stderr.decode("utf-8", errors="replace").strip()[-500:]
|
run_error = str(details or "kernel did not return a persisted SUCCESS terminal")
|
|
kernel_timing_path = _path_or_none(terminal.get("timing_path"))
|
timing_summary = summarize_timing_files(
|
[kernel_timing_path] if kernel_timing_path is not None else []
|
)
|
item_timings.append({
|
"item_index": index,
|
"item_id": item["item_id"],
|
"cli_elapsed_ms": elapsed_ms,
|
"kernel_timing_path": str(kernel_timing_path) if kernel_timing_path else None,
|
**timing_summary,
|
})
|
rows.append({
|
"batch_id": batch["batch_id"],
|
"item_index": index,
|
"item_id": item["item_id"],
|
"source_id": item["source_id"],
|
"query": item["query"],
|
"title": item["title"],
|
"publisher": item["publisher"],
|
"publication_date": item["publication_date"],
|
"pages": item["pages"] if item["pages"] is not None else "",
|
"pages_assumed": str(item["pages"] is None).lower(),
|
"raw_destination": str(item["raw_destination"]),
|
"status": "SUCCESS" if success else "FAILED",
|
"kernel_exit_code": kernel_exit_code,
|
"kernel_status": kernel_status or "",
|
"kernel_stop_code": terminal.get("stop_code") or "",
|
"elapsed_ms": elapsed_ms,
|
"task_path": str(task_path),
|
"stdout_path": str(stdout_path),
|
"stderr_path": str(stderr_path),
|
"kernel_manifest_path": terminal.get("manifest_path") or "",
|
"kernel_delivery_path": terminal.get("delivery_path") or "",
|
"kernel_timing_path": terminal.get("timing_path") or "",
|
"kernel_terminal_path": str(kernel_terminal_path) if kernel_terminal_path else "",
|
"kernel_terminal_present": str(kernel_terminal_present).lower(),
|
"error": "" if success else run_error,
|
})
|
print(
|
f"[{index}/{len(task_specs)}] {item['item_id']}: "
|
f"{'SUCCESS' if success else 'FAILED'} ({elapsed_ms / 1000:.3f}s)",
|
file=sys.stderr,
|
flush=True,
|
)
|
|
succeeded = sum(row["status"] == "SUCCESS" for row in rows)
|
failed = len(rows) - succeeded
|
status = "SUCCESS" if failed == 0 else "PARTIAL_SUCCESS" if succeeded else "FAILED"
|
exit_code = 0 if failed == 0 else 2 if succeeded else 3
|
manifest_path = batch_root / "batch_manifest.csv"
|
delivery_path = batch_root / "batch_delivery.md"
|
timing_path = batch_root / "batch_timing.json"
|
terminal_path = batch_root / "batch_terminal.json"
|
_write_create_new(manifest_path, _manifest_bytes(rows))
|
_write_create_new(delivery_path, _delivery_text(batch["batch_id"], rows).encode("utf-8"))
|
|
stage_totals: dict[str, int] = {}
|
for item_timing in item_timings:
|
for stage, milliseconds in item_timing["stage_totals_ms"].items():
|
stage_totals[stage] = stage_totals.get(stage, 0) + int(milliseconds)
|
ended_at = _utc_now()
|
total_elapsed_ms = max(0, (time.monotonic_ns() - started_ns) // 1_000_000)
|
timing_value = {
|
"schema_version": "HIBOR_EXACT_BATCH_TIMING_V001",
|
"batch_id": batch["batch_id"],
|
"started_at_utc": started_at,
|
"ended_at_utc": ended_at,
|
"total_elapsed_ms": total_elapsed_ms,
|
"stage_totals_ms": stage_totals,
|
"items": item_timings,
|
}
|
_write_create_new(timing_path, _json_bytes(timing_value))
|
|
generated_paths = [path for path in batch_root.rglob("*") if path.is_file()]
|
generated_paths.append(terminal_path)
|
max_generated_path_chars = max(len(str(path.resolve(strict=False))) for path in generated_paths)
|
if max_generated_path_chars > MAX_GENERATED_PATH_CHARS:
|
raise BatchInputError(
|
f"generated path exceeded {MAX_GENERATED_PATH_CHARS} chars after execution"
|
)
|
terminal_value = {
|
"schema_version": "HIBOR_EXACT_BATCH_TERMINAL_V001",
|
"batch_id": batch["batch_id"],
|
"status": status,
|
"exit_code": exit_code,
|
"started_at_utc": started_at,
|
"ended_at_utc": ended_at,
|
"total_elapsed_ms": total_elapsed_ms,
|
"requested": len(rows),
|
"succeeded": succeeded,
|
"failed": failed,
|
"manifest_path": str(manifest_path),
|
"delivery_path": str(delivery_path),
|
"timing_path": str(timing_path),
|
"terminal_path": str(terminal_path),
|
"terminal_present": True,
|
"max_generated_path_chars": max_generated_path_chars,
|
"path_limit_chars": MAX_GENERATED_PATH_CHARS,
|
"items": [{
|
"item_id": row["item_id"],
|
"source_id": row["source_id"],
|
"status": row["status"],
|
"kernel_exit_code": row["kernel_exit_code"],
|
"kernel_terminal_path": row["kernel_terminal_path"] or None,
|
"error": row["error"] or None,
|
} for row in rows],
|
}
|
_write_create_new(terminal_path, _json_bytes(terminal_value))
|
return terminal_value
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
parser = argparse.ArgumentParser(
|
description="Run 1..10 preselected exact Hibor reports through the existing CLI."
|
)
|
parser.add_argument("batch", type=Path, help="UTF-8 HIBOR_EXACT_BATCH_V001 JSON")
|
parser.add_argument("--project-root", type=Path, default=None)
|
parser.add_argument("--kernel-root", type=Path, default=None)
|
parser.add_argument("--kernel-python", default=sys.executable)
|
parser.add_argument("--kernel-entry", type=Path, default=None, help=argparse.SUPPRESS)
|
parser.add_argument(
|
"--item-timeout-seconds", type=int, default=DEFAULT_ITEM_TIMEOUT_SECONDS
|
)
|
return parser
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
args = build_parser().parse_args(argv)
|
try:
|
terminal = execute_batch(
|
args.batch,
|
project_root=args.project_root,
|
kernel_root=args.kernel_root,
|
python_executable=args.kernel_python,
|
kernel_entry=args.kernel_entry,
|
item_timeout_seconds=args.item_timeout_seconds,
|
)
|
except (BatchInputError, OSError) as exc:
|
print(f"HIBOR_BATCH_INPUT_ERROR:{exc}", file=sys.stderr)
|
return 12
|
sys.stdout.buffer.write(_json_bytes(terminal, compact=True))
|
return int(terminal["exit_code"])
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
