diff --git a/client/src/pdd_u2_purchase_reconcile_adapter.py b/client/src/pdd_u2_purchase_reconcile_adapter.py
index b26f9fc..a8c8db4 100644
--- a/client/src/pdd_u2_purchase_reconcile_adapter.py
+++ b/client/src/pdd_u2_purchase_reconcile_adapter.py
@@ -50,6 +50,8 @@ _PAYMENT_ACTION_MARKERS = ("立即支付", "确认支付", "输入支付密码")
_UNPAID_MARKERS = ("待付款", "待支付")
_NON_UNPAID_MARKERS = ("已付款", "交易成功", "交易完成", "已取消", "退款")
_SAFE_NAVIGATION_LABELS = ("个人中心", "我的订单", "待付款")
+_ORDER_DETAIL_PAY_LABELS = ("去支付",)
+_ORDER_DETAIL_CANCEL_LABELS = ("取消订单",)
class PddPurchaseReconcileError(RuntimeError):
@@ -83,6 +85,38 @@ def _subtree_text(node: ET.Element) -> str:
)
+def _package_hint_from_tree(root: ET.Element) -> str:
+ """可靠控件树可以证明 PDD 在前台,避免错误的前台查询结果。"""
+
+ pdd_node_count = sum(
+ 1
+ for node in root.iter("node")
+ if node.get("package") == PDD_PACKAGE_NAME
+ )
+ return PDD_PACKAGE_NAME if pdd_node_count >= 3 else ""
+
+
+def _label_count(root: ET.Element, targets: tuple[str, ...]) -> int:
+ return sum(
+ 1
+ for node in root.iter("node")
+ if _label(node) in targets
+ )
+
+
+def _is_single_order_detail(root: ET.Element) -> bool:
+ """只在页面上只有一组订单操作时允许跨视口累积字段。"""
+
+ pay_count = _label_count(root, _ORDER_DETAIL_PAY_LABELS)
+ cancel_count = _label_count(root, _ORDER_DETAIL_CANCEL_LABELS)
+ order_numbers = {
+ match.group(1)
+ for node in root.iter("node")
+ if (match := _ORDER_NO_PATTERN.search(_label(node))) is not None
+ }
+ return pay_count == 1 and cancel_count <= 1 and len(order_numbers) <= 1
+
+
def _parse_bounds(value: str) -> Optional[Bounds]:
match = _BOUNDS_PATTERN.fullmatch(str(value or "").strip())
if match is None:
@@ -141,6 +175,16 @@ def _parse_ordered_at(text: str) -> tuple[str, str]:
return "", raw
+def _payment_status(text: str) -> str:
+ """只在页面有明确状态文字时返回状态。"""
+
+ if any(marker in text for marker in _NON_UNPAID_MARKERS):
+ return "other"
+ if any(marker in text for marker in _UNPAID_MARKERS):
+ return "unpaid"
+ return ""
+
+
def parse_order_candidates(
xml_data: str | bytes, query: PurchaseReconcileQuery
) -> tuple[PurchaseOrderCandidate, ...]:
@@ -184,12 +228,6 @@ def parse_order_candidates(
for key, value in query.options.items()
if str(value).strip() and str(value).strip() in card_text
}
- payment_status = (
- "unpaid"
- if any(marker in card_text for marker in _UNPAID_MARKERS)
- and not any(marker in card_text for marker in _NON_UNPAID_MARKERS)
- else "other"
- )
candidate = PurchaseOrderCandidate(
order_no=order_no,
goods_id=goods_match.group(1) if goods_match else "",
@@ -198,7 +236,7 @@ def parse_order_candidates(
total_price_cent=_parse_money_cent(card_text),
ordered_at=ordered_at,
ordered_at_raw=ordered_at_raw,
- payment_status=payment_status,
+ payment_status=_payment_status(card_text),
)
previous = parsed.get(order_no)
if previous is None or previous == candidate:
@@ -206,6 +244,81 @@ def parse_order_candidates(
return tuple(parsed.values())
+def _parse_order_detail_evidence(
+ root: ET.Element, query: PurchaseReconcileQuery
+) -> PurchaseOrderCandidate:
+ """读取单个订单详情当前视口;字段可以暂时不完整。"""
+
+ text = _subtree_text(root)
+ order_numbers = tuple(dict.fromkeys(_ORDER_NO_PATTERN.findall(text)))
+ goods_match = _GOODS_ID_PATTERN.search(text)
+ ordered_at, ordered_at_raw = _parse_ordered_at(text)
+ options = {
+ key: value
+ for key, value in query.options.items()
+ if str(value).strip() and str(value).strip() in text
+ }
+ return PurchaseOrderCandidate(
+ order_no=order_numbers[0] if len(order_numbers) == 1 else "",
+ goods_id=goods_match.group(1) if goods_match else "",
+ options=options,
+ quantity=_parse_quantity(text),
+ total_price_cent=_parse_money_cent(text),
+ ordered_at=ordered_at,
+ ordered_at_raw=ordered_at_raw,
+ payment_status=_payment_status(text),
+ )
+
+
+def _merge_order_evidence(
+ previous: PurchaseOrderCandidate,
+ current: PurchaseOrderCandidate,
+) -> Optional[PurchaseOrderCandidate]:
+ """合并同一订单的相邻视口;关键字段冲突时拒绝合并。"""
+
+ scalar_fields = (
+ "order_no",
+ "goods_id",
+ "quantity",
+ "total_price_cent",
+ "ordered_at",
+ "ordered_at_raw",
+ )
+ values: dict[str, object] = {}
+ for field_name in scalar_fields:
+ old_value = getattr(previous, field_name)
+ new_value = getattr(current, field_name)
+ if old_value and new_value and old_value != new_value:
+ return None
+ values[field_name] = old_value or new_value
+
+ options = dict(previous.options)
+ for key, value in current.options.items():
+ if key in options and options[key] != value:
+ return None
+ options[key] = value
+
+ statuses = {
+ value
+ for value in (previous.payment_status, current.payment_status)
+ if value
+ }
+ if "other" in statuses:
+ payment_status = "other"
+ else:
+ payment_status = "unpaid" if "unpaid" in statuses else ""
+ return PurchaseOrderCandidate(
+ order_no=str(values["order_no"]),
+ goods_id=str(values["goods_id"]),
+ options=options,
+ quantity=int(values["quantity"]),
+ total_price_cent=int(values["total_price_cent"]),
+ ordered_at=str(values["ordered_at"]),
+ ordered_at_raw=str(values["ordered_at_raw"]),
+ payment_status=payment_status,
+ )
+
+
class U2PddPurchaseReconcileAdapter(PddPurchaseReconcileAdapter):
"""只导航和滚动读取待付款订单,不包含任何订单写操作。"""
@@ -237,19 +350,47 @@ class U2PddPurchaseReconcileAdapter(PddPurchaseReconcileAdapter):
device = self._connect()
self._open_unpaid_orders(device)
found: dict[str, PurchaseOrderCandidate] = {}
- previous_signature = ""
+ detail_evidence: Optional[PurchaseOrderCandidate] = None
+ detail_context_valid = True
+ previous_xml = ""
pages_scanned = 0
for _ in range(self._max_pages):
self._check_cancelled()
xml_data = device.dump_hierarchy(compressed=False)
- self._raise_for_special_page(_parse_xml(xml_data))
+ root = _parse_xml(xml_data)
+ self._raise_for_special_page(root)
pages_scanned += 1
for candidate in parse_order_candidates(xml_data, query):
- found.setdefault(candidate.order_no, candidate)
- signature = "|".join(sorted(found)) + f":{len(str(xml_data))}"
- if signature == previous_signature:
+ previous = found.get(candidate.order_no)
+ merged = (
+ _merge_order_evidence(previous, candidate)
+ if previous is not None
+ else candidate
+ )
+ if merged is not None:
+ found[candidate.order_no] = merged
+
+ if detail_context_valid and _is_single_order_detail(root):
+ current_evidence = _parse_order_detail_evidence(root, query)
+ if detail_evidence is None:
+ detail_evidence = current_evidence
+ else:
+ merged = _merge_order_evidence(
+ detail_evidence, current_evidence
+ )
+ if merged is None:
+ detail_context_valid = False
+ detail_evidence = None
+ else:
+ detail_evidence = merged
+ else:
+ detail_context_valid = False
+ detail_evidence = None
+
+ current_xml = str(xml_data)
+ if current_xml == previous_xml:
break
- previous_signature = signature
+ previous_xml = current_xml
width, height = device.window_size()
device.swipe(
width // 2,
@@ -259,6 +400,15 @@ class U2PddPurchaseReconcileAdapter(PddPurchaseReconcileAdapter):
0.35,
)
self._settle()
+ if detail_context_valid and detail_evidence and detail_evidence.order_no:
+ previous = found.get(detail_evidence.order_no)
+ merged = (
+ _merge_order_evidence(previous, detail_evidence)
+ if previous is not None
+ else detail_evidence
+ )
+ if merged is not None:
+ found[detail_evidence.order_no] = merged
return PurchaseReconcileScan(
candidates=tuple(found.values()),
diagnostics={
@@ -281,26 +431,32 @@ class U2PddPurchaseReconcileAdapter(PddPurchaseReconcileAdapter):
return self._device
def _open_unpaid_orders(self, device: Any) -> None:
- current = device.app_current()
- if str(current.get("package") or "") != PDD_PACKAGE_NAME:
- device.app_start(PDD_PACKAGE_NAME)
- self._settle()
+ root = _parse_xml(device.dump_hierarchy(compressed=False))
+ if not _package_hint_from_tree(root):
+ current = device.app_current()
+ if str(current.get("package") or "") != PDD_PACKAGE_NAME:
+ device.app_start(PDD_PACKAGE_NAME)
+ self._settle()
opened_order_area = False
for _ in range(6):
self._check_cancelled()
- current = device.app_current()
- if str(current.get("package") or "") != PDD_PACKAGE_NAME:
- raise PddPurchaseReconcileError(
- "RECONCILE_WRONG_APP", "核单时 PDD 不在前台"
- )
root = _parse_xml(device.dump_hierarchy(compressed=False))
+ if not _package_hint_from_tree(root):
+ current = device.app_current()
+ if str(current.get("package") or "") != PDD_PACKAGE_NAME:
+ raise PddPurchaseReconcileError(
+ "RECONCILE_WRONG_APP", "核单时 PDD 不在前台"
+ )
self._raise_for_special_page(root)
combined = _subtree_text(root)
- if opened_order_area and (
- _ORDER_NO_PATTERN.search(combined)
- or "暂无订单" in combined
- or "暂无相关订单" in combined
+ if _is_single_order_detail(root) or (
+ opened_order_area
+ and (
+ _ORDER_NO_PATTERN.search(combined)
+ or "暂无订单" in combined
+ or "暂无相关订单" in combined
+ )
):
return
if any(marker in combined for marker in _PAYMENT_ACTION_MARKERS):
diff --git a/client/test/test_pdd_u2_purchase_reconcile_adapter.py b/client/test/test_pdd_u2_purchase_reconcile_adapter.py
index 27224ba..8709854 100644
--- a/client/test/test_pdd_u2_purchase_reconcile_adapter.py
+++ b/client/test/test_pdd_u2_purchase_reconcile_adapter.py
@@ -1,9 +1,20 @@
"""只读订单核对 XML 解析测试;不连接真实手机。"""
import unittest
+from contextlib import AbstractContextManager
-from src.pdd_purchase_reconcile_adapter import PurchaseReconcileQuery
-from src.pdd_u2_purchase_reconcile_adapter import parse_order_candidates
+from src.pdd_purchase_reconcile_adapter import (
+ PurchaseOrderCandidate,
+ PurchaseReconcileQuery,
+)
+from src.pdd_u2_purchase_reconcile_adapter import (
+ U2PddPurchaseReconcileAdapter,
+ _is_single_order_detail,
+ _merge_order_evidence,
+ _parse_xml,
+ parse_order_candidates,
+)
+from src.purchase_reconcile_service import PurchaseReconcileService
def query() -> PurchaseReconcileQuery:
@@ -28,6 +39,63 @@ def xml_card(text: str) -> str:
)
+def xml_detail(text: str) -> str:
+ package = "com.xunmeng.pinduoduo"
+ return (
+ ''
+ ''
+ f''
+ f''
+ f''
+ ""
+ )
+
+
+class FakeSession(AbstractContextManager):
+ def __init__(self, device):
+ self.device = device
+
+ def __enter__(self):
+ return self.device
+
+ def __exit__(self, exc_type, exc_value, traceback):
+ return None
+
+
+class FakeDeviceService:
+ def __init__(self, device):
+ self.device = device
+
+ def connect(self, device_address):
+ return FakeSession(self.device)
+
+
+class DetailDevice:
+ def __init__(self, pages):
+ self.pages = pages
+ self.page_index = 0
+ self.app_current_calls = 0
+ self.app_start_calls = 0
+ self.swipes = 0
+
+ def dump_hierarchy(self, compressed=False):
+ return self.pages[self.page_index]
+
+ def app_current(self):
+ self.app_current_calls += 1
+ return {"package": "com.android.settings"}
+
+ def app_start(self, package):
+ self.app_start_calls += 1
+
+ def window_size(self):
+ return 1080, 2400
+
+ def swipe(self, *args):
+ self.swipes += 1
+ self.page_index = min(self.page_index + 1, len(self.pages) - 1)
+
+
class PddU2PurchaseReconcileAdapterTest(unittest.TestCase):
def test_parser_extracts_only_required_unpaid_order_fields(self):
candidates = parse_order_candidates(
@@ -70,6 +138,70 @@ class PddU2PurchaseReconcileAdapterTest(unittest.TestCase):
self.assertEqual(candidates[0].goods_id, "")
self.assertEqual(candidates[0].ordered_at, "")
+ def test_adapter_merges_one_order_detail_across_viewports(self):
+ device = DetailDevice(
+ [
+ xml_detail(
+ "待付款 商品编号:737116531267 黑色 L "
+ "共2件 合计 ¥84.00"
+ ),
+ xml_detail(
+ "订单编号:ORDER-20260810 "
+ "下单时间:2026-08-10 16:03:00"
+ ),
+ ]
+ )
+ adapter = U2PddPurchaseReconcileAdapter(
+ "serial",
+ device_service=FakeDeviceService(device),
+ settle_seconds=0,
+ max_pages=3,
+ )
+
+ scan = adapter.read_order_candidates(query())
+ adapter.close()
+
+ self.assertEqual(len(scan.candidates), 1)
+ candidate = scan.candidates[0]
+ self.assertEqual(candidate.order_no, "ORDER-20260810")
+ self.assertEqual(candidate.goods_id, "737116531267")
+ self.assertEqual(candidate.options, {"color": "黑色", "size": "L"})
+ self.assertEqual(candidate.quantity, 2)
+ self.assertEqual(candidate.total_price_cent, 8400)
+ self.assertEqual(candidate.ordered_at, "2026-08-10T08:03:00Z")
+ self.assertEqual(candidate.payment_status, "unpaid")
+ self.assertTrue(PurchaseReconcileService._candidate_matches(candidate, query()))
+ self.assertEqual(device.app_current_calls, 0)
+ self.assertEqual(device.app_start_calls, 0)
+
+ def test_conflicting_order_numbers_are_not_merged(self):
+ first = PurchaseOrderCandidate(
+ order_no="ORDER-A",
+ goods_id="737116531267",
+ payment_status="unpaid",
+ )
+ second = PurchaseOrderCandidate(
+ order_no="ORDER-B",
+ quantity=2,
+ total_price_cent=8400,
+ payment_status="unpaid",
+ )
+
+ self.assertIsNone(_merge_order_evidence(first, second))
+
+ def test_multiple_order_actions_disable_detail_evidence_merge(self):
+ root = _parse_xml(
+ ''
+ ''
+ ''
+ ''
+ ""
+ )
+
+ self.assertFalse(_is_single_order_detail(root))
+
if __name__ == "__main__":
unittest.main()