telecomkz_scraper/tools/legacy/stitch_opencv.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

135 lines
4.7 KiB
Python

"""
TelecomKz Pixel-Perfect OpenCV Long Screenshot Stitcher
Uses Template Matching to find exact subpixel scroll offsets and preserves fixed top & bottom navigation bars.
"""
import sys
import subprocess
import time
import io
import json
from pathlib import Path
import numpy as np
import cv2
from PIL import Image
ADB_PATHS = [
r"C:\Users\user\AppData\Local\Android\Sdk\platform-tools\adb.exe",
"adb"
]
def find_adb():
for p in ADB_PATHS:
try:
res = subprocess.run([p, "version"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
if res.returncode == 0:
return p
except Exception:
continue
return "adb"
ADB_BIN = find_adb()
def get_device():
res = subprocess.run([ADB_BIN, "devices"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
lines = res.stdout.strip().split("\n")[1:]
for line in lines:
parts = line.strip().split("\t")
if len(parts) >= 2 and parts[1] == "device":
return parts[0]
return None
def capture_frame(dev):
cmd = [ADB_BIN]
if dev:
cmd.extend(["-s", dev])
cmd.extend(["exec-out", "screencap", "-p"])
res = subprocess.run(cmd, stdout=subprocess.PIPE)
if res.returncode == 0 and len(res.stdout) > 1000:
arr = np.frombuffer(res.stdout, dtype=np.uint8)
return cv2.imdecode(arr, cv2.IMREAD_UNCHANGED)
return None
def stitch_perfect_long_screen(screen_id: str = "main_dashboard", scroll_steps: int = 3):
dev = get_device()
if not dev:
print(json.dumps({"success": False, "error": "No Android device connected."}))
return False
print(f"Connecting to Android device {dev} for pixel-perfect stitching...")
# 1. Capture Frame 1 (at top of page)
f1 = capture_frame(dev)
if f1 is None:
print("Failed to capture Frame 1")
return False
h, w = f1.shape[:2] # 2340 x 1080
top_bar_h = 255
bottom_nav_y = 2140
bottom_nav_h = h - bottom_nav_y # 200px
top_bar = f1[0:top_bar_h, 0:w]
bottom_nav = f1[bottom_nav_y:h, 0:w]
# Current stitched scrollable canvas starts with Frame 1 middle content
scrollable_canvas = f1[top_bar_h:bottom_nav_y, 0:w]
# Frames capture loop with template matching
swipes_done = 0
for step in range(scroll_steps):
# Swipe inside the scrollable window
subprocess.run([ADB_BIN, "-s", dev, "shell", "input", "swipe", "540", "1700", "540", "900", "350"])
time.sleep(1.0)
swipes_done += 1
f_next = capture_frame(dev)
if f_next is None:
break
f_next_scroll = f_next[top_bar_h:bottom_nav_y, 0:w]
# Take a 150px template strip from the bottom of current canvas (excluding right edge scrollbar)
template_h = 150
template_w = w - 80 # ignore right scrollbar zone
template = scrollable_canvas[-template_h:, 20:template_w]
# Match template in f_next_scroll
res = cv2.matchTemplate(f_next_scroll[:, 20:template_w], template, cv2.TM_CCOEFF_NORMED)
min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(res)
print(f"Step {step+1}: Match confidence = {max_val:.3f}, matched_y = {max_loc[1]}")
if max_val > 0.70:
match_y = max_loc[1]
# New unique content is from match_y + template_h to bottom
append_y = match_y + template_h
if append_y < f_next_scroll.shape[0]:
new_slice = f_next_scroll[append_y:, 0:w]
scrollable_canvas = np.vstack([scrollable_canvas, new_slice])
else:
print(f"Low match confidence ({max_val:.2f}), using fixed fallback slice")
new_slice = f_next_scroll[800:, 0:w]
scrollable_canvas = np.vstack([scrollable_canvas, new_slice])
# Assemble master image: [Top Bar] + [Continuous Scrollable Canvas] + [Bottom Navigation Bar]
master_image = np.vstack([top_bar, scrollable_canvas, bottom_nav])
# Save image
screens_dir = Path("public/assets/screens")
screens_dir.mkdir(parents=True, exist_ok=True)
out_path = screens_dir / f"{screen_id}_long.png"
cv2.imwrite(str(out_path), master_image)
print(f"Flawless Stitched Image saved to: {out_path} (Resolution: {master_image.shape[1]}x{master_image.shape[0]})")
# Scroll back up to initial position
for _ in range(swipes_done):
subprocess.run([ADB_BIN, "-s", dev, "shell", "input", "swipe", "540", "900", "540", "1700", "250"])
time.sleep(0.3)
return master_image.shape[1], master_image.shape[0]
if __name__ == "__main__":
stitch_perfect_long_screen("main_dashboard", scroll_steps=2)