synth/generators/playwright/web_table_extract/__init__.py

"""`web_table_extract` generator (REAL WEB, aligned to playwright/eval_web).

Mirrors the real ``eval_web/extraction_table`` task **on its own site**: the model
opens ``https://eval-web.mcpmark.ai/extraction`` (the exact page the benchmark
uses), reads its table (Title / Rating / Likes / Views / Replies), and reports a
value. We keep the *site and observation distribution identical* to the eval set
and only **vary the question** (换问法), per the project decision to reuse existing
sites rather than invent new ones.

Why not Wikipedia: an earlier version anchored to Wikipedia "List of…" pages, but
(a) that changed the site away from the real eval_web distribution and (b) those
pages carry hundreds of images that blow past the browser navigate timeout through
a single SOCKS proxy. The eval-web page is ~32 KB and loads in seconds.

Ground truth: the page's 97 rows are fixed (verified stable across requests) and
are already committed to the repo as the original task's ``data.csv``. We read that
canonical table at build time, compute the answer for a randomly chosen analytical
question deterministically, and freeze it to ``content.txt`` — no scraping, no LLM
for the answer. The oracle ``solve()`` copies that snapshot, proving the verifier
accepts the intended value and rejects a blank attempt.

Requires network egress (proxy in $SYNTH_PROXY) at *rollout* so the agent's browser
can reach eval-web; build itself only reads the local canonical CSV.
"""
import csv
import os
import re
from pathlib import Path

from ..base import Generator, _render_verify, _write, diversify_question

# The real eval_web extraction page — identical to the benchmark task.
EXTRACTION_URL = "https://eval-web.mcpmark.ai/extraction"

# Canonical 97-row table, committed as the original task's ground truth.
_REPO_ROOT = Path(__file__).resolve().parents[4]
DATA_CSV = _REPO_ROOT / "tasks/playwright/standard/eval_web/extraction_table/data.csv"


def _clean(s):
    return " ".join(re.sub(r"\[[^\]]*\]", "", str(s)).split()).strip()


def _num(s):
    """Parse a numeric cell (strip quotes/commas/%). None if not numeric."""
    t = str(s).replace(",", "").replace("%", "").replace('"', "").strip()
    try:
        return float(t)
    except ValueError:
        return None


_TABLE = None


def _load_table():
    """Return (header, rows) from the canonical data.csv (cached)."""
    global _TABLE
    if _TABLE is not None:
        return _TABLE
    with open(DATA_CSV, newline="", encoding="utf-8") as f:
        reader = csv.reader(f, skipinitialspace=True)
        all_rows = [[_clean(c) for c in r] for r in reader if any(c.strip() for c in r)]
    header, rows = all_rows[0], all_rows[1:]
    _TABLE = (header, rows)
    return _TABLE


def _pick_question(header, rows, rng):
    """Choose an analytical question; return (question_text, answer, must_include)."""
    ncol = len(header)
    numeric_cols = []
    for c in range(ncol):
        vals = [_num(r[c]) for r in rows]
        if sum(v is not None for v in vals) >= max(5, int(0.8 * len(rows))):
            numeric_cols.append(c)
    text_cols = [c for c in range(ncol) if c not in numeric_cols]

    kinds = []
    if numeric_cols and text_cols:
        kinds += ["max_by", "min_by"]
    if text_cols and ncol >= 2:
        kinds.append("lookup")
    if numeric_cols:
        kinds.append("count_ge")
    kind = rng.choice(kinds) if kinds else "lookup"

    if kind in ("max_by", "min_by"):
        nc = rng.choice(numeric_cols)
        ac = rng.choice([c for c in range(ncol) if c != nc])
        keyed = [(_num(r[nc]), r) for r in rows if _num(r[nc]) is not None]
        want = max(keyed, key=lambda x: x[0]) if kind == "max_by" else min(keyed, key=lambda x: x[0])
        ties = [r for v, r in keyed if v == want[0]]
        if len(ties) != 1:  # need an unambiguous extremum
            return _pick_question(header, rows, rng)
        sup = "highest" if kind == "max_by" else "lowest"
        q = (f"On the page, find the row with the {sup} “{header[nc]}”, "
             f"and report its “{header[ac]}”.")
        return q, want[1][ac], [header[nc], header[ac]]

    if kind == "lookup":
        kc = rng.choice(text_cols)
        ac = rng.choice([c for c in range(ncol) if c != kc])
        order = list(rows)
        rng.shuffle(order)
        for r in order:
            if sum(1 for x in rows if _clean(x[kc]) == _clean(r[kc])) == 1 and r[kc] and r[ac]:
                q = (f"On the page, find the row where “{header[kc]}” is "
                     f"“{r[kc]}”, and report its “{header[ac]}”.")
                return q, r[ac], [header[kc], r[kc], header[ac]]
        return _pick_question(header, rows, rng)

    # count_ge
    nc = rng.choice(numeric_cols)
    vals = sorted(v for v in (_num(r[nc]) for r in rows) if v is not None)
    thr = vals[len(vals) // 2]  # median threshold
    cnt = sum(1 for v in vals if v >= thr)
    q = (f"On the page, how many rows have “{header[nc]}” greater than or "
         f"equal to {thr:g}? Report a single integer.")
    return q, str(cnt), [header[nc], f"{thr:g}"]


class WebTableExtract(Generator):
    KEY = "web_table_extract"
    CATEGORY_NAME = "Web Table Extract"
    DIFFICULTY = "L3"  # real-web navigation + read full table + compute
    TAGS = ["eval web", "real web", "table extraction"]
    NEEDS_NET = True

    def build(self, env_dir, llm, rng):
        header, rows = _load_table()
        question, answer, must = _pick_question(header, rows, rng)
        if not _clean(answer):
            raise RuntimeError("empty answer computed")
        # Paraphrase for variety; keep column names/conditions, never leak answer.
        question = diversify_question(llm, question, must_include=must,
                                      forbid=(_clean(answer),))
        _write(env_dir / "content.txt", _clean(answer))
        return {"url": EXTRACTION_URL, "question": question}

    def description(self, spec):
        return (
            "Please use Playwright MCP tools to finish the following task:\n\n"
            "### Task: Extract a value from a web table\n\n"
            f"Open this page in the browser: {spec['url']}\n\n"
            "Wait for the page to fully load (all rows). Then:\n\n"
            f"{spec['question']}\n\n"
            "Your final reply must contain ONLY that value — no preamble, labels, "
            "units, or extra words."
        )

    def verify_src(self, spec):
        body = """
C = json.loads(__CONSTS__)

def main():
    expected = read_page("content.txt").strip()
    sub = get_submitted_answer()
    if not sub:
        fail("no answer found (no chat reply / answer.txt)")
    ne, ns = _norm(expected), _norm(sub)
    last = ""
    for ln in reversed(sub.splitlines()):
        if ln.strip():
            last = _norm(ln); break
    multiword = len(ne.split()) >= 2
    if ne == ns or ne == last or (multiword and ne in ns):
        ok(f"correct answer: {expected}"); print("\\U0001f389 All checks passed!"); sys.exit(0)
    fail(f"expected {expected!r}, got {ns[:200]!r}")

if __name__ == "__main__":
    main()
"""
        return _render_verify(body, {})

    def solve(self, work_dir, spec):
        # The answer is frozen in content.txt at build time (computed from the
        # canonical table); copy it through to prove the verifier round-trips.
        src = work_dir / "content.txt"
        ans = src.read_text(encoding="utf-8").strip() if src.exists() else ""
        _write(work_dir / "answer.txt", ans + "\n")