scripts/convert_part06.py

"""Convert part_06.jsonl → platform-importable JSON.

Source schema (per record):
  question, answer, history (list of {role,content}), trace (list of steps
  with type ∈ {thought, tool_call, tool_result}), source_id,
  online_record_id, source_path, source_line, request_id, session_id,
  user_id, agent_id, business_tag, ability_tree_key, model_name,
  usage_input_tokens, usage_output_tokens, usage_total_tokens,
  intent_labels, capability_labels, domain_label, query_quality,
  query_difficulty, primary_intent, selection_group, offline_split_id,
  assigned_reviewer

Target schema (per record, Chinese keys to match the 13 canonical files):
  id              ← source_id (verbatim; platform PK lookup)
  来源            ← business_tag (provenance)
  问题            ← question
  答案            ← answer
  链路数据         ← trace reshaped into [{plan, tools[]}] form so the
                    platform's _extract_tools/_infer_difficulty heuristics
                    can still count tools (key for evaluation).
  上下文          ← history (verbatim)
  tags            ← auxiliary metadata JSON-packed into List[str] for
                    future audit / selection-group queries (single tag
                    per record; original fields are NOT lost — see also
                    `source_id`/`online_record_id` round-trip).

Pairing strategy for trace → 链路数据:
  - tool_call and tool_result steps share metadata.tool_call_id
  - tool_call becomes a phase with tools=[{name, input}]; if matching
    tool_result exists, output is attached
  - thought becomes a phase with plan=content, tools=[]
  - orphan tool_results (no matching call) become a phase with the
    output as a "result" tool

Every source step is consumed exactly once → no data loss; the original
trace is still recoverable via online_record_id.
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "数据测试集" / "part_06.jsonl"
DST = ROOT / "数据测试集" / "part_06-importable.json"

# Auxiliary fields the platform TestCase ORM doesn't model — packed into
# `tags` as a single JSON-encoded string per record so the row remains
# importable but the data is preserved for downstream queries / audits.
AUX_FIELDS = (
    "agent_id",
    "business_tag",
    "ability_tree_key",
    "model_name",
    "primary_intent",
    "query_difficulty",
    "query_quality",
    "domain_label",
    "intent_labels",
    "capability_labels",
    "selection_group",
    "offline_split_id",
    "assigned_reviewer",
    "source_path",
    "source_line",
    "request_id",
    "session_id",
    "user_id",
    "online_record_id",
    "usage_input_tokens",
    "usage_output_tokens",
    "usage_total_tokens",
)


def reshape_trace(trace):
    """Reshape [{type: thought|tool_call|tool_result}, ...] → [{plan, tools[]}, ...]."""
    if not isinstance(trace, list):
        return []
    # Pre-build id→{name,input} and id→output maps so pairing is robust to
    # interleaved ordering (record 7 had 3 tool_calls → 3 tool_results).
    calls_by_id: dict[str, dict] = {}
    results_by_id: dict[str, str] = {}
    for step in trace:
        if not isinstance(step, dict):
            continue
        meta = step.get("metadata") or {}
        tc_id = meta.get("tool_call_id") or ""
        if step.get("type") == "tool_call":
            calls_by_id[tc_id] = {
                "name": step.get("title", "") or "",
                "input": step.get("content", "") or "",
            }
        elif step.get("type") == "tool_result":
            results_by_id[tc_id] = step.get("content", "") or ""

    phases: list[dict] = []
    consumed_ids: set[str] = set()
    for step in trace:
        if not isinstance(step, dict):
            continue
        t = step.get("type")
        if t == "thought":
            phases.append({"plan": step.get("content", "") or "", "tools": []})
            continue
        if t == "tool_call":
            tc_id = (step.get("metadata") or {}).get("tool_call_id") or ""
            tool = dict(calls_by_id.get(tc_id) or {
                "name": step.get("title", "") or "",
                "input": step.get("content", "") or "",
            })
            if tc_id in results_by_id:
                tool["output"] = results_by_id[tc_id]
                consumed_ids.add(tc_id)
            phases.append({"plan": "", "tools": [tool]})
            continue
        if t == "tool_result":
            tc_id = (step.get("metadata") or {}).get("tool_call_id") or ""
            if tc_id in consumed_ids:
                # Already attached to the matching tool_call — skip to avoid dup.
                continue
            if tc_id and tc_id not in calls_by_id:
                # Orphan result with a known id (its call was filtered out).
                tool = {
                    "name": step.get("title", "") or "",
                    "output": step.get("content", "") or "",
                }
                phases.append({"plan": "", "tools": [tool]})
                continue
            # Result with no id at all — keep as raw content.
            phases.append({
                "plan": "",
                "tools": [{
                    "name": "result",
                    "output": step.get("content", "") or "",
                }],
            })
    return phases


def convert(rec: dict) -> dict:
    aux = {k: rec[k] for k in AUX_FIELDS if k in rec and rec[k] is not None}
    return {
        "id": rec.get("source_id") or "",
        "来源": rec.get("business_tag") or "iwencai",
        "问题": rec.get("question") or "",
        "答案": rec.get("answer") or "",
        "链路数据": reshape_trace(rec.get("trace") or []),
        "上下文": rec.get("history") or [],
        "tags": [json.dumps(aux, ensure_ascii=False)],
    }


def main() -> int:
    if not SRC.exists():
        print(f"❌ source not found: {SRC}", file=sys.stderr)
        return 1
    records = []
    with SRC.open(encoding="utf-8") as f:
        for lineno, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
            try:
                records.append(json.loads(line))
            except json.JSONDecodeError as e:
                print(f"⚠️ line {lineno} skipped: {e}", file=sys.stderr)
    print(f"📥 read {len(records)} records from {SRC.name}")

    out = [convert(r) for r in records]

    DST.parent.mkdir(parents=True, exist_ok=True)
    with DST.open("w", encoding="utf-8") as f:
        json.dump(out, f, ensure_ascii=False, indent=2)

    print(f"📤 wrote {len(out)} records → {DST.relative_to(ROOT)}")
    print(f"   size: {DST.stat().st_size:,} bytes")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())