from __future__ import annotations
|
|
import argparse
|
import json
|
from pathlib import Path
|
import sys
|
|
from .models import ContractError, TaskSpec
|
from .adb import AdbClient
|
from .process import ProcessSupervisor
|
from .terminal import canonical_terminal_bytes, legacy_v003_projection
|
from .ui import HiborUiDriver, UiConfig
|
from .workflow import HiborWorkflow
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
parser = argparse.ArgumentParser(prog="hibor-fast-collection")
|
parser.add_argument("--task", type=Path, required=True, help="UTF-8 TaskSpec JSON")
|
parser.add_argument("--terminal-schema", choices=("V009", "V003"), default="V009")
|
parser.add_argument("--dry-run", action="store_true", help="force no-external-action validation")
|
parser.add_argument("--discover", action="store_true",
|
help="read-only APP candidate discovery; never reserve or trigger")
|
parser.add_argument("--execute", action="store_true",
|
help="perform the explicitly authorized APP collection run")
|
return parser
|
|
|
def main(argv: list[str] | None = None) -> int:
|
args = build_parser().parse_args(argv)
|
try:
|
raw = args.task.read_bytes()
|
if raw.startswith(b"\xef\xbb\xbf") or b"\x00" in raw:
|
raise ValueError("BOM/NUL forbidden")
|
value = json.loads(raw.decode("utf-8"))
|
spec = TaskSpec.from_mapping(value)
|
if args.dry_run or spec.mode == "dry-run":
|
if args.execute or args.discover:
|
raise ValueError("--dry-run is mutually exclusive with execution/discovery")
|
terminal = HiborWorkflow(spec).dry_run()
|
elif args.discover or spec.mode == "discover":
|
if args.execute:
|
raise ValueError("--discover is a distinct read-only APP mode")
|
supervisor = ProcessSupervisor()
|
adb = AdbClient(Path(spec.adb_executable), supervisor, serial=spec.device_serial,
|
package_name=spec.package_name, cache_root=spec.cache_root)
|
ui = HiborUiDriver(adb, UiConfig.from_source_scope(spec.source_scope))
|
discovery = HiborWorkflow(spec).discover(adb=adb, ui=ui, execute=True)
|
data = json.dumps(discovery, ensure_ascii=False, separators=(",", ":"),
|
sort_keys=False).encode("utf-8")
|
sys.stdout.buffer.write(data + b"\n")
|
return 0
|
else:
|
if not args.execute:
|
raise ValueError("real APP collection requires explicit --execute")
|
supervisor = ProcessSupervisor()
|
adb = AdbClient(Path(spec.adb_executable), supervisor, serial=spec.device_serial,
|
package_name=spec.package_name, cache_root=spec.cache_root)
|
ui = HiborUiDriver(adb, UiConfig.from_source_scope(spec.source_scope))
|
terminal = HiborWorkflow(spec).collect(adb=adb, ui=ui, execute=True)
|
if args.terminal_schema == "V003":
|
data = json.dumps(legacy_v003_projection(terminal), ensure_ascii=False,
|
separators=(",", ":"), sort_keys=False).encode("utf-8")
|
else:
|
data = canonical_terminal_bytes(terminal)
|
sys.stdout.buffer.write(data + b"\n")
|
return int(terminal["exit_code"])
|
except (OSError, ValueError, ContractError, json.JSONDecodeError) as exc:
|
sys.stderr.write(f"HIBOR_FAST_COLLECTION_INPUT_ERROR:{type(exc).__name__}\n")
|
return 12
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
