telecomkz_scraper/tools/reclassify_hotspots.py

"""
Re-apply the event-key rules and the metrics catalog to an already-captured app map.

Useful when EVENT_RULES changes or after a fresh ClickHouse pull: it fixes keys and
metrics without going near the phone. Hotspots whose key was observed on the wire
(keyConfidence "observed") or set by hand ("manual") are never touched.

    py tools/reclassify_hotspots.py             # all screens
    py tools/reclassify_hotspots.py --dry-run
    py tools/reclassify_hotspots.py --screen main_dashboard
"""

import argparse
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

import telecom_cdp as T
from capture_screen import classify

PROTECTED = {"observed", "manual", "dom-attribute"}


def reclassify(screen_filter=None, dry_run=False):
    app_map = T.load_app_map()
    catalog = T.load_metrics_catalog()
    source = T.metrics_source()

    changed_keys = 0
    rechecked = 0

    for screen in app_map.get("screens", []):
        if screen_filter and screen["id"] != screen_filter:
            continue
        for hs in screen.get("hotspots", []):
            rechecked += 1
            if hs.get("keyConfidence") in PROTECTED:
                T.attach_metrics(hs, catalog, source)
                continue
            if hs.get("source") == "native-uiautomator":
                T.attach_metrics(hs, catalog, source)
                continue
            if "source" not in hs:
                # Legacy row from before provenance was tracked. Its key is of unknown
                # origin, so flag it for review rather than overwrite it with another
                # caption-derived guess - several would collapse onto the same key.
                hs.setdefault("keyConfidence", "guessed")
                hs.setdefault("source", "legacy")
                T.attach_metrics(hs, catalog, source)
                continue

            key, name_ru, target, confidence = classify(hs.get("label") or "")
            if key != hs.get("eventKey"):
                print(
                    "  "
                    + screen["id"]
                    + ": "
                    + str(hs.get("eventKey"))
                    + " -> "
                    + key
                    + "   ("
                    + (hs.get("label") or "")[:45]
                    + ")"
                )
                changed_keys += 1
                if not dry_run:
                    hs["eventKey"] = key
                    hs["eventNameRu"] = name_ru
                    hs["keyConfidence"] = confidence
                    if target and not hs.get("targetScreenId"):
                        hs["targetScreenId"] = target
            if not dry_run:
                T.attach_metrics(hs, catalog, source)

    # Two hotspots on one screen must not share an event key, and a targetScreenId
    # pointing at a screen that was never captured is a dead link in the simulator.
    from capture_screen import slugify_event_key, unique_key

    known_ids = {sc["id"] for sc in app_map.get("screens", [])}
    deduped = dangling = 0
    for screen in app_map.get("screens", []):
        if screen_filter and screen["id"] != screen_filter:
            continue
        used = set()
        for hs in screen.get("hotspots", []):
            target = hs.get("targetScreenId")
            if target and target not in known_ids:
                print("  " + screen["id"] + ": dropping dead link -> " + target)
                hs.pop("targetScreenId", None)
                dangling += 1

            original = hs.get("eventKey") or ""
            if hs.get("keyConfidence") == "observed":
                used.add(original)  # never rename a verified key
                continue
            fixed = unique_key(original or slugify_event_key(hs.get("label") or ""), used)
            if fixed != original:
                print("  " + screen["id"] + ": duplicate " + original + " -> " + fixed)
                hs["eventKey"] = fixed
                deduped += 1

    print("\nDeduplicated keys: " + str(deduped) + ", dead links removed: " + str(dangling))

    if not dry_run:
        app_map["metricsSource"] = source
        T.save_app_map(app_map)

    measured = sum(
        1
        for s in app_map.get("screens", [])
        for h in s.get("hotspots", [])
        if h.get("metrics")
    )
    print(
        "\nHotspots checked: "
        + str(rechecked)
        + ", keys changed: "
        + str(changed_keys)
        + ", with metrics: "
        + str(measured)
        + " (source: "
        + source
        + ")"
    )
    if dry_run:
        print("Dry run - nothing was written.")


def main():
    parser = argparse.ArgumentParser(description="Re-apply event rules and metrics to the app map.")
    parser.add_argument("--screen", default=None)
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()
    reclassify(args.screen, args.dry_run)
    return 0


if __name__ == "__main__":
    sys.exit(main())