diff --git a/client/scripts/run_t103_sku_selection.py b/client/scripts/run_t103_sku_selection.py
index 7438acf..695813d 100644
--- a/client/scripts/run_t103_sku_selection.py
+++ b/client/scripts/run_t103_sku_selection.py
@@ -15,7 +15,7 @@ from cmbuyer_client.device.adb import AdbClient, DeviceConnectionError, Subproce
from cmbuyer_client.device.baseline import NoReconnectUiautomatorConnector
from cmbuyer_client.pdd.product_url import ProductUrlError, parse_product_url
from cmbuyer_client.pdd.sku_selection import EXPECTED_GOODS_ID, SkuSelectionError, TASK_TO_UI_SELECTION
-from cmbuyer_client.pdd.sku_selection_runner import SkuSelectionRunError, SkuSelectionRunner
+from cmbuyer_client.pdd.sku_selection_runner import SkuSelectionRunError, SkuSelectionRunner, safe_failure_stage
def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
@@ -66,7 +66,10 @@ def main(argv: list[str] | None = None) -> int:
result = runner.run(arguments.serial, arguments.url, arguments.color, arguments.size, arguments.output_dir)
except (DeviceConnectionError, SkuSelectionRunError, SkuSelectionError) as error:
# Flow 可能来自测试替身或未来实现；CLI 不回显任何异常正文，避免泄露节点树或页面文本。
- print("规格恢复失败：已停止，未发布本地证据目录。", file=sys.stderr)
+ print(
+ f"规格恢复失败：stage={safe_failure_stage(error)}；已停止，未发布本地证据目录。",
+ file=sys.stderr,
+ )
return 1
except OSError:
print("规格恢复失败：无法创建或发布本地证据目录。", file=sys.stderr)
diff --git a/client/src/cmbuyer_client/pdd/sku_selection_runner.py b/client/src/cmbuyer_client/pdd/sku_selection_runner.py
index 5f6600e..c0f118a 100644
--- a/client/src/cmbuyer_client/pdd/sku_selection_runner.py
+++ b/client/src/cmbuyer_client/pdd/sku_selection_runner.py
@@ -23,7 +23,7 @@ from uiautomator2.exceptions import HTTPTimeoutError
from ..device.adb import AdbClient, DeviceConnectionError, DeviceInspection
from ..device.baseline import PDD_PACKAGE, SCREENSHOT_PARAMS, _save_base64_screenshot, _sha256_file
from .product_open import EXPECTED_PDD_VERSION
-from .product_url import ProductUrl, parse_product_url
+from .product_url import ProductUrl, ProductUrlError, parse_product_url
from .sku_selection import (
EXPECTED_GOODS_ID,
EXPECTED_UNIT_PRICE,
@@ -39,6 +39,24 @@ EXPECTED_DEVICE_MODEL = "PKG110"
EXPECTED_ANDROID_VERSION = "16"
EXPECTED_SCREEN_SIZE = (1080, 2376)
+# CLI 只允许输出这些固定阶段码。阶段码描述运行器自己的控制流，不包含页面
+# 文本、节点属性、serial、路径或第三方异常；未知/伪造值统一降级为 unknown。
+_FAILURE_STAGES = frozenset(
+ (
+ "precheck",
+ "device_inspection",
+ "device_session",
+ "product_open",
+ "sku_entry",
+ "sku_selection",
+ "price_verification",
+ "screenshot_capture",
+ "screenshot_reverify",
+ "safe_exit",
+ "publish",
+ )
+)
+
class SkuSelectionRunError(RuntimeError):
"""T-103 运行未完整完成；错误文本不携带设备或页面原文。"""
@@ -60,6 +78,29 @@ class SkuSelectionDeviceAdapterError(SkuSelectionRunError):
"""第三方设备接口失败的脱敏映射。"""
+def safe_failure_stage(error: BaseException) -> str:
+ """返回允许公开的固定阶段码，绝不回显异常正文。"""
+
+ try:
+ stage = getattr(error, "_cmbuyer_failure_stage", None)
+ # exact str 避免恶意 str 子类在 hash/eq 中执行任意异常；诊断路径
+ # 自己也必须失败闭合，不能让异常正文越过 CLI 的统一脱敏出口。
+ return stage if type(stage) is str and stage in _FAILURE_STAGES else "unknown"
+ except BaseException:
+ return "unknown"
+
+
+def _annotate_failure(error: BaseException, stage: str) -> None:
+ """只给本次异常附加白名单控制流事实；原异常文本仍不对外输出。"""
+
+ safe_stage = stage if stage in _FAILURE_STAGES else "unknown"
+ try:
+ setattr(error, "_cmbuyer_failure_stage", safe_stage)
+ except BaseException:
+ # 极端第三方异常不允许写属性时仍保持原失败闭合语义。
+ pass
+
+
@dataclass(frozen=True)
class SkuSelectionRunResult:
"""已发布的截图和无页面正文 manifest 摘要。"""
@@ -183,38 +224,52 @@ class SkuSelectionRunner:
task_size: str,
output_directory: Path,
) -> SkuSelectionRunResult:
- link = parse_product_url(product_url)
- if link.goods_id != EXPECTED_GOODS_ID:
- raise SkuSelectionRunError("商品不是 T-103 已取证目标，已停止操作。")
- selection = resolve_task_selection(task_color, task_size)
- target = Path(output_directory)
- _validate_new_target(target)
-
+ stage = "precheck"
+ staging: Path | None = None
adapter: UiautomatorSkuPanelAdapter | None = None
flow: SkuSelectionFlow | None = None
- staging = _prepare_staging(target)
- deadline = self._monotonic_clock() + self._timeout_seconds
try:
+ link = parse_product_url(product_url)
+ if link.goods_id != EXPECTED_GOODS_ID:
+ raise SkuSelectionRunError("商品不是 T-103 已取证目标，已停止操作。")
+ selection = resolve_task_selection(task_color, task_size)
+ target = Path(output_directory)
+ _validate_new_target(target)
+ staging = _prepare_staging(target)
+ deadline = self._monotonic_clock() + self._timeout_seconds
+
+ stage = "device_inspection"
inspection = self._adb_client.inspect(serial)
_require_expected_device(inspection)
+
+ stage = "device_session"
adapter = UiautomatorSkuPanelAdapter(self._connector(serial), self._timeout_seconds)
_require_expected_version(adapter.app_info(PDD_PACKAGE))
if adapter.display_size() != EXPECTED_SCREEN_SIZE:
raise SkuSelectionRunError("设备不是已取证的竖屏坐标空间，已停止操作。")
pre_intent_hierarchy = adapter.dump_window_hierarchy()
+
# 固定 ACTION_VIEW、固定 PDD package 和 canonical goods_id；不接受任意 URL 或 shell。
+ stage = "product_open"
self._adb_client.start_pdd_view_intent(serial, link.goods_id)
remaining = deadline - self._monotonic_clock()
if remaining <= 0:
raise SkuSelectionRunTimeoutError("等待规格入口超时，未执行点击。")
+
+ stage = "sku_entry"
flow = SkuSelectionFlow(adapter, entry_wait_timeout_seconds=remaining)
flow.open_sku_panel(link.canonical_url, pre_intent_hierarchy)
+
+ stage = "sku_selection"
flow.select_sku_options(selection)
+
+ stage = "price_verification"
unit_price = flow.verify_target_selection_and_read_price(selection)
if unit_price != EXPECTED_UNIT_PRICE:
raise SkuSelectionUnexpectedPriceError("规格面板现价不是本任务已确认值，已停止操作。")
+ stage = "screenshot_capture"
screenshot_path = staging / "screenshot.png"
try:
_save_base64_screenshot(adapter.capture_screenshot(), screenshot_path)
@@ -226,11 +281,15 @@ class SkuSelectionRunner:
manifest_path = staging / "manifest.json"
# 截图可能落在动态页面切换边界；发布前必须用一棵更新节点树同时重证两维和现价。
+ stage = "screenshot_reverify"
final_price = flow.verify_target_selection_and_read_price(selection)
if final_price != EXPECTED_UNIT_PRICE:
raise SkuSelectionUnexpectedPriceError("截图后规格面板现价不是本任务已确认值，已停止操作。")
# 正常路径仍经 Flow 做最后一次前台和面板判定；返回操作只发生一次。
+ stage = "safe_exit"
flow.exit_sku_panel_safely()
+
+ stage = "publish"
manifest_path.write_text(
json.dumps(_manifest(inspection, serial, link, screenshot_path, task_color, task_size), ensure_ascii=False, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
@@ -238,18 +297,25 @@ class SkuSelectionRunner:
# Windows 的 rename 不替换既有目标；并发创建 target 时保留其内容并把本次运行判失败。
os.rename(staging, target)
staging = None
- except (DeviceConnectionError, SkuSelectionRunError, SkuSelectionError):
+ except (DeviceConnectionError, ProductUrlError, SkuSelectionRunError, SkuSelectionError) as error:
_clean_staging(staging)
+ _annotate_failure(error, stage)
raise
except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
_clean_staging(staging)
- raise SkuSelectionRunTimeoutError("规格面板运行超时，未发布任何证据产物。") from error
+ mapped = SkuSelectionRunTimeoutError("规格面板运行超时，未发布任何证据产物。")
+ _annotate_failure(mapped, stage)
+ raise mapped from error
except OSError as error:
_clean_staging(staging)
- raise SkuSelectionRunError("规格面板证据目录无法创建或发布，未发布任何证据产物。") from error
+ mapped = SkuSelectionRunError("规格面板证据目录无法创建或发布，未发布任何证据产物。")
+ _annotate_failure(mapped, stage)
+ raise mapped from error
except Exception as error:
_clean_staging(staging)
- raise SkuSelectionRunError("规格面板运行未完成，未发布任何证据产物。") from error
+ mapped = SkuSelectionRunError("规格面板运行未完成，未发布任何证据产物。")
+ _annotate_failure(mapped, stage)
+ raise mapped from error
finally:
# 失败路径只能复用 Flow 的版本、前台和面板证明；证明不了便停止，绝不盲目返回。
if flow is not None and adapter is not None and adapter.entry_was_tapped and not adapter.left_panel:
diff --git a/client/tests/pdd/test_sku_selection.py b/client/tests/pdd/test_sku_selection.py
index a0f7f0c..6b28bbb 100644
--- a/client/tests/pdd/test_sku_selection.py
+++ b/client/tests/pdd/test_sku_selection.py
@@ -15,7 +15,7 @@ from PIL import Image
import cmbuyer_client.pdd as pdd
import cmbuyer_client.pdd.sku_selection_runner as runner_module
-from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
+from cmbuyer_client.device.adb import AdbDevice, DeviceConnectionError, DeviceInspection
from cmbuyer_client.pdd import SkuSelectionError, SkuSelectionFlow, SkuSelectionRunner
from cmbuyer_client.pdd.sku_selection import SkuPanelDevice, _action_bounds, resolve_task_selection
from cmbuyer_client.pdd.sku_selection_runner import (
@@ -23,6 +23,7 @@ from cmbuyer_client.pdd.sku_selection_runner import (
SkuSelectionRunError,
SkuSelectionScreenshotError,
UiautomatorSkuPanelAdapter,
+ safe_failure_stage,
)
@@ -383,6 +384,38 @@ class SkuSelectionRunnerTests(unittest.TestCase):
self.assertFalse((target / "hierarchy.xml").exists())
self.assertEqual(_actions(device, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)])
+ def test_failure_stage_is_fixed_control_flow_metadata_without_error_text(self) -> None:
+ class NoPanelAfterEntry(_RawDevice):
+ def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+ if method == "click":
+ self.calls.append(("jsonrpc", method, params, timeout))
+ return ""
+ return super().jsonrpc_call(method, params, timeout)
+
+ with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionError) as raised:
+ self._runner(_FakeAdb(), NoPanelAfterEntry()).run(
+ "device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
+ )
+ self.assertEqual(safe_failure_stage(raised.exception), "sku_entry")
+
+ forged = SkuSelectionError("private")
+ setattr(forged, "_cmbuyer_failure_stage", "")
+ self.assertEqual(safe_failure_stage(forged), "unknown")
+
+ class HostileSetterError(DeviceConnectionError):
+ def __setattr__(self, name: str, value: object) -> None:
+ raise KeyboardInterrupt("SERIAL=private secret")
+
+ class FailingAdb(_FakeAdb):
+ def inspect(self, serial: str) -> DeviceInspection:
+ raise HostileSetterError("private")
+
+ with TemporaryDirectory() as temporary, self.assertRaises(HostileSetterError) as hostile:
+ SkuSelectionRunner(FailingAdb(), lambda serial: _RawDevice(), 10).run(
+ "device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
+ )
+ self.assertEqual(safe_failure_stage(hostile.exception), "unknown")
+
def test_target_created_during_publish_is_preserved_without_staging_residue(self) -> None:
with TemporaryDirectory() as temporary:
target = Path(temporary) / "result"
@@ -401,8 +434,9 @@ class SkuSelectionRunnerTests(unittest.TestCase):
def test_bad_screenshot_or_existing_target_never_publishes_manifest(self) -> None:
with TemporaryDirectory() as temporary:
target = Path(temporary) / "result"
- with self.assertRaises(SkuSelectionScreenshotError):
+ with self.assertRaises(SkuSelectionScreenshotError) as screenshot_failure:
self._runner(_FakeAdb(), _RawDevice(screenshot="not-image")).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
+ self.assertEqual(safe_failure_stage(screenshot_failure.exception), "screenshot_capture")
self.assertFalse(target.exists())
self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
@@ -418,8 +452,9 @@ class SkuSelectionRunnerTests(unittest.TestCase):
target.mkdir()
sentinel = target / "keep"
sentinel.write_text("keep", encoding="utf-8")
- with self.assertRaises(SkuSelectionRunError):
+ with self.assertRaises(SkuSelectionRunError) as existing_target_failure:
self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
+ self.assertEqual(safe_failure_stage(existing_target_failure.exception), "precheck")
self.assertEqual(adb.calls, [])
self.assertEqual(device.calls, [])
self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")
@@ -654,25 +689,43 @@ class SkuSelectionCliTests(unittest.TestCase):
def test_cli_main_catches_flow_error_without_traceback_or_page_body(self) -> None:
script = _load_runner_script()
- class FlowFailingRunner:
- def __init__(self, *args: object, **kwargs: object) -> None: pass
- def run(self, *args: object, **kwargs: object) -> object:
- raise SkuSelectionError("page-body")
+ secret = "SERIAL=192.168.0.173:5555 PATH=C:/private page-body"
- stderr = BytesIO()
- # TextIOWrapper keeps the assertion independent from host console encoding.
- import io
- text_stderr = io.TextIOWrapper(stderr, encoding="utf-8")
- with patch.object(script, "SkuSelectionRunner", FlowFailingRunner), redirect_stderr(text_stderr):
- status = script.main([
- "--serial", "device-1", "--url", _TARGET_URL, "--color", _TASK_COLOR,
- "--size", _TASK_SIZE, "--output-dir", "evidence",
- ])
- text_stderr.flush()
- output = stderr.getvalue().decode("utf-8")
- self.assertEqual(status, 1)
- self.assertNotIn("Traceback", output)
- self.assertNotIn("page-body", output)
+ class HostileGetterError(SkuSelectionError):
+ def __getattribute__(self, name: str) -> object:
+ if name == "_cmbuyer_failure_stage":
+ raise RuntimeError(secret)
+ return super().__getattribute__(name)
+
+ class HostileStage(str):
+ def __hash__(self) -> int:
+ raise RuntimeError(secret)
+
+ hostile_string = SkuSelectionError(secret)
+ setattr(hostile_string, "_cmbuyer_failure_stage", HostileStage("sku_entry"))
+
+ for failure in (SkuSelectionError(secret), HostileGetterError(secret), hostile_string):
+ class FlowFailingRunner:
+ def __init__(self, *args: object, **kwargs: object) -> None: pass
+ def run(self, *args: object, **kwargs: object) -> object:
+ raise failure
+
+ stderr = BytesIO()
+ # TextIOWrapper keeps the assertion independent from host console encoding.
+ import io
+ text_stderr = io.TextIOWrapper(stderr, encoding="utf-8")
+ with patch.object(script, "SkuSelectionRunner", FlowFailingRunner), redirect_stderr(text_stderr):
+ status = script.main([
+ "--serial", "device-1", "--url", _TARGET_URL, "--color", _TASK_COLOR,
+ "--size", _TASK_SIZE, "--output-dir", "evidence",
+ ])
+ text_stderr.flush()
+ output = stderr.getvalue().decode("utf-8")
+ self.assertEqual(status, 1)
+ self.assertIn("stage=unknown", output)
+ self.assertNotIn("Traceback", output)
+ self.assertNotIn(secret, output)
+ self.assertNotIn("page-body", output)
def _load_runner_script() -> object: