telecomkz_scraper/tools/legacy/crawl_direct_taps.py

import asyncio
import json
import sys
import time
import subprocess
import requests
import websockets
from pathlib import Path
import numpy as np
import cv2

if sys.platform == "win32":
    try:
        sys.stdout.reconfigure(encoding='utf-8')
    except Exception:
        pass

ADB_BIN = r"C:\Users\user\AppData\Local\Android\Sdk\platform-tools\adb.exe"

def capture_screen_direct(name):
    res = subprocess.run([ADB_BIN, "exec-out", "screencap", "-p"], stdout=subprocess.PIPE)
    screens_dir = Path("public/assets/screens")
    screens_dir.mkdir(parents=True, exist_ok=True)
    with open(screens_dir / f"{name}.png", "wb") as f:
        f.write(res.stdout)
    with open(screens_dir / f"{name}_long.png", "wb") as f:
        f.write(res.stdout)
    return f"/assets/screens/{name}.png"

async def get_dom_buttons(ws_url):
    js = """
    (() => {
        const buttons = [];
        const scale = 1080 / (window.innerWidth || 384);
        const topBarOffset = 255;
        
        document.querySelectorAll('button, a, [role="button"], [class*="card"], [class*="item"], [class*="btn"]').forEach((el, i) => {
            const rect = el.getBoundingClientRect();
            const text = (el.innerText || el.getAttribute('aria-label') || '').trim().replace(/\\n/g, ' ');
            
            if (rect.width > 20 && rect.height > 15 && text.length > 0 && text.length < 50) {
                buttons.push({
                    text: text,
                    rect: {
                        x: Math.max(0, Math.round(rect.left * scale)),
                        y: Math.max(0, Math.round(topBarOffset + (rect.top + window.scrollY) * scale)),
                        width: Math.min(1080, Math.round(rect.width * scale)),
                        height: Math.round(rect.height * scale)
                    }
                });
            }
        });
        return {
            route: window.location.pathname,
            buttons: buttons
        };
    })()
    """
    async with websockets.connect(ws_url, max_size=50*1024*1024) as ws:
        await ws.send(json.dumps({"id": 1, "method": "Runtime.evaluate", "params": {"expression": js, "returnByValue": True}}))
        res = json.loads(await ws.recv())
        return res.get("result", {}).get("result", {}).get("value", {})

async def inspect_all_main_routes():
    subprocess.run([ADB_BIN, "forward", "tcp:9222", "localabstract:webview_devtools_remote_18732"])
    pages = requests.get("http://localhost:9222/json/list").json()
    ws_url = pages[0]["webSocketDebuggerUrl"]
    
    # Load current app map
    map_file = Path("public/data/telecomkz_app_map.json")
    with open(map_file, "r", encoding="utf-8") as f:
        app_map = json.load(f)
        
    screens_map = { s["id"]: s for s in app_map["screens"] }
    
    # Target screens to capture with direct ADB taps & DOM inspection
    # (tap_x, tap_y, screen_id, screen_name, category)
    destinations = [
        (160, 1280, "services_screen", "Экран «Мои услуги»", "Услуги"),
        (410, 1280, "traffic_screen", "Экран «Трафик»", "Услуги"),
        (670, 1280, "payments_screen", "Экран «Платежи»", "Финансы"),
        (920, 1280, "orders_screen", "Экран «Заявки»", "Услуги"),
        (160, 1490, "music_screen", "Экран «Музыка»", "Развлечения"),
        (410, 1490, "qr_screen", "Экран «QR Оплата»", "Финансы"),
        (670, 1490, "documents_screen", "Экран «Удостоверение»", "Профиль"),
        (920, 1490, "services_catalog", "Каталог сервисов", "Услуги"),
        (800, 940, "bonuses_screen", "Экран «Мои бонусы»", "Финансы")
    ]
    
    for tx, ty, sid, sname, scat in destinations:
        print(f"\n--- Navigating to {sname} (tap {tx},{ty}) ---")
        subprocess.run([ADB_BIN, "shell", "input", "tap", str(tx), str(ty)])
        time.sleep(2.0)
        
        img_url = capture_screen_direct(sid)
        dom_res = await get_dom_buttons(ws_url)
        buttons = dom_res.get("buttons", [])
        print(f"Captured {len(buttons)} buttons on {sname} (route: {dom_res.get('route')})")
        
        hotspots = []
        # Back button
        hotspots.append({
            "id": f"hs_back_{sid}",
            "label": "Назад на Главный",
            "rect": { "x": 40, "y": 110, "width": 140, "height": 140 },
            "eventKey": f"{sid.upper()}_BACK_CLICK",
            "eventNameRu": f"Назад из «{sname}»",
            "category": "navigation",
            "targetScreenId": "main_dashboard",
            "metrics": { "totalEvents": 4200, "uniqueUsers": 1800, "avgEventsPerUser": 2.3, "shareOfClicks": 28.0 }
        })
        
        for i, b in enumerate(buttons):
            text = b["text"]
            ek = "CLICK_" + "".join(c if c.isalnum() else "_" for c in text.upper())[:24]
            hotspots.append({
                "id": f"hs_{sid}_{i+1}",
                "label": text,
                "rect": b["rect"],
                "eventKey": ek,
                "eventNameRu": f"Нажатие «{text}»",
                "category": "action",
                "metrics": {
                    "totalEvents": 1500 + (i * 280),
                    "uniqueUsers": 600 + (i * 90),
                    "avgEventsPerUser": 2.5,
                    "shareOfClicks": 12.0
                }
            })
            
        screens_map[sid] = {
            "id": sid,
            "name": sname,
            "category": scat,
            "image": img_url,
            "isScrollable": True,
            "viewportWidth": 1080,
            "viewportHeight": 2340,
            "totalHeight": 2340,
            "hotspots": hotspots
        }
        
        # Tap back on phone
        subprocess.run([ADB_BIN, "shell", "input", "tap", "90", "180"])
        time.sleep(1.0)
        subprocess.run([ADB_BIN, "shell", "input", "keyevent", "4"])
        time.sleep(1.2)
        
    app_map["screens"] = list(screens_map.values())
    with open(map_file, "w", encoding="utf-8") as f:
        json.dump(app_map, f, ensure_ascii=False, indent=2)
        
    print("\n✅ All screens visited, captured, and written to telecomkz_app_map.json!")

if __name__ == "__main__":
    asyncio.run(inspect_all_main_routes())