telecomkz_scraper/tools/live_auto_recorder.py
Iliyas Kyrykbayev 5cb347a44b TelecomKz Analytics Mapper: audit fixes, real event keys, 32 mapped screens
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>
2026-08-24 18:16:46 +05:00

120 lines
4.1 KiB
Python

"""
Live WebView -> app map sync.
Re-reads the DOM of whatever the phone is currently showing and refreshes the
hotspots of one screen, without re-taking the screenshot. Cheap enough to poll.
Two things it deliberately does NOT do:
* it never touches a screen other than the one it was asked for (the previous
version ignored argv and rewrote main_dashboard on every call);
* it never rewrites the screen's totalHeight to the DOM document height. The
hotspot grid belongs to the stored screenshot; if the page has grown taller
than the last capture, that is reported so the UI can ask for a re-capture
instead of silently producing coordinates that point off the image.
"""
import argparse
import json
import sys
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import telecom_cdp as T
from capture_screen import DOM_ELEMENTS_JS, build_hotspots
def sync_screen(screen_id):
T.validate_screen_id(screen_id)
device, target = T.connect()
layout = T.probe_native_layout(device)
catalog = T.load_metrics_catalog()
source = T.metrics_source()
with T.CdpSession(target["webSocketDebuggerUrl"]) as session:
page = json.loads(
session.evaluate(
"JSON.stringify({route: location.pathname, innerWidth: window.innerWidth,"
" innerHeight: window.innerHeight, scrollHeight: Math.max("
"document.documentElement.scrollHeight, document.body ? document.body.scrollHeight : 0)})"
)
)
dom = json.loads(session.evaluate("JSON.stringify(" + DOM_ELEMENTS_JS + ")"))
app_map = T.load_app_map()
existing = next((s for s in app_map.get("screens", []) if s["id"] == screen_id), None)
scale = layout["screenWidth"] / max(page["innerWidth"], 1)
wv_top = layout["webviewTop"]
band_h = layout["webviewBottom"] - wv_top
nav_h = layout["screenHeight"] - layout["bottomNavTop"]
# Trust the stored screenshot's geometry when we have one; the overlay is drawn
# on that image, so its height is the authority.
if existing and existing.get("totalHeight"):
total_h = existing["totalHeight"]
content_h = total_h - wv_top - nav_h
else:
content_h = max(round(page["scrollHeight"] * scale), band_h)
total_h = wv_top + content_h + nav_h
geo = {
"scale": scale,
"webviewTop": wv_top,
"contentHeight": content_h,
"navHeight": nav_h,
"totalHeight": total_h,
}
hotspots = build_hotspots(dom.get("elements", []), layout, geo, catalog, source)
expected_content_h = max(round(page["scrollHeight"] * scale), band_h)
stale = abs(expected_content_h - content_h) > 24
screen = dict(existing) if existing else {
"id": screen_id,
"name": "Экран " + screen_id,
"category": "Основное",
"image": "/assets/screens/" + screen_id + "_long.png",
"viewportWidth": layout["screenWidth"],
"viewportHeight": layout["screenHeight"],
"totalHeight": total_h,
"isScrollable": total_h > layout["screenHeight"],
}
screen["hotspots"] = hotspots
screen["route"] = page.get("route")
screen["syncedAt"] = int(time.time() * 1000)
screen["screenshotStale"] = stale
T.upsert_screen(app_map, screen)
app_map["metricsSource"] = source
T.save_app_map(app_map)
return {
"success": True,
"screenId": screen_id,
"route": page.get("route"),
"hotspotCount": len(hotspots),
"screenshotStale": stale,
"metricsSource": source,
}
def main():
parser = argparse.ArgumentParser(description="Sync one screen's hotspots from the live WebView.")
parser.add_argument("screen_id", nargs="?", default="main_dashboard")
args = parser.parse_args()
try:
result = sync_screen(args.screen_id)
except (T.DeviceError, ValueError) as exc:
print(json.dumps({"success": False, "error": str(exc)}, ensure_ascii=False))
return 1
print(json.dumps(result, ensure_ascii=False))
return 0
if __name__ == "__main__":
sys.exit(main())