Kagamicho/cs_chatbot

1"""Convert CS-supplied CSV/Excel into canonical qa_pairs/*.jsonl.2 3Modes:4  --mapping <yaml>   use a saved column-mapping YAML (recurring formats)5  --interactive      auto-detect encoding + columns, confirm, optionally save mapping6 7Usage:8    uv run python scripts/csv_to_qa_jsonl.py \\9        --input data/sources/cs_inbox/2026-Q2-cs.csv \\10        --mapping data/sources/cs_inbox/2026-Q2-cs.mapping.yaml \\11        --output data/sources/qa_pairs/2026-Q2-cs.jsonl12"""13import argparse14import json15import sys16from datetime import datetime, timezone17from pathlib import Path18 19import pandas as pd20import yaml21from charset_normalizer import from_bytes22 23_SYNONYMS = {24    "question": ["question", "q", "質問", "お問い合わせ", "お問合せ", "お問い合わせ内容", "ご質問", "内容"],25    "answer": ["answer", "a", "回答", "ご回答", "回答内容", "ご返答", "対応"],26    "tags": ["tags", "tag", "カテゴリ", "区分", "種別", "ジャンル"],27    "source": ["source", "受付日", "日付", "対応日", "date"],28}29 30 31def _detect_encoding(path: Path) -> str:32    blob = path.read_bytes()[:4096]33    res = from_bytes(blob).best()34    return res.encoding if res else "utf-8"35 36 37def _fuzzy_pick(candidates: list[str], synonyms: list[str]) -> str | None:38    lowered = {c.lower(): c for c in candidates}39    for s in synonyms:40        if s in candidates:41            return s42        if s.lower() in lowered:43            return lowered[s.lower()]44    for s in synonyms:45        for c in candidates:46            if s.lower() in c.lower() or c.lower() in s.lower():47                return c48    return None49 50 51def autodetect_mapping(input_path: Path) -> dict:52    enc = _detect_encoding(input_path)53    df = pd.read_csv(input_path, encoding=enc, nrows=5)54    col_map: dict[str, str] = {}55    for canonical, syns in _SYNONYMS.items():56        picked = _fuzzy_pick(list(df.columns), syns)57        if picked:58            col_map[canonical] = picked59    return {60        "encoding": enc,61        "delimiter": ",",62        "skip_rows": 0,63        "column_map": col_map,64        "defaults": {"verified_by": "CS_team", "priority": 1.2},65        "filters": {"drop_if_question_empty": True, "drop_if_answer_shorter_than": 10},66    }67 68 69def _generate_id(input_path: Path, idx: int) -> str:70    today = datetime.now(timezone.utc).strftime("%Y%m%d")71    return f"qa-{today}-{input_path.stem}-{idx:04d}"72 73 74def convert_with_mapping(75    *,76    input_path: Path,77    mapping_path: Path,78    output_path: Path,79) -> int:80    mapping = yaml.safe_load(mapping_path.read_text(encoding="utf-8"))81    df = pd.read_csv(82        input_path,83        encoding=mapping.get("encoding", "utf-8"),84        sep=mapping.get("delimiter", ","),85        skiprows=mapping.get("skip_rows", 0),86    )87    cm = mapping["column_map"]88    defaults = mapping.get("defaults", {})89    filters = mapping.get("filters", {})90 91    output_path.parent.mkdir(parents=True, exist_ok=True)92    n = 093    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")94    with output_path.open("w", encoding="utf-8") as f:95        for i, row in df.iterrows():96            q = str(row.get(cm.get("question", ""), "")).strip()97            a = str(row.get(cm.get("answer", ""), "")).strip()98            if filters.get("drop_if_question_empty") and not q:99                continue100            if len(a) <= int(filters.get("drop_if_answer_shorter_than", 0)):101                continue102 103            tags_raw = row.get(cm.get("tags", ""), None) if cm.get("tags") else None104            tags = [t.strip() for t in str(tags_raw).split(",") if t and pd.notna(tags_raw)] if tags_raw else []105            source = str(row.get(cm.get("source", ""), "")) if cm.get("source") else None106 107            rec = {108                "id": _generate_id(input_path, i),109                "question": q,110                "answer": a,111                "tags": tags,112                "source": source or defaults.get("source", input_path.stem),113                "verified_by": defaults.get("verified_by"),114                "verified_at": today,115                "priority": float(defaults.get("priority", 1.2)),116                "supersedes": [],117            }118            f.write(json.dumps(rec, ensure_ascii=False) + "\n")119            n += 1120    return n121 122 123def main() -> None:124    parser = argparse.ArgumentParser()125    parser.add_argument("--input", required=True, type=Path)126    parser.add_argument("--output", required=True, type=Path)127    parser.add_argument("--mapping", type=Path, default=None)128    parser.add_argument("--interactive", action="store_true")129    args = parser.parse_args()130 131    if args.interactive or args.mapping is None:132        mapping = autodetect_mapping(args.input)133        print("Detected mapping:", json.dumps(mapping, ensure_ascii=False, indent=2))134        ans = input("Proceed with this mapping? [Y/n] ")135        if ans.lower() == "n":136            sys.exit(1)137        save = input("Save mapping YAML for reuse? [Y/n] ")138        if save.lower() != "n":139            target = args.input.with_suffix(".mapping.yaml")140            target.write_text(yaml.safe_dump(mapping, allow_unicode=True), encoding="utf-8")141            args.mapping = target142            print(f"Saved {target}")143        else:144            tmp = args.input.with_suffix(".mapping.tmp.yaml")145            tmp.write_text(yaml.safe_dump(mapping, allow_unicode=True), encoding="utf-8")146            args.mapping = tmp147 148    n = convert_with_mapping(149        input_path=args.input, mapping_path=args.mapping, output_path=args.output150    )151    print(f"Wrote {n} Q&A pairs to {args.output}")152 153 154if __name__ == "__main__":155    main()156