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))
|
