Blame for src/miniwebgen/formvalues.py

"""Field values for generated form tasks (privileged, parameter stage).

The sampler picks WHICH form; this module proposes WHAT to enter: K diverse, realistic value
sets per form, written once by an LLM from the form's fields (labels, types, options, current
values) and a little site context, cached on disk. Sign-in forms take real accounts from the
site's users table instead. Every set is validated by the dry run (the site must accept it).
"""
from __future__ import annotations

import hashlib
import json
import re

from miniwebgen import config

CACHE = config.DATAGEN_DIR / "formvalues"
SYSTEM = ("You fill in web forms for realistic test tasks. Propose DIFFERENT value sets a real user of this "
          "site might submit. Use only values the form can accept: for fields with options pick one of the "
          "options exactly; dates as YYYY-MM-DD (bookings/deadlines in the future: 2026-10 to 2027-06), times "
          "as HH:MM, numbers plain; emails @example.com unless an existing person below fits. For an edit or "
          "settings form change 1–3 fields from their current values and repeat the others unchanged. Never "
          "invent other people when the context lists who exists. Reply ONLY JSON.")


def field_key(f):
    return f.get("name") or f["css"]


def _people(site, limit=15):
    """A few accounts of the site (privileged, base data; users tables are small)."""
    try:
        from miniweb.app import db
        rows = db.query(site, "users", limit=limit)
    except Exception:
        return []
    keep = ("username", "name", "display_name", "full_name", "email", "handle")
    return [{k: r.get(k) for k in keep if r.get(k)} for r in rows if isinstance(r, dict)]


def records(site, tables=10, per=25):
    """A few real records of the site (id + title/name per table; base data, bounded queries) so
    values that must exist — a meeting code, an order number — can be chosen, not invented."""
    try:
        from miniweb.app import db
        prefix = site.replace("-", "_") + "_"
        names = [r["name"] for r in db.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name LIKE ? ORDER BY name LIMIT ?",
            (prefix + "%", tables * 3))]
    except Exception:
        return []
    out = []
    for t in names:
        if t.endswith("_users") or len(out) >= tables * per:
            continue
        try:
            cols = [c["name"] for c in db.execute(f"PRAGMA table_info({t})")]
            label = next((c for c in ("title", "name", "subject", "code", "number", "label") if c in cols), None)
            if "id" not in cols:
                continue
            rows = db.execute(f"SELECT id{', ' + label if label else ''} FROM {t} ORDER BY id LIMIT ?", (per,))
        except Exception:
            continue
        out += [{"collection": t[len(prefix):], "id": r["id"], **({label: r[label]} if label else {})} for r in rows]
    return out


def accounts(site, limit=20):
    """Sign-in credentials from the site's users table: [{login, email, password, name}]."""
    try:
        from miniweb.app import db
        rows = db.query(site, "users", limit=limit)
    except Exception:
        return []
    out = []
    for r in rows or []:
        pw = r.get("password") or r.get("pass") or r.get("pwd")
        if not pw or not isinstance(pw, str) or len(pw) > 40 or pw.startswith(("$2", "pbkdf2", "sha")):
            continue                           # hashed: the task could not state it
        login = r.get("username") or r.get("user") or r.get("handle") or r.get("email")
        if login:
            out.append({"login": str(login), "email": str(r.get("email") or ""), "password": pw,
                        "name": str(r.get("name") or r.get("display_name") or r.get("full_name") or login)})
    return out


def _key(ctrl):
    ident = json.dumps([ctrl.get("site_id"), ctrl.get("page"), ctrl.get("role"),
                        [(f.get("name"), f.get("type")) for f in ctrl.get("fields", [])]], sort_keys=True)
    return hashlib.sha1(ident.encode()).hexdigest()[:16]


K = 12      # value sets per form (a cache written with fewer is topped up with new, different ones)


def value_sets(ctrl, site, k=K, model=None, force=False, grounded=False):
    """[{values: {field_key: value}, summary}] for a form control (cached). `grounded`: also show the
    site's own records so values that must exist are picked from them."""
    path = CACHE / site / f"{_key(ctrl)}.json"
    have = [] if force or not path.exists() else json.loads(path.read_text())
    if have and (len(have) >= k or ctrl.get("_topped_up") == k):
        return have
    out, seen = list(have), {json.dumps(s["values"], sort_keys=True) for s in have}
    for _ in range(8):                                   # batches: one call returns ~a dozen sets at most
        if len(out) >= k:
            break
        more = _generate(ctrl, site, min(15, k - len(out)), model, grounded or ctrl.get("grounded"), avoid=out)
        fresh = [s for s in more if json.dumps(s["values"], sort_keys=True) not in seen]
        if not fresh:
            break
        seen |= {json.dumps(s["values"], sort_keys=True) for s in fresh}
        out += fresh
    ctrl["_topped_up"] = k
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(out, indent=1, ensure_ascii=False))
    return out


def _generate(ctrl, site, k, model=None, grounded=False, avoid=()):
    if k <= 0:
        return []
    from miniweb.annotation import macros as registry
    from helpers.llm import call_llm
    fields = [{"key": field_key(f), "label": f.get("label") or f.get("placeholder") or f.get("name"), "type": f["type"],
               "current": f.get("value") or None, "required": f.get("required", False),
               **({"options": [o["text"] for o in f.get("options", [])][:30]} if f.get("options") else {}),
               **({"min": f.get("min"), "max": f.get("max")} if f.get("min") or f.get("max") else {})}
              for f in ctrl["fields"]]
    prompt = json.dumps({"site": ctrl.get("site_name") or site, "page": ctrl.get("page_title"), "form": ctrl.get("heading"),
                         "purpose": registry.describe(ctrl["macro"]).get("description"), "submit_button": ctrl["submit"].get("text"),
                         "fields": fields, "people_on_this_site": _people(site), "page_context": ctrl.get("context", [])[:25],
                         **({"records_on_this_site": records(site),
                             "note": "a value that refers to something (a code, an id, a number) must be one of these"}
                            if grounded else {}),
                         **({"already_used_do_not_repeat": [x["values"] for x in avoid][-40:]} if avoid else {}),
                         "output": {"sets": [{"values": {"<field key>": "<value>"},
                                              "summary": "one short line: what this submission does"}]},
                         "n_sets": k}, ensure_ascii=False)
    raw = call_llm(prompt, system=SYSTEM, json_mode=True, model=model or config.MODEL_SUGGEST, max_tokens=6000,
                   temperature=0.9)
    try:
        sets = json.loads(raw or "").get("sets") or []
    except (ValueError, AttributeError):
        sets = []
    keys = {field_key(f) for f in ctrl["fields"]}
    clean = []
    for s in sets:
        vals = {k2: str(v) for k2, v in (s.get("values") or {}).items() if k2 in keys and v not in (None, "")}
        if vals:
            clean.append({"values": vals, "summary": str(s.get("summary") or "")[:200]})
    return clean


def option_for(field, value):
    """The option of a select/radio field matching a value (by text, then value)."""
    v = re.sub(r"\s+", " ", str(value)).strip().casefold()
    for o in field.get("options") or []:
        if str(o.get("text", "")).strip().casefold() == v:
            return o
    for o in field.get("options") or []:
        if str(o.get("value", "")).strip().casefold() == v:
            return o
    return None