chore(project-info): archive Bilibili dynamic 344d583568e4

MB-X Bilibili Pipeline

6 days ago af8d9f9dc4d7df459c112d72c14daa3879914411

 dev/project-dev/hibor_batch_executor.py

@@ -19,6 +19,8 @@
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}$")
@@ -313,6 +315,86 @@
    )


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""
@@ -409,6 +491,7 @@
    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:
@@ -421,6 +504,17 @@
    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}")
@@ -439,9 +533,6 @@
    started_ns = time.monotonic_ns()
    rows: list[dict[str, Any]] = []
    item_timings: list[dict[str, Any]] = []
    environment = dict(os.environ)
    environment["PYTHONUTF8"] = "1"
    environment["PYTHONIOENCODING"] = "utf-8"

    for index, (item, task_spec) in enumerate(zip(batch["items"], task_specs), 1):
        task_path = batch_root / "t" / f"{index:02d}.json"