File size: 2,447 Bytes
adf912b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 | """Turn click cases into realistic DONE cases by actually executing the click in the browser.
python finetune/make_done_cases.py out/pages.jsonl out/cases.jsonl out/done_cases.jsonl [max=250]
For a click case (page A, goal, gold element): open A, click the gold element, observe the landing page B.
If the page changed, emit {goal, gold_op: DONE, page_obj: B, history: [that click]} with the same goal phrasing.
"""
import json, os, random, sys, time
sys.path.insert(0, "/home/ckl/projects/S/jev-ultrafast")
os.environ.setdefault("BU_CDP_URL", "http://127.0.0.1:9222")
from jev_ultrafast.browser import Browser
def main():
pages_f, cases_f, out = sys.argv[1:4]; mx = int(sys.argv[4]) if len(sys.argv) > 4 else 250
pages = [json.loads(l) for l in open(pages_f)]
cases = [c for c in (json.loads(l) for l in open(cases_f)) if c["kind"] == "click"]
rng = random.Random(3); rng.shuffle(cases)
n_ok = n_try = 0
with open(out, "w") as f:
for c in cases:
if n_ok >= mx: break
src = pages[c["page"]]; n_try += 1
try:
b = Browser(src["url"])
try:
page = b.observe(screenshot=False)
act = next((a for a in page["actions"] if a["kind"] == "click" and a["label"] == c["label"]), None)
if act is None:
continue
b.act(act, page); time.sleep(0.3)
dest = b.observe(screenshot=False)
finally:
b.close()
except Exception as e:
print("fail", type(e).__name__, str(e)[:60], flush=True); continue
if dest["url"] == page["url"] and dest["title"] == page["title"]:
continue
hist = [{"action": act["label"], "kind": "click", "text": None, "page_changed": True}]
f.write(json.dumps({"page": c["page"], "url": dest["url"], "title": dest["title"], "goal": c["goal"], "gold_op": "DONE",
"gold_id": "DONE", "kind": "done", "label": act["label"], "history": hist,
"page_obj": {k: dest[k] for k in ("url", "title", "text", "actions")}}, ensure_ascii=False) + "\n"); f.flush()
n_ok += 1
if n_ok % 25 == 0: print(f" {n_ok} done cases from {n_try} tries", flush=True)
print("wrote", n_ok, "DONE cases")
if __name__ == "__main__":
main()
|