from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timezone import hashlib import re import time from typing import Any, Mapping, Sequence from .adb import AdbClient, UiNode from .models import Candidate, ContractError, ErrorCode from .selection import FastScanner, ScanPage, fingerprint_text, score_candidate from .quota import AppQuotaObservation ACCESS_TERMS = ("验证码", "付费", "开通会员", "订阅后阅读", "登录后阅读") @dataclass(frozen=True) class UiConfig: search_entry_texts: tuple[str, ...] search_input_resource_id: str search_submit_texts: tuple[str, ...] result_resource_id: str open_texts: tuple[str, ...] institution_resource_id: str date_resource_id: str analysts_resource_id: str page_count_resource_id: str swipe: tuple[int, int, int, int, int] quota_remaining_resource_id: str = "" quota_remaining_pattern: str = "" @classmethod def from_source_scope(cls, source_scope: Mapping[str, Any]) -> "UiConfig": ui = source_scope.get("ui") if isinstance(source_scope, Mapping) else None if not isinstance(ui, Mapping): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "source_scope.ui", "mapping required") try: swipe = tuple(int(x) for x in ui["swipe"]) value = cls(tuple(ui["search_entry_texts"]), str(ui["search_input_resource_id"]), tuple(ui["search_submit_texts"]), str(ui["result_resource_id"]), tuple(ui["open_texts"]), str(ui.get("institution_resource_id", "")), str(ui.get("date_resource_id", "")), str(ui.get("analysts_resource_id", "")), str(ui.get("page_count_resource_id", "")), swipe, str(ui.get("quota_remaining_resource_id", "")), str(ui.get("quota_remaining_pattern", ""))) except (KeyError, TypeError, ValueError) as exc: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "source_scope.ui", "invalid config") from exc if len(value.swipe) != 5: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "swipe", "five ints") return value class HiborUiDriver: def __init__(self, adb: AdbClient, config: UiConfig): self.adb = adb self.config = config @staticmethod def _center(node: UiNode) -> tuple[int, int]: x1, y1, x2, y2 = node.bounds return (x1 + x2) // 2, (y1 + y2) // 2 @staticmethod def _assert_no_access(nodes: Sequence[UiNode]) -> None: visible = "\n".join(node.text for node in nodes) if any(term in visible for term in ACCESS_TERMS): raise ContractError(ErrorCode.ACCESS_CONTROL_PRESENT, "ui", "access boundary visible") def search(self, query: str) -> None: nodes = self.adb.ui_nodes() started = False for _ in range(20): self._assert_no_access(nodes) field = next(( n for n in nodes if n.resource_id == self.config.search_input_resource_id ), None) if field is not None: break search_icon = next(( n for n in nodes if n.resource_id == "cn.com.hibor:id/iv_search" ), None) entry = next((n for n in nodes if n.text in self.config.search_entry_texts), None) if search_icon is not None: self.adb.tap(*self._center(search_icon)) elif any(n.resource_id.endswith("detail_iv_left") for n in nodes): self.adb.back() elif entry is not None: self.adb.tap(*self._center(entry)) elif not started: self.adb.start_package() started = True time.sleep(0.25) nodes = self.adb.ui_nodes() else: raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "search_input", "anchor absent") self.adb.tap(*self._center(field)) self.adb.input_text(query) nodes = self.adb.ui_nodes() exact_field = next(( n for n in nodes if n.resource_id == self.config.search_input_resource_id and n.text == query ), None) if exact_field is None: raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "search_input", "text mismatch") submit = next((n for n in nodes if n.text in self.config.search_submit_texts), None) if submit is None: raise ContractError(ErrorCode.UI_ANCHOR_DRIFT, "search_submit", "anchor absent") self.adb.tap(*self._center(submit)) def observe_app_remaining(self, quota_date: str) -> AppQuotaObservation | None: """Return a durable-observation preimage only when the configured UI anchor is exact.""" if not self.config.quota_remaining_resource_id or not self.config.quota_remaining_pattern: return None nodes = self.adb.ui_nodes() self._assert_no_access(nodes) matched = [node for node in nodes if node.resource_id == self.config.quota_remaining_resource_id] if len(matched) != 1: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "app_quota", "anchor not unique") pattern = re.fullmatch(self.config.quota_remaining_pattern, matched[0].text) if pattern is None or len(pattern.groups()) != 1: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "app_quota", "value mismatch") remaining = int(pattern.group(1)) captured = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") fingerprint = fingerprint_text( f"{node.resource_id}|{node.text}|{node.bounds}" for node in nodes ) return AppQuotaObservation.build( device_serial=str(self.adb.serial or ""), quota_date=quota_date, captured_at_utc=captured, visible_remaining=remaining, ui_snapshot_fingerprint=fingerprint, ) def scan(self, *, query: str, quantity: int, aliases: Sequence[str], institutions: Sequence[str], minimum_pages: int, min_screens: int, normal_max_screens: int, hard_max_screens: int, hard_max_candidates: int): def provider(screen: int) -> ScanPage: nodes = self.adb.ui_nodes() self._assert_no_access(nodes) texts = [n.text for n in nodes if n.text] candidates: list[Candidate] = [] for index, node in enumerate(nodes): if not node.text: continue if self.config.result_resource_id and node.resource_id != self.config.result_resource_id: continue if not self.config.result_resource_id and query.casefold() not in node.text.casefold(): continue candidate_id = hashlib.sha256(f"{node.text}|{node.bounds}".encode("utf-8")).hexdigest()[:24] candidate = Candidate(candidate_id, node.text, None, None, (), None, 0, node.bounds) candidate = Candidate(**{**candidate.__dict__, "score": score_candidate( candidate, query=query, aliases=aliases, institutions=institutions, minimum_pages=minimum_pages)}) candidates.append(candidate) page = ScanPage(tuple(candidates), fingerprint_text(texts)) if screen < hard_max_screens: self.adb.swipe(*self.config.swipe) return page return FastScanner(provider).scan(quantity=quantity, min_screens=min_screens, normal_max_screens=normal_max_screens, hard_max_screens=hard_max_screens, hard_max_candidates=hard_max_candidates) def open_detail(self, candidate: Candidate, *, max_locate_screens: int = 20) -> Candidate: located = None for screen in range(1, max_locate_screens + 1): nodes = self.adb.ui_nodes() self._assert_no_access(nodes) located = next(( node for node in nodes if node.text == candidate.title and ( not self.config.result_resource_id or node.resource_id == self.config.result_resource_id ) ), None) if located is not None: break if screen < max_locate_screens: self.adb.swipe(*self.config.swipe) if located is None: raise ContractError(ErrorCode.UI_CURSOR_RESTORE_FAILED, "candidate", "title not relocated") self.adb.tap(*self._center(located)) nodes = self.adb.ui_nodes() self._assert_no_access(nodes) def value_for(resource_id: str) -> str | None: if not resource_id: return None node = next((n for n in nodes if n.resource_id == resource_id and n.text), None) return node.text.strip() if node else None def value_after_label(label: str) -> str | None: for index, node in enumerate(nodes): if node.text != label: continue for following in nodes[index + 1:]: if following.text and following.resource_id.endswith("detail_tv_typecontent"): return following.text.strip() if following.text.endswith("："): break return None institution = value_for(self.config.institution_resource_id) if not institution: institution = value_after_label("研报出处：") report_date = value_for(self.config.date_resource_id) analyst_values = tuple( n.text.strip() for n in nodes if self.config.analysts_resource_id and n.resource_id == self.config.analysts_resource_id and n.text.strip() ) page_text = value_for(self.config.page_count_resource_id) if not page_text or not re.search(r"\d+\s*页", page_text): page_text = value_after_label("研报大小：") pages = None if page_text: match = re.search(r"(\d+)\s*页", page_text) pages = int(match.group(1)) if match else None date_match = re.search( r"\b(20\d{2}[-./]\d{1,2}[-./]\d{1,2})\b", report_date or "\n".join(n.text for n in nodes), ) report_date = (date_match.group(1).replace("/", "-").replace(".", "-") if date_match else None) analysts = tuple( part for value in analyst_values for part in (x.strip() for x in re.split(r"[、,，;/]", value)) if part ) if not institution or not report_date or pages is None: raise ContractError(ErrorCode.DETAIL_RESULT_MISMATCH, "detail", "institution/date/page required") return Candidate(candidate.candidate_id, candidate.title, institution, report_date, analysts, pages, candidate.score, candidate.bounds) def trigger_current_detail(self) -> None: nodes = self.adb.ui_nodes() self._assert_no_access(nodes) opener = next((n for n in nodes if n.text in self.config.open_texts), None) if opener is None: raise ContractError(ErrorCode.DETAIL_RESULT_MISMATCH, "open", "anchor absent") self.adb.tap(*self._center(opener))