pi-skill-hub/skills/sap-timesheet/scripts/bridge_helper.py

#!/usr/bin/env python3
"""Helper SAP timesheet via bridge BiDi (127.0.0.1:8787, token firefox-bidi-local)."""
import json, sys, time, urllib.request

BRIDGE = "http://127.0.0.1:8787"
TOKEN = "firefox-bidi-local"
HDR = {"Authorization": "Bearer " + TOKEN, "Content-Type": "application/json"}

def post(path, payload):
    req = urllib.request.Request(BRIDGE + path, data=json.dumps(payload).encode(), headers=HDR, method="POST")
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.loads(r.read().decode())

def ev(expr):
    return post("/eval", {"expression": expr, "awaitPromise": True})

def cmd(method, params):
    return post("/cmd", {"method": method, "params": params})

def click(x, y, button=0):
    return cmd("input.performActions", {"context": _ctx(), "actions": [{"id": "m", "type": "pointer", "parameters": {"pointerType": "mouse"}, "actions": [
        {"type": "pointerMove", "x": x, "y": y, "origin": "viewport"},
        {"type": "pointerDown", "button": button}, {"type": "pause", "duration": 90},
        {"type": "pointerUp", "button": button}, {"type": "pause", "duration": 450}]}]})

_CTX = None
def _ctx():
    global _CTX
    if _CTX is None:
        _CTX = post("/status", {})["context"] if False else None
        # /status è GET
    return _CTX

def status():
    req = urllib.request.Request(BRIDGE + "/status", headers=HDR)
    with urllib.request.urlopen(req, timeout=10) as r:
        return json.loads(r.read().decode())

def ctx():
    global _CTX
    if _CTX is None:
        _CTX = status()["context"]
    return _CTX

def click_ctx(x, y, button=0):
    return cmd("input.performActions", {"context": ctx(), "actions": [{"id": "m", "type": "pointer", "parameters": {"pointerType": "mouse"}, "actions": [
        {"type": "pointerMove", "x": x, "y": y, "origin": "viewport"},
        {"type": "pointerDown", "button": button}, {"type": "pause", "duration": 90},
        {"type": "pointerUp", "button": button}, {"type": "pause", "duration": 450}]}]})

CTRL = "\ue009"; ENTER = "\ue007"; F2 = "\ue002"

def type_text(text, commit=ENTER):
    acts = [{"type": "keyDown", "value": CTRL}, {"type": "keyDown", "value": "a"}, {"type": "keyUp", "value": "a"}, {"type": "keyUp", "value": CTRL}, {"type": "pause", "duration": 200}]
    for ch in text:
        acts += [{"type": "keyDown", "value": ch}, {"type": "keyUp", "value": ch}, {"type": "pause", "duration": 110}]
    acts += [{"type": "pause", "duration": 250}, {"type": "keyDown", "value": commit}, {"type": "keyUp", "value": commit}, {"type": "pause", "duration": 400}]
    return cmd("input.performActions", {"context": ctx(), "actions": [{"id": "k", "type": "key", "actions": acts}]})

def cell_center(r, c):
    out = ev(f"(() => {{ const e = document.getElementById('M0:46:2:1[{r},{c}]'); if (!e) return JSON.stringify({{err:'nofound'}}); e.scrollIntoView({{block:'center', inline:'center'}}); return 'ok'; }})()")
    time.sleep(0.6)
    out = ev(f"(() => {{ const e = document.getElementById('M0:46:2:1[{r},{c}]'); const b = e.getBoundingClientRect(); return JSON.stringify({{x: Math.round(b.x + b.width/2), y: Math.round(b.y + b.height/2)}}); }})()")
    v = out.get("result", out)
    if isinstance(v, str):
        v = json.loads(v)
    return v

def dialogs():
    out = ev("""(() => { const d = [...document.querySelectorAll('[role="dialog"], .urDlg, [class*="Dialog"], [class*="dialog"]')].filter(e => e.getBoundingClientRect().width > 0).map(e => (e.textContent||'').trim().slice(0,120)); return JSON.stringify(d); })()""")
    v = out.get("result", "")
    return json.loads(v) if isinstance(v, str) else v

def cell_text(r, c):
    out = ev(f"(() => {{ const e = document.getElementById('M0:46:2:1[{r},{c}]'); return e ? (e.textContent||'').trim() : null; }})()")
    v = out.get("result")
    return v

def write_cell(r, c, text, expected=None):
    """Scrive text nella cella [r,c]: click reale, Ctrl+A, digitazione, Invio. Ritorna dict esito."""
    p = cell_center(r, c)
    if "err" in p:
        return {"error": p["err"]}
    click_ctx(p["x"], p["y"])
    time.sleep(0.2)
    type_text(text)
    time.sleep(0.5)
    val = cell_text(r, c)
    exp = expected if expected is not None else text
    if val != exp:
        # retry con F2 come commit
        type_text(text, commit=F2)
        time.sleep(0.5)
        val = cell_text(r, c)
    return {"cell": f"[{r},{c}]", "sent": text, "read": val, "ok": val == exp, "dialogs": dialogs()}

if __name__ == "__main__":
    what = sys.argv[1]
    if what == "grid":
        cols = [int(x) for x in sys.argv[2].split(",")] if len(sys.argv) > 2 else list(range(2, 41))
        rows = [int(x) for x in sys.argv[3].split(",")] if len(sys.argv) > 3 else [1, 2, 3]
        res = {}
        for r in rows:
            res[r] = {c: cell_text(r, c) for c in cols}
        print(json.dumps(res, ensure_ascii=False))
    elif what == "dialogs":
        print(json.dumps(dialogs(), ensure_ascii=False))
    elif what == "write":
        r, c, text = int(sys.argv[2]), int(sys.argv[3]), sys.argv[4]
        print(json.dumps(write_cell(r, c, text), ensure_ascii=False))
    elif what == "center":
        r, c = int(sys.argv[2]), int(sys.argv[3])
        print(json.dumps(cell_center(r, c)))
    elif what == "click":
        x, y = int(sys.argv[2]), int(sys.argv[3])
        b = int(sys.argv[4]) if len(sys.argv) > 4 else 0
        click_ctx(x, y, b)
        print("clicked", x, y, "btn", b)