Kagamicho/cs_chatbot

1"""Weekly export: chat logs → CS review CSV.2 3Filters: 👎 feedback, retrieval confidence < threshold, Pro escalations, no sources.4Outputs data/review_queue/<week>.csv for CS to fill in correct answers.5"""6import argparse7import csv8import json9from pathlib import Path10 11 12def _load_log_lines(log_dir: Path) -> list[dict]:13    out: list[dict] = []14    for path in sorted(log_dir.glob("chat_*.jsonl")):15        for line in path.read_text(encoding="utf-8").splitlines():16            if not line.strip():17                continue18            try:19                out.append(json.loads(line))20            except json.JSONDecodeError:21                continue22    return out23 24 25def collect_review_candidates(log_dir: Path) -> list[dict]:26    lines = _load_log_lines(log_dir)27    turns: dict[str, dict] = {}28    feedback: dict[str, list[dict]] = {}29    for rec in lines:30        if rec.get("type") == "turn":31            turns[rec["turn_id"]] = rec32        elif rec.get("type") == "feedback":33            feedback.setdefault(rec["turn_id"], []).append(rec)34 35    cands: list[dict] = []36    for tid, turn in turns.items():37        reasons: list[str] = []38        if any(f.get("rating", 0) < 0 for f in feedback.get(tid, [])):39            reasons.append("thumbs_down")40        if "pro" in str(turn.get("model", "")):41            reasons.append("pro_escalation")42        if not turn.get("sources"):43            reasons.append("no_sources")44        if reasons:45            cands.append({**turn, "review_reasons": reasons})46    return cands47 48 49def write_review_csv(candidates: list[dict], output_path: Path) -> None:50    output_path.parent.mkdir(parents=True, exist_ok=True)51    with output_path.open("w", encoding="utf-8-sig", newline="") as f:52        w = csv.writer(f)53        w.writerow(["turn_id", "question", "current_answer", "model", "reasons", "correct_answer", "tags"])54        for c in candidates:55            w.writerow([56                c["turn_id"],57                c.get("question", ""),58                c.get("answer", ""),59                c.get("model", ""),60                ",".join(c.get("review_reasons", [])),61                "",  # CS fills in62                "",  # CS fills in63            ])64 65 66def main() -> None:67    parser = argparse.ArgumentParser()68    parser.add_argument("--log-dir", type=Path, default=Path("data/logs"))69    parser.add_argument("--output", type=Path, required=True)70    args = parser.parse_args()71    cands = collect_review_candidates(log_dir=args.log_dir)72    write_review_csv(cands, args.output)73    print(f"Wrote {len(cands)} candidates to {args.output}")74 75 76if __name__ == "__main__":77    main()78