Move a table

import csv, json, os, sys

rows = list(csv.DictReader(open(sys.argv[1], newline="", encoding="utf-8-sig")))
os.makedirs("clay-export", exist_ok=True)
columns = [{"id": name, "name": name} for name in (rows[0] if rows else [])]
json.dump({"data": columns}, open("clay-export/columns.json", "w"))
with open("clay-export/rows.jsonl", "w") as fh:
    for row in rows:
        cells = {k: {"status": "success", "value": v} for k, v in row.items() if v}
        fh.write(json.dumps({"cells": cells}) + "\n")
print(f"{len(rows)} rows")

import json, re, sys

mapping = json.load(open("clay-to-autumn.json"))
start, size = map(int, sys.argv[1:3]) if len(sys.argv) > 2 else (0, 25)

columns = json.load(open("clay-export/columns.json"))["data"]
names = {c["id"]: c["name"] for c in columns}
inputs = mapping["inputs"]
key = inputs[mapping["key"]]
fields = {c["field"]: c for c in mapping["research"].values()}
if set(fields) & set(inputs.values()):
    sys.exit("research fields must not reuse input names")
for name in mapping["rules"]:
    if not re.fullmatch(r"[a-z][a-z0-9_]*", name) or name.startswith("rule_"):
        sys.exit(f"rename rule {name!r} to snake_case without a rule_ prefix")

rows = []
for line in open("clay-export/rows.jsonl"):
    cells = json.loads(line)["cells"]
    row = {names[k]: v.get("value") for k, v in cells.items()
           if k in names and v.get("status") == "success"}
    rows.append({f: "" if row.get(col) is None else row[col]
                 for col, f in inputs.items()})
rows = rows[start:start + size]

task = {
    "name": f"Clay migration: {mapping['clay_table']} from row {start}",
    "prompt": (
        "Enrich every input row. Return exactly one output row per input row, "
        f"keeping each row's {key}. Fill each field only from evidence; "
        "leave it empty rather than guessing."
    ),
    "input_schema": {f: "str" for f in inputs.values()},
    "input_rows": rows,
    "output": {
        "id": "clay-migration",
        "kind": "research",
        "path": "outputs/clay-migration.jsonl",
        "schema": {f: {"type": c["type"], "description": c["description"]}
                   for f, c in fields.items()},
        "schema_order": list(fields),
        "target_count": len(rows),
    },
    "rules": mapping["rules"],
    "clarify": False,
}
json.dump(task, open("task.json", "w"), indent=2)
print(f"task.json: {len(rows)} rows")

import csv, json, os, sys, requests

AUTUMN = "https://api.autumn.ai"
HEADERS = {"X-API-Key": os.environ["AUTUMN_API_KEY"]}
task_id = sys.argv[1]
mapping = json.load(open("clay-to-autumn.json"))
key = mapping["inputs"][mapping["key"]]

def autumn_rows():
    params = {"limit": 500}
    while True:
        page = requests.get(f"{AUTUMN}/task/{task_id}/output",
                            headers=HEADERS, params=params).json()
        for row in page["rows"]:
            yield {k: v.get("value") for k, v in row.items()
                   if not k.startswith("_") and isinstance(v, dict)}
        if not page.get("_next_cursor"):
            return
        params["cursor"] = page["_next_cursor"]

def clay_rows():
    columns = json.load(open("clay-export/columns.json"))["data"]
    names = {c["id"]: c["name"] for c in columns}
    for line in open("clay-export/rows.jsonl"):
        cells = json.loads(line)["cells"]
        yield {names[k]: v.get("value") for k, v in cells.items()
               if k in names and v.get("status") == "success"}

def text(value):
    return "" if value is None else str(value).strip().lower()

def rid_of(row):
    return "" if row.get(key) is None else str(row[key])

def same(a, b, tolerance):
    try:
        return abs(float(a) - float(b)) <= tolerance * max(abs(float(a)), 1)
    except ValueError:
        return a == b

sent = {str(r[key]) for r in json.load(open("task.json"))["input_rows"]}
required = {f"rule_{name}" for name in mapping["rules"]}
rows = list(autumn_rows())
counts = {}
for row in rows:
    counts[rid_of(row)] = counts.get(rid_of(row), 0) + 1

autumn, held = {}, []
for row in rows:
    rid = rid_of(row)
    problems = [k for k, v in row.items() if k.startswith("rule_") and v != "match"]
    problems += [f"{k} missing" for k in required - set(row)]
    if rid not in sent:
        problems.append("unexpected key")
    elif counts[rid] > 1:
        problems.append("duplicate key")
    if problems:
        held.append({**row, "problems": problems})
    else:
        autumn[rid] = row
clay = {str(r.get(mapping["key"])): r for r in clay_rows()}

diffs = []
for clay_col, spec in mapping["research"].items():
    f, tol = spec["field"], spec.get("tolerance", 0)
    n_clay = n_autumn = both = agree = 0
    for rid, row in autumn.items():
        c = text(clay.get(rid, {}).get(clay_col))
        a = text(row.get(f))
        n_clay += bool(c)
        n_autumn += bool(a)
        if c and a:
            both += 1
            if same(c, a, tol):
                agree += 1
            else:
                diffs.append({key: rid, "column": f, "clay": c, "autumn": a})
    print(f"{f}: Clay filled {n_clay}/{len(autumn)}, "
          f"Autumn filled {n_autumn}/{len(autumn)}, agree {agree}/{both}")
returned = {rid_of(r) for r in rows}
print(f"held for review: {len(held)}, "
      f"sent but not returned: {len(sent - returned)}")

out = f"results/{task_id}"
os.makedirs(out, exist_ok=True)
with open(f"{out}/disagreements.csv", "w", newline="") as fh:
    writer = csv.DictWriter(fh, fieldnames=[key, "column", "clay", "autumn"])
    writer.writeheader()
    writer.writerows(diffs)
json.dump(list(autumn.values()), open(f"{out}/autumn_rows.json", "w"), indent=2)
json.dump(held, open(f"{out}/held_rows.json", "w"), indent=2)
print(f"saved to {out}/")