From 873dca205129f123d5ea9dd768bfa1c2e5452830 Mon Sep 17 00:00:00 2001
From: MB-X Bilibili Pipeline <mbx-bili-pipeline@localhost>
Date: Sat, 05 Sep 2026 22:27:58 +0800
Subject: [PATCH] chore(project-info): archive Bilibili dynamic 66e47c0ffa2d
---
dev/ana-dev/hibor_fast_collection/adb.py | 55 +++++++++++++++++++++++++++++++++++++++++++++++--------
1 files changed, 47 insertions(+), 8 deletions(-)
diff --git a/dev/ana-dev/hibor_fast_collection/adb.py b/dev/ana-dev/hibor_fast_collection/adb.py
index 8b0cfa8..d451ba8 100644
--- a/dev/ana-dev/hibor_fast_collection/adb.py
+++ b/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)
--
Gitblit v1.9.3