chore(project-info): archive Bilibili dynamic e80fe51983e9

MB-X Bilibili Pipeline

6 days ago 2b2c880a7dcabb2505780ce0850e9dd7f2afdcfa

 dev/ana-dev/hibor_fast_collection/adb.py

@@ -40,6 +40,9 @@


BOUNDS_RE = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$")
SAFE_ANDROID_INPUT_TEXT_RE = re.compile(r"^[A-Za-z0-9._-]+$")
MEMU_IME_COMPONENT = "com.microvirt.memuime/.MemuIME"
WELCOME_ACTIVITY = "cn.com.hibor/.WelcomeActivity"


class AdbClient:
@@ -197,25 +200,61 @@
    def input_text(self, value: str) -> None:
        if any(ch in value for ch in "\n\r\x00"):
            raise ContractError(ErrorCode.TASK_SPEC_INVALID, "query", "control character")
        # The real MEmu device uses MemuIME.  Android's generic ``input text``
        # drops non-ASCII search terms, so replace the focused field through the
        # emulator's UTF-8/Base64 clipboard service and paste key.  The repeated
        # keycodes are a single argv invocation (not a remote shell script).
        use_android_input = SAFE_ANDROID_INPUT_TEXT_RE.fullmatch(value) is not None
        if not use_android_input:
            ime_list = self.run("shell", "ime", "list", "-s")
            installed_imes = {
                line.strip()
                for line in ime_list.stdout_bytes.decode("utf-8", "replace").splitlines()
                if line.strip()
            }
            if ime_list.exit_code != 0 or MEMU_IME_COMPONENT not in installed_imes:
                raise ContractError(
                    ErrorCode.UI_ANCHOR_DRIFT, "input_text",
                    "safe input unavailable: MemuIME unsupported",
                )

        end = self.run("shell", "input", "keyevent", "123")
        clear = self.run("shell", "input", "keyevent", *("67" for _ in range(256)))
        if end.exit_code != 0 or clear.exit_code != 0:
            raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "failed")

        if use_android_input:
            typed = self.run("shell", "input", "text", value)
            if typed.exit_code != 0:
                raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "generic input failed")
            return

        encoded = base64.b64encode(value.encode("utf-8")).decode("ascii")
        clipboard = self.run(
            "shell", "am", "startservice", "-a", "memu_clip_board_base64",
            "--es", "clip", encoded,
        )
        if clipboard.exit_code != 0:
            # Never issue paste after a failed clipboard service call: on some
            # emulators that would paste unrelated host clipboard contents.
            raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "Memu clipboard failed")
        paste = self.run("shell", "input", "keyevent", "279")
        if any(result.exit_code != 0 for result in (end, clear, clipboard, paste)):
            raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "failed")
        if paste.exit_code != 0:
            raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "input_text", "paste failed")

    def start_package(self) -> None:
        result = self.run("shell", "monkey", "-p", self.package_name, "-c", "android.intent.category.LAUNCHER", "1")
        if result.exit_code != 0:
            raise ContractError(ErrorCode.PACKAGE_MISSING, "start_package", "failed")
        if result.exit_code == 0:
            return
        diagnostic = (result.stdout_bytes + b"\n" + result.stderr_bytes).decode("utf-8", "replace").casefold()
        monkey_missing = result.exit_code == 127 or (
            "monkey" in diagnostic and any(token in diagnostic for token in ("not found", "inaccessible"))
        )
        if not monkey_missing:
            raise ContractError(ErrorCode.PACKAGE_MISSING, "start_package", "monkey failed")
        fallback = self.run("shell", "am", "start", "-W", "-n", WELCOME_ACTIVITY)
        fallback_text = (fallback.stdout_bytes + b"\n" + fallback.stderr_bytes).decode(
            "utf-8", "replace"
        ).casefold()
        if (fallback.exit_code != 0 or "error:" in fallback_text
                or "error type" in fallback_text or "exception" in fallback_text):
            raise ContractError(ErrorCode.PACKAGE_MISSING, "start_package", "WelcomeActivity failed")

    def _require_cache_path(self, path: str) -> None:
        root = PurePosixPath(self.cache_root)