Kh0128/Aphasia_Classificifier

1#!/usr/bin/env python32# -*- coding: utf-8 -*-3"""4cha_json.py — 將單一 CLAN .cha 轉成 JSON（強化 %mor/%wor/%gra 對齊）5用法：6    # CLI7    python3 cha_json.py --input /path/to/input.cha --output /path/to/output.json8 9程式化呼叫（供 pipeline 使用）：10    from cha_json import cha_to_json_file, cha_to_dict11    out_path, data = cha_to_json_file("/path/in.cha", "/path/out.json")12    data2 = cha_to_dict("/path/in.cha")13"""14 15from __future__ import annotations16import re17import json18import sys19import argparse20from pathlib import Path21from collections import defaultdict22from typing import List, Dict, Any, Tuple, Optional23 24# 可接受的跨行停止條件（用於 %mor/%wor/%gra 合併）25TAG_PREFIXES = ("*PAR", "*INV", "%mor:", "%gra:", "%wor:", "@")26WORD_RE      = re.compile(r"[A-Za-z0-9]+")27 28# 病人角色：PAR / PAR0 / PAR1 / ...29ID_PAR_RE = re.compile(r"\|PAR\d*\|")30 31# 對話行：*INV: 或 *PAR0: / *PAR1: / ...32UTTER_RE = re.compile(r"^\*(INV|PAR\d+):")33 34# ────────── 同義集合（對齊時容忍形態變化） ──────────35SYN_SETS = [36    {"be", "am", "is", "are", "was", "were", "been", "being"},37    {"have", "has", "had"},38    {"do", "does", "did", "done", "doing"},39    {"go", "goes", "going", "went", "gone"},40    {"run", "runs", "running", "ran"},41    {"see", "sees", "seeing", "saw", "seen"},42    {"get", "gets", "getting", "got", "gotten"},43    {"drop", "drops", "dropping", "dropped"},44    {"swim", "swims", "swimming", "swam", "swum"},45]46def same_syn(a: str, b: str) -> bool:47    if not a or not b:48        return False49    for s in SYN_SETS:50        if a in s and b in s:51            return True52    return False53 54def canonical(txt: str) -> str:55    """token/word → 比對用字串：去掉 & ~ - | 之後的非字母數字、轉小寫"""56    head = re.split(r"[~\-\&|]", txt, 1)[0]57    m = WORD_RE.search(head)58    return m.group(0).lower() if m else ""59 60def merge_multiline(block_lines: List[str]) -> str:61    """62    合併跨行的 %mor/%wor/%gra。63    規則：以 '%' 開頭者作為起始，往下串，遇到新標籤或 @ 開頭就停。64    """65    merged, buf = [], None66    for raw in block_lines:67        ln = raw.rstrip("\n").replace("\x15", "")  # 去掉 CLAN 控制字68        if ln.lstrip().startswith("%") and ":" in ln:69            if buf:70                merged.append(buf)71            buf = ln72        else:73            if buf and ln.strip():74                buf += " " + ln.strip()75            else:76                merged.append(ln)77    if buf:78        merged.append(buf)79    return "\n".join(merged)80 81def cha_to_json(lines: List[str]) -> Dict[str, Any]:82    """83    將 .cha 檔行列表轉 JSON 結構。84    回傳格式：85    {86      "sentences": [...],87      "pos_mapping": {...},88      "grammar_mapping": {...},89      "aphasia_types": {...},90      "text_all": "..."        # 方便下游模型使用的 PAR 合併文字91    }92    """93    # 對應表（pos / gra 從 1 起算；aphasia 類型 0 起）94    pos_map: Dict[str, int]     = defaultdict(lambda: len(pos_map) + 1)95    gra_map: Dict[str, int]     = defaultdict(lambda: len(gra_map) + 1)96    aphasia_map: Dict[str, int] = defaultdict(lambda: len(aphasia_map))97 98    data: List[Dict[str, Any]] = []99    sent: Optional[Dict[str, Any]] = None100 101    i = 0102    while i < len(lines):103        line = lines[i].rstrip("\n")104 105        # 啟段106        if line.startswith("@Begin"):107            sent = {108                "sentence_id": f"S{len(data)+1}",109                "sentence_pid": None,110                "aphasia_type": None,   # 若最後仍沒有，就標 UNKNOWN111                "dialogues": []         # [ { "INV": [...], "PAR": [...] }, ... ]112            }113            i += 1114            continue115 116        # 結束117        if line.startswith("@End"):118            if sent and sent["dialogues"]:119                if not sent.get("aphasia_type"):120                    sent["aphasia_type"] = "UNKNOWN"121                    aphasia_map["UNKNOWN"]122                data.append(sent)123            sent = None124            i += 1125            continue126 127        # 句子屬性128        if sent and line.startswith("@PID:"):129            parts = line.split("\t")130            if len(parts) > 1:131                sent["sentence_pid"] = parts[1].strip()132            i += 1133            continue134 135        if sent and line.startswith("@ID:"):136            # 是否為病人那位 PAR*137            if ID_PAR_RE.search(line):138                aph = "UNKNOWN"139                # 如果 @ID 有標註失語類型，可在此使用 regex 抓出來並替換 aph140                # m = re.search(r"WAB:([A-Za-z]+)", line)141                # if m: aph = m.group(1)142                aph = aph.upper()143                aphasia_map[aph]            # 建立 map（自動編號）144                sent["aphasia_type"] = aph145            i += 1146            continue147 148        # 對話行：*INV: 或 *PARx:149        if sent and UTTER_RE.match(line):150            role_tag = UTTER_RE.match(line).group(1)151            role = "INV" if role_tag == "INV" else "PAR"152 153            if not sent["dialogues"]:154                sent["dialogues"].append({"INV": [], "PAR": []})155            # 新輪對話：若來的是 INV 且上一輪已有 PAR，視為下一輪156            if role == "INV" and sent["dialogues"][-1]["PAR"]:157                sent["dialogues"].append({"INV": [], "PAR": []})158 159            # 新增一個空 turn（之後 %mor/%wor/%gra 會補）160            sent["dialogues"][-1][role].append(161                {"tokens": [], "word_pos_ids": [], "word_grammar_ids": [], "word_durations": [], "utterance_text": ""}162            )163            i += 1164            continue165 166        # %mor167        if sent and line.startswith("%mor:"):168            blk = [line]; i += 1169            while i < len(lines) and not lines[i].lstrip().startswith(TAG_PREFIXES):170                blk.append(lines[i]); i += 1171 172            units = merge_multiline(blk).replace("%mor:", "").strip().split()173            toks, pos_ids = [], []174            for u in units:175                if "|" in u:176                    pos, rest = u.split("|", 1)177                    word = rest.split("|", 1)[0]178                    toks.append(word)179                    pos_ids.append(pos_map[pos])180 181            dlg = sent["dialogues"][-1]182            tgt = dlg["PAR"][-1] if dlg["PAR"] else dlg["INV"][-1]183            tgt["tokens"], tgt["word_pos_ids"] = toks, pos_ids184            # 也保存 plain text 供下游模型使用185            tgt["utterance_text"] = " ".join(toks).strip()186            continue187 188        # %wor189        if sent and line.startswith("%wor:"):190            blk = [line]; i += 1191            while i < len(lines) and not lines[i].lstrip().startswith(TAG_PREFIXES):192                blk.append(lines[i]); i += 1193 194            merged = merge_multiline(blk).replace("%wor:", "").strip()195            # 抓 <word> <start>_<end>196            raw_pairs = re.findall(r"(\S+)\s+(\d+)_(\d+)", merged)197            wor = [(w, int(s), int(e)) for (w, s, e) in raw_pairs]198 199            dlg = sent["dialogues"][-1]200            tgt = dlg["PAR"][-1] if dlg["PAR"] else dlg["INV"][-1]201 202            # 與 %mor tokens 對齊，duration = end - start203            aligned: List[Tuple[str, int]] = []204            j = 0205            for tok in tgt.get("tokens", []):206                c_tok = canonical(tok)207                match = None208                for k in range(j, len(wor)):209                    c_w = canonical(wor[k][0])210                    if (211                        c_tok == c_w212                        or c_w.startswith(c_tok)213                        or c_tok.startswith(c_w)214                        or same_syn(c_tok, c_w)215                    ):216                        match = wor[k]217                        j = k + 1218                        break219                dur = (match[2] - match[1]) if match else 0220                aligned.append([tok, dur])221            tgt["word_durations"] = aligned222            continue223 224        # %gra225        if sent and line.startswith("%gra:"):226            blk = [line]; i += 1227            while i < len(lines) and not lines[i].lstrip().startswith(TAG_PREFIXES):228                blk.append(lines[i]); i += 1229 230            units = merge_multiline(blk).replace("%gra:", "").strip().split()231            triples = []232            for u in units:233                # 例：1|2|DET234                parts = u.split("|")235                if len(parts) == 3:236                    a, b, r = parts237                    if a.isdigit() and b.isdigit():238                        triples.append([int(a), int(b), gra_map[r]])239 240            dlg = sent["dialogues"][-1]241            tgt = dlg["PAR"][-1] if dlg["PAR"] else dlg["INV"][-1]242            tgt["word_grammar_ids"] = triples243            continue244 245        # 其他行246        i += 1247 248    # 收尾（檔案若意外沒 @End）249    if sent and sent["dialogues"]:250        if not sent.get("aphasia_type"):251            sent["aphasia_type"] = "UNKNOWN"252            aphasia_map["UNKNOWN"]253        data.append(sent)254 255    # 建立 text_all：把所有 PAR utterance_text 串起來256    par_texts: List[str] = []257    for s in data:258        for turn in s.get("dialogues", []):259            for par_ut in turn.get("PAR", []):260                if par_ut.get("utterance_text"):261                    par_texts.append(par_ut["utterance_text"])262    text_all = "\n".join(par_texts).strip()263 264    return {265        "sentences": data,266        "pos_mapping": dict(pos_map),267        "grammar_mapping": dict(gra_map),268        "aphasia_types": dict(aphasia_map),269        "text_all": text_all270    }271 272# ────────── 封裝：檔案 → dict / 檔案 → 檔案 ──────────273def cha_to_dict(cha_path: str) -> Dict[str, Any]:274    """讀取 .cha 檔並回傳 dict（不寫檔）。"""275    p = Path(cha_path)276    if not p.exists():277        raise FileNotFoundError(f"找不到檔案: {cha_path}")278    with p.open("r", encoding="utf-8") as fh:279        lines = fh.readlines()280    return cha_to_json(lines)281 282def cha_to_json_file(cha_path: str, output_json: Optional[str] = None) -> Tuple[str, Dict[str, Any]]:283    """284    將 .cha 轉成 JSON 並寫檔。285    回傳：(output_json_path, data_dict)286    """287    data = cha_to_dict(cha_path)288    out_path = Path(output_json) if output_json else Path(cha_path).with_suffix(".json")289    out_path.parent.mkdir(parents=True, exist_ok=True)290    with out_path.open("w", encoding="utf-8") as fh:291        json.dump(data, fh, ensure_ascii=False, indent=4)292    return str(out_path), data293 294# ────────── CLI ──────────295def parse_args():296    p = argparse.ArgumentParser()297    p.add_argument("--input", "-i", type=str, required=True, help="輸入 .cha 檔")298    p.add_argument("--output", "-o", type=str, required=True, help="輸出 .json 檔")299    return p.parse_args()300 301def cha_to_json_path(cha_path: str, output_json: str | None = None) -> str:302    """Backward-compatible alias for old code."""303    out, _ = cha_to_json_file(cha_path, output_json=output_json)304    return out305 306def main():307    args = parse_args()308    in_path  = Path(args.input)309    out_path = Path(args.output)310 311    if not in_path.exists():312        sys.exit(f"❌ 找不到檔案: {in_path}")313 314    with in_path.open("r", encoding="utf-8") as fh:315        lines = fh.readlines()316 317    dataset = cha_to_json(lines)318 319    out_path.parent.mkdir(parents=True, exist_ok=True)320    with out_path.open("w", encoding="utf-8") as fh:321        json.dump(dataset, fh, ensure_ascii=False, indent=4)322 323    print(324        f"✅ 轉換完成 → {out_path}（句數 {len(dataset['sentences'])}，"325        f"pos={len(dataset['pos_mapping'])}，gra={len(dataset['grammar_mapping'])}，"326        f"類型鍵={list(dataset['aphasia_types'].keys())}）"327    )328 329if __name__ == "__main__":330    main()331