add serve/kev_convert.py

serve/kev_convert.py

@@ -0,0 +1,84 @@
+"""Kev / TypeSafe System One records -> candidate-v2 rows. Exact inverse of final/jev_fresh.request_for: converting a Kev
+record here and sending the row through request_for reproduces the record's state, instructions and criteria meanings.
+Kev record: {"state": ..., "questions": {qid: {"type", "instructions", "criteria", "label", "src", "target_distribution"?}}, "_meta": {...}}.
+Choice criteria may be a dict {name: description|None} or a list of names; choice questions with more than 20 options are
+capped to 20 (gold kept, others drawn deterministically from the record id) because candidate-v2 has 20 keys. Serving passes
+keys=WIDE_CHOICE_KEYS, so a request is never capped below its own option count."""
+import hashlib
+import json
+import random
+
+from system1_client import CHOICE_KEYS, validate_row
+
+NOUL_DEFAULT = {"true": "the answer is yes", "false": "the answer is no"}
+
+
+def _sha(text):
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+def _text(value):
+    return value if isinstance(value, str) else json.dumps(value, ensure_ascii=False)
+
+
+def _criteria_items(criteria):
+    if criteria is None:
+        return []
+    if isinstance(criteria, dict):
+        return [(k, v) for k, v in criteria.items()]
+    return [(c, None) for c in criteria]
+
+
+def kev_record_to_rows(record, source, source_revision, split, keys=CHOICE_KEYS):
+    meta = record.get("_meta") or {}
+    state = record["state"]
+    rid = meta.get("id") or _sha(_text(state))[:16]
+    gid = meta.get("group_id") or rid
+    rows = []
+    for qid, q in record["questions"].items():
+        kind = q["type"]
+        question = _text(q.get("instructions") or "")
+        row_meta = {"rubric_family": f"kev:{source}:{qid}", "novelty": "seen", "domain": source, "label_origin": "kev_public",
+                    "kev_src": q.get("src"), "kev_question_id": qid}
+        soft = q.get("target_distribution")
+        if kind == "noul":
+            crit = dict(_criteria_items(q.get("criteria")))
+            crit = {k: v for k, v in crit.items() if isinstance(v, str) and v}
+            options = {"Y": f"true: {crit.get('true', NOUL_DEFAULT['true'])}", "N": f"false: {crit.get('false', NOUL_DEFAULT['false'])}"}
+            label = q["label"]
+            target = "Y" if (label is True or label in ("true", "yes", 1)) else "N"
+            dist = {"Y": float(soft.get("true", 0.0)), "N": float(soft.get("false", 0.0))} if isinstance(soft, dict) else None
+        elif kind == "choice":
+            items = _criteria_items(q["criteria"])
+            names = [n for n, _ in items]
+            gold = q["label"]
+            if gold not in names:
+                raise ValueError(f"{rid}/{qid}: label {gold!r} not among criteria")
+            if len(items) > len(keys):
+                rng = random.Random(_sha(f"{rid}:{qid}:cap"))
+                others = [it for it in items if it[0] != gold]
+                rng.shuffle(others)
+                keep = {gold} | {n for n, _ in others[: len(keys) - 1]}
+                items = [it for it in items if it[0] in keep]  # original relative order, gold kept
+                row_meta["capped_from"] = len(names)
+            ks = keys[: len(items)]
+            options = {k: (f"{n}: {_text(d)}" if d not in (None, "") else n) for k, (n, d) in zip(ks, items)}
+            row_meta["candidate_ids"] = {k: n for k, (n, _) in zip(ks, items)}
+            target = ks[[n for n, _ in items].index(gold)]
+            dist = {k: float(soft.get(n, 0.0)) for k, (n, _) in zip(ks, items)} if isinstance(soft, dict) else None
+        elif kind == "score":
+            levels = [_text(c) for c in q["criteria"]]
+            options = {str(i): f"{i}: {lvl}" for i, lvl in enumerate(levels)}
+            target = str(int(q["label"]))
+            dist = {str(i): float(soft.get(str(i), soft.get(i, 0.0))) for i in range(len(levels))} if isinstance(soft, dict) else None
+        else:
+            raise ValueError(f"unknown question type {kind}")
+        row = {"id": f"kev:{source}:{rid}:{qid}", "group_id": f"kev:{source}:{gid}", "source": f"kev_{source}",
+               "source_revision": meta.get("revision") or source_revision, "source_split": meta.get("split") or split, "split": split,
+               "task": "general", "type": kind, "state": state, "question": question, "options": options, "target": target, "meta": row_meta}
+        if dist and abs(sum(dist.values()) - 1) < 1e-6:
+            row["target_distribution"] = dist
+            row["target"] = max(dist, key=dist.get)
+        validate_row(row)
+        rows.append(row)
+    return rows