5cb347a44b
Rebuilt the capture and mapping pipeline after an audit found the simulator's data could not be trusted: * Hotspot coordinates never matched the screenshots. Capture now scrolls the page over CDP and pastes each frame at the measured scrollY, so image pixels and DOM coordinates share one grid by construction. * Metrics were synthesised (1200 + n*410) and presented as analytics. Numbers are now attached only when the catalog has a matching row; metrics.json carries a `source` label and the UI says "no data" instead of showing zeros. * Event interception hooked a connector bridge that never fires. The app posts to api.amplitude.com using the legacy form-urlencoded v1 API; the hook now reads event_type off the wire. 36 keys are verified as `observed`. * All device access moved into tools/telecom_cdp.py: dynamic WebView socket discovery (the PID was hardcoded), id-matched CDP, measured native geometry. * Editor edits can now be saved to disk; API failures no longer report success from a stale result file; screenId is no longer interpolated into a shell. Screens went from 7 (with fabricated markup) to 32, all verified: image height equals map height, no out-of-bounds hotspots, no dead links. The id_card screenshot has been manually redacted - it showed a national ID. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
469 lines
19 KiB
Python
469 lines
19 KiB
Python
"""
|
|
Walk every control on one screen: record the event key it really fires, and capture
|
|
whatever screen it opens.
|
|
|
|
This is the workflow that turns a screen from "drawn" into "mapped". For each DOM
|
|
control it clicks the element, reads the analytics event off the wire, notes whether
|
|
the route changed, and - with --capture-new - screenshots and maps the destination.
|
|
|
|
py tools/map_screen_controls.py payments_screen --via "Платежи" --capture-new
|
|
py tools/map_screen_controls.py main_dashboard --capture-new --skip "Лицевой счет"
|
|
|
|
Why clicks and not taps: DOM clicks reach controls below the fold without scrolling
|
|
maths, and they cannot miss. Coordinates are only used for native chrome.
|
|
|
|
Safety rails, all learned the hard way on this app:
|
|
* refuses to touch anything once the app leaves the expected route, and never taps
|
|
into a PIN screen;
|
|
* reconnects per control, because opening some sections destroys the CDP target;
|
|
* a settle long enough that an event chain from the previous control cannot be
|
|
misattributed to the next one.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
|
|
import telecom_cdp as T
|
|
from amplitude_hook import DRAIN_JS, INSTALL_HOOK_JS
|
|
from capture_screen import capture, dom_elements_js, slugify_event_key
|
|
|
|
FORBIDDEN_ROUTES = ("pincode", "verification", "login", "auth")
|
|
|
|
LIST_CONTROLS_JS_TEMPLATE = """
|
|
(() => {
|
|
const found = %s;
|
|
// Only controls that are actually on this screen horizontally. The side drawer is
|
|
// a fixed panel parked off-screen at x ~= 670 when closed, and its rows stay in the
|
|
// DOM at full size - without this clamp the walk clicks menu rows while the drawer
|
|
// is shut and files their keys against the dashboard.
|
|
// Number repeated captions. A tariff list has five identical "Подробнее"
|
|
// buttons; without an occurrence index every one of them resolves to the first.
|
|
const seen = {};
|
|
return found.elements
|
|
.filter(e => e.cssLeft < window.innerWidth && e.cssLeft + e.cssWidth > 0)
|
|
.map((e, i) => {
|
|
const n = seen[e.text] = (seen[e.text] || 0) + 1;
|
|
return { i: i, text: e.text, nth: n - 1, top: Math.round(e.cssTop) };
|
|
});
|
|
})()
|
|
"""
|
|
|
|
CLICK_NTH_JS = """
|
|
((wanted, nth) => {
|
|
const found = %s;
|
|
const el = document.querySelectorAll(
|
|
'button, a[href], [role="button"], [onclick], .menu-list-item, .extra-menu__card,'
|
|
+ ' .bonuses-card, .user-balance-card, .customer-account-card__item,'
|
|
+ ' [class*="banner"], [class*="promo"], .swiper-slide, .nav-link'
|
|
);
|
|
// Re-resolve by caption: the DOM can re-render between listing and clicking.
|
|
// Match only among controls actually on screen. The closed side drawer keeps
|
|
// full-size rows parked at x ~= 670, several of whose captions are identical to
|
|
// dashboard tiles ("Мои услуги", "Платежи") - and the drawer row comes first in
|
|
// document order, so an unclamped match returns MYSERVICES where the tile really
|
|
// fires HOMETAPMYSERVICES.
|
|
// Also require the element to be the one a finger would actually hit there. With
|
|
// the drawer OPEN both its row and the dashboard tile are on screen and share a
|
|
// caption; only the row is hittable, and clicking the covered tile silently
|
|
// measures the wrong control (HOMEPAGEPAYMENTS instead of the drawer's PAYMENTS).
|
|
const candidates = [...el].filter(node => {
|
|
const t = (node.innerText || node.getAttribute('aria-label') || '').trim().replace(/\\s+/g, ' ');
|
|
if (!t || t.slice(0, 80) !== wanted) return false;
|
|
const r = node.getBoundingClientRect();
|
|
return r.left < window.innerWidth && r.right > 0;
|
|
});
|
|
const hittable = candidates.filter(node => {
|
|
const r = node.getBoundingClientRect();
|
|
const x0 = Math.max(r.left, 0), x1 = Math.min(r.right, window.innerWidth);
|
|
const y0 = Math.max(r.top, 0), y1 = Math.min(r.bottom, window.innerHeight);
|
|
if (x1 <= x0 || y1 <= y0) return true; // off-screen vertically: cannot test
|
|
const top = document.elementFromPoint((x0 + x1) / 2, (y0 + y1) / 2);
|
|
return top && (node.contains(top) || top.contains(node));
|
|
});
|
|
const pool = hittable.length ? hittable : candidates;
|
|
const hit = pool[nth || 0] || pool[0];
|
|
if (!hit) return { clicked: false };
|
|
hit.scrollIntoView({ block: 'center' });
|
|
hit.click();
|
|
return { clicked: true };
|
|
})
|
|
"""
|
|
|
|
|
|
def slug(text, index):
|
|
"""
|
|
Screen id from a caption. Transliterates, because these captions are Cyrillic and
|
|
dropping non-ASCII characters turns every one of them into the same empty stem.
|
|
"""
|
|
key = slugify_event_key(text) # CLICK_MOY_AVTOPLATEZH
|
|
base = key[6:] if key.startswith("CLICK_") else key
|
|
base = base.strip("_").lower()[:28]
|
|
return "scr_" + (base or "x") + "_" + str(index)
|
|
|
|
|
|
class Walker:
|
|
def __init__(self, screen_id, route, settle, capture_new, skip, via=None):
|
|
self.screen_id = screen_id
|
|
self.route = route
|
|
self.settle = settle
|
|
self.capture_new = capture_new
|
|
self.skip = [s.lower() for s in skip]
|
|
self.via = via
|
|
self.device = T.get_device()
|
|
self.pairs = []
|
|
self.discovered = []
|
|
self._session = None
|
|
|
|
# --- plumbing ---------------------------------------------------------
|
|
def session(self):
|
|
"""
|
|
Reuse one CDP session across the walk.
|
|
|
|
T.connect() costs about five adb round trips (lock check, foreground check,
|
|
pidof, socket scan, forward). Paying that for every route poll made a single
|
|
screen take minutes. The session is only rebuilt when the page really goes
|
|
away, which is the case the reconnect exists for.
|
|
"""
|
|
if self._session is not None and not getattr(self._session, "closed", False):
|
|
return self._session
|
|
_, tgt = T.connect(self.device)
|
|
self._session = T.CdpSession(tgt["webSocketDebuggerUrl"])
|
|
return self._session
|
|
|
|
def drop_session(self):
|
|
if self._session is not None:
|
|
self._session.close()
|
|
self._session = None
|
|
|
|
def adb_key(self, code):
|
|
subprocess.run(
|
|
[T.find_adb(), "-s", self.device, "shell", "input", "keyevent", str(code)],
|
|
capture_output=True,
|
|
)
|
|
|
|
def go_back(self):
|
|
"""
|
|
Step back inside the WebView rather than pressing the hardware back key.
|
|
|
|
The hardware key closes the whole app once the history is empty, and the app
|
|
then demands a PIN on relaunch. history.back() can never do that, so it is
|
|
always tried first; the key is only a fallback when the route refuses to move.
|
|
"""
|
|
before = self.current_route()
|
|
try:
|
|
self.session().evaluate("window.history.back()")
|
|
time.sleep(2.0)
|
|
if self.current_route() != before:
|
|
return True
|
|
except T.DeviceError:
|
|
self.drop_session()
|
|
# Only now risk the hardware key, and never at the app root.
|
|
if before in (None, "/"):
|
|
return False
|
|
self.adb_key(4)
|
|
time.sleep(2.0)
|
|
return self.current_route() != before
|
|
|
|
def current_route(self):
|
|
try:
|
|
return self.session().evaluate("location.pathname")
|
|
except T.DeviceError:
|
|
self.drop_session()
|
|
return None
|
|
|
|
def navigate_from_home(self):
|
|
"""Click the dashboard entry point that opens this screen."""
|
|
if not self.via:
|
|
return False
|
|
for _ in range(5):
|
|
if self.current_route() == "/":
|
|
break
|
|
if not self.go_back():
|
|
break
|
|
if self.current_route() != "/":
|
|
return False
|
|
try:
|
|
s = self.session()
|
|
js = CLICK_NTH_JS % dom_elements_js(False)
|
|
hit = json.loads(
|
|
s.evaluate("JSON.stringify((" + js + ")(" + json.dumps(self.via) + ", 0))")
|
|
)
|
|
except T.DeviceError:
|
|
self.drop_session()
|
|
return False
|
|
if not hit.get("clicked"):
|
|
print(" entry point " + repr(self.via) + " not found on the dashboard")
|
|
return False
|
|
time.sleep(self.settle)
|
|
return self.current_route() == self.route
|
|
|
|
MENU_BUTTON = (1012, 174)
|
|
|
|
def ensure_drawer(self):
|
|
"""Re-open the side drawer if a navigation closed it."""
|
|
if self.drawer_is_open():
|
|
return True
|
|
subprocess.run(
|
|
[T.find_adb(), "-s", self.device, "shell", "input", "tap",
|
|
str(self.MENU_BUTTON[0]), str(self.MENU_BUTTON[1])],
|
|
capture_output=True,
|
|
)
|
|
time.sleep(3.0)
|
|
return self.drawer_is_open()
|
|
|
|
def back_to_screen(self, tries=5):
|
|
for _ in range(tries):
|
|
here = self.current_route()
|
|
if here == self.route:
|
|
if self.screen_id == "side_menu" and not self.ensure_drawer():
|
|
print(" !! could not re-open the drawer")
|
|
return False
|
|
return True
|
|
if here == "/" and self.via:
|
|
if self.navigate_from_home():
|
|
return True
|
|
if here and any(bad in here.lower() for bad in FORBIDDEN_ROUTES):
|
|
# A PIN gate is a normal outcome for some controls. Step back out of
|
|
# it and carry on rather than abandoning the walk - and never tap
|
|
# anything while it is on screen.
|
|
print(" (PIN gate reached; backing out without touching it)")
|
|
if not self.go_back():
|
|
print(" !! stuck on " + here + " - stopping this walk")
|
|
return False
|
|
continue
|
|
if not self.go_back():
|
|
break
|
|
return self.current_route() == self.route
|
|
|
|
# --- the walk ---------------------------------------------------------
|
|
def list_controls(self):
|
|
js = LIST_CONTROLS_JS_TEMPLATE % dom_elements_js(False)
|
|
return json.loads(self.session().evaluate("JSON.stringify(" + js + ")"))
|
|
|
|
def probe(self, caption, nth=0):
|
|
"""Click one control, return (event_key, via, new_route)."""
|
|
try:
|
|
session = self.session()
|
|
session.evaluate(INSTALL_HOOK_JS)
|
|
session.evaluate(DRAIN_JS)
|
|
js = CLICK_NTH_JS % dom_elements_js(False)
|
|
hit = json.loads(
|
|
session.evaluate(
|
|
"JSON.stringify((" + js + ")(" + json.dumps(caption) + ", " + str(nth) + "))"
|
|
)
|
|
)
|
|
if not hit.get("clicked"):
|
|
return None, "not found", None
|
|
except T.DeviceError as exc:
|
|
self.drop_session()
|
|
return None, "error: " + str(exc), None
|
|
|
|
# Poll rather than sleep: some controls replace the WebView within a second.
|
|
amplitude, metrika, new_route = [], [], None
|
|
deadline = time.time() + self.settle
|
|
while time.time() < deadline:
|
|
time.sleep(0.4)
|
|
try:
|
|
for item in session.evaluate(DRAIN_JS) or []:
|
|
if item.get("kind") == "amplitude" and item.get("eventType"):
|
|
amplitude.append(item["eventType"])
|
|
elif item.get("kind") == "metrika-goal" and item.get("eventType"):
|
|
metrika.append(item["eventType"])
|
|
new_route = session.evaluate("location.pathname")
|
|
except T.DeviceError:
|
|
new_route = None # target destroyed by this control
|
|
self.drop_session()
|
|
break
|
|
|
|
if amplitude:
|
|
return amplitude[0], "amplitude", new_route
|
|
if metrika:
|
|
return metrika[0], "metrika-goal", new_route
|
|
return None, "no event", new_route
|
|
|
|
DRAWER_OPEN_JS = (
|
|
"(() => { const p = document.querySelector('.mobile-navigation');"
|
|
" return !!p && p.getBoundingClientRect().left < window.innerWidth * 0.5; })()"
|
|
)
|
|
|
|
def drawer_is_open(self):
|
|
try:
|
|
return bool(self.session().evaluate(self.DRAWER_OPEN_JS))
|
|
except T.DeviceError:
|
|
self.drop_session()
|
|
return False
|
|
|
|
def run(self):
|
|
if not self.back_to_screen():
|
|
print("Cannot reach " + self.route + "; aborting.")
|
|
return
|
|
|
|
# The drawer and the dashboard share captions ("Платежи", "Мои услуги", ...).
|
|
# With it open, a click resolves to whichever comes first in the DOM and the
|
|
# key would be written onto the wrong control - the drawer's PAYMENTS would
|
|
# overwrite the tile's HOMEPAGEPAYMENTS. Refuse rather than guess.
|
|
drawer = self.drawer_is_open()
|
|
if drawer and self.screen_id != "side_menu":
|
|
print("ABORT: the side drawer is open; close it before walking " + self.screen_id)
|
|
return
|
|
if not drawer and self.screen_id == "side_menu":
|
|
print("ABORT: the side drawer is closed; open it before walking side_menu")
|
|
return
|
|
|
|
controls = self.list_controls()
|
|
print("Controls on " + self.screen_id + ": " + str(len(controls)))
|
|
|
|
for c in controls:
|
|
caption = c["text"]
|
|
if any(sk in caption.lower() for sk in self.skip):
|
|
print(" skip " + caption[:44])
|
|
continue
|
|
if not self.back_to_screen():
|
|
break
|
|
|
|
key, via, new_route = self.probe(caption, c.get("nth", 0))
|
|
changed = new_route not in (None, self.route)
|
|
note = key + " [" + via + "]" if key else "(" + via + ")"
|
|
print(" " + caption[:40].ljust(42) + note + (" -> " + str(new_route) if changed else ""))
|
|
|
|
if key and via in ("amplitude", "metrika-goal"):
|
|
self.pairs.append((caption, key, via, c.get("nth", 0)))
|
|
|
|
if changed and self.capture_new:
|
|
self.capture_destination(caption, new_route)
|
|
|
|
if new_route is None:
|
|
# The page was replaced (in-app browser or another WebView); only the
|
|
# hardware key can leave that, and the app is not at its root here.
|
|
self.adb_key(4)
|
|
time.sleep(2.5)
|
|
|
|
self.write_keys()
|
|
|
|
def capture_destination(self, caption, route):
|
|
# Several controls lead to a screen that is already in the map (the dashboard
|
|
# tile and the drawer row both open /payments). Re-capturing it under a new id
|
|
# would just litter the map with duplicates, so reuse the existing screen and
|
|
# only record the link.
|
|
existing = next(
|
|
(s for s in T.load_app_map().get("screens", []) if s.get("route") == route),
|
|
None,
|
|
)
|
|
if existing is not None:
|
|
print(" -> already mapped as " + existing["id"] + ", linking only")
|
|
self.discovered.append(
|
|
{"id": existing["id"], "from": caption, "route": route, "caption": caption}
|
|
)
|
|
return
|
|
|
|
screen_id = slug(caption, len(self.discovered) + 1)
|
|
self.drop_session() # capture() opens its own session to the same page
|
|
try:
|
|
res = capture(screen_id, "Экран «" + caption[:34] + "»", "Раздел")
|
|
print(
|
|
" captured "
|
|
+ screen_id
|
|
+ " "
|
|
+ str(res["dimensions"]["width"])
|
|
+ "x"
|
|
+ str(res["dimensions"]["height"])
|
|
+ " ("
|
|
+ str(len(res["hotspots"]))
|
|
+ " зон)"
|
|
)
|
|
self.discovered.append(
|
|
{"id": screen_id, "from": caption, "route": route, "caption": caption}
|
|
)
|
|
except T.DeviceError as exc:
|
|
print(" capture failed: " + str(exc))
|
|
|
|
def write_keys(self):
|
|
if not self.pairs:
|
|
print("\nNo event keys observed on " + self.screen_id + ".")
|
|
return
|
|
app_map = T.load_app_map()
|
|
catalog = T.load_metrics_catalog()
|
|
source = T.metrics_source()
|
|
updated = 0
|
|
for screen in app_map.get("screens", []):
|
|
if screen["id"] != self.screen_id:
|
|
continue
|
|
for hs in screen.get("hotspots", []):
|
|
label = (hs.get("label") or "").strip().lower()
|
|
for caption, key, _via, _nth in self.pairs:
|
|
if not label or label != caption.strip().lower():
|
|
continue
|
|
existing = hs.get("eventKey")
|
|
if hs.get("keyConfidence") == "observed" and existing != key:
|
|
# Two verified readings disagree. Keep the earlier one and
|
|
# report it: silently replacing a confirmed key with another
|
|
# destroys evidence and hides the discrepancy.
|
|
print(
|
|
" ! conflict on "
|
|
+ hs["label"][:30]
|
|
+ ": keeping "
|
|
+ str(existing)
|
|
+ ", not overwriting with "
|
|
+ key
|
|
)
|
|
break
|
|
hs["eventKey"] = key
|
|
hs["keyConfidence"] = "observed"
|
|
T.attach_metrics(hs, catalog, source)
|
|
updated += 1
|
|
break
|
|
# Point the control at the screen it was observed to open, so the
|
|
# simulator can actually follow the transition.
|
|
for found in self.discovered:
|
|
if label and label == found["caption"].strip().lower():
|
|
hs["targetScreenId"] = found["id"]
|
|
break
|
|
if updated:
|
|
T.save_app_map(app_map)
|
|
print("\nObserved keys written into " + self.screen_id + ": " + str(updated))
|
|
for caption, key, via, nth in self.pairs:
|
|
suffix = " #" + str(nth + 1) if nth else ""
|
|
print(" " + key + " <- " + caption[:44] + suffix + " [" + via + "]")
|
|
if self.discovered:
|
|
print("\nNew screens captured: " + str(len(self.discovered)))
|
|
for d in self.discovered:
|
|
print(" " + d["id"] + " <- " + d["from"][:34] + " route " + str(d["route"]))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Map every control on one screen.")
|
|
parser.add_argument("screen_id")
|
|
parser.add_argument("--route", default=None, help="Expected route (defaults to the map's)")
|
|
parser.add_argument("--settle", type=float, default=6.0)
|
|
parser.add_argument("--capture-new", action="store_true", help="Screenshot screens that open")
|
|
parser.add_argument("--skip", nargs="*", default=[], help="Caption fragments to leave alone")
|
|
parser.add_argument(
|
|
"--via", default=None, help="Exact dashboard caption that opens this screen"
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
route = args.route
|
|
if not route:
|
|
app_map = T.load_app_map()
|
|
screen = next((s for s in app_map["screens"] if s["id"] == args.screen_id), None)
|
|
route = (screen or {}).get("route") or "/"
|
|
|
|
try:
|
|
Walker(
|
|
args.screen_id, route, args.settle, args.capture_new, args.skip, args.via
|
|
).run()
|
|
except T.DeviceError as exc:
|
|
print("Error: " + str(exc))
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|