187 lines
8.2 KiB
Python
187 lines
8.2 KiB
Python
"""
|
||
직접 청취 교정본 일괄 반영 스크립트
|
||
====================================
|
||
사용법:
|
||
VERIFIED_CORRECTIONS 딕셔너리에 교정본 추가 후 실행
|
||
python apply_corrections.py
|
||
|
||
반영 대상:
|
||
1. final_output/{chunk_name}.json (개별 청크)
|
||
2. final_output/result_summary.json (전체 요약)
|
||
3. sample_extractor/railway_config.json (correction 사전)
|
||
"""
|
||
|
||
import json
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
|
||
BASE_DIR = Path(__file__).parent
|
||
OUTPUT_DIR = BASE_DIR / "final_output"
|
||
SUMMARY_PATH = OUTPUT_DIR / "result_summary.json"
|
||
CONFIG_PATH = BASE_DIR / "sample_extractor" / "railway_config.json"
|
||
|
||
# ════════════════════════════════════════════════════════
|
||
# ① 직접 청취 교정본 (사용자가 여기에 추가)
|
||
# speaker: "기관사" | "관제" | "모터카" | "입환취급" | "불명확"
|
||
# ════════════════════════════════════════════════════════
|
||
|
||
VERIFIED_CORRECTIONS: dict[str, dict] = {
|
||
"0528_ch020": {
|
||
"transcription": "예 노포 1005 출고, 알겠습니다.",
|
||
"speaker": "관제",
|
||
"reason": "직접청취 교정: 노포 1005열차 출고 보고 수신 확인 응답 → 관제",
|
||
},
|
||
"0528_ch021": {
|
||
"transcription": "전철 범일, 신평 1번선에 7909열차 45편성 기관사 승차했습니다.",
|
||
"speaker": "기관사",
|
||
"reason": "직접청취 교정: 전철 범일 호출 + 신평 1번선 7909열차 45편성 기관사 승차 보고",
|
||
},
|
||
"0528_ch022": {
|
||
"transcription": "전철 범일, 신평 출고선에 7903열차 상선까지 신호 확인했습니다. 수고하십시오.",
|
||
"speaker": "기관사",
|
||
"reason": "직접청취 교정: 전철 범일 호출 + 신평 출고선 7903열차 상선 신호 확인 보고",
|
||
},
|
||
"0528_ch023": {
|
||
"transcription": "7105열차 34편성 신평 출고 알겠습니다.",
|
||
"speaker": "관제",
|
||
"reason": "직접청취 교정: 7105열차 34편성 신평 출고 보고 수신 확인 → 관제 응답",
|
||
},
|
||
"0528_ch024": {
|
||
"transcription": "예, 노포 출고선에 1005열차 상회차선 신호 진로 잘 보고 나오십시오.",
|
||
"speaker": "관제",
|
||
"reason": "직접청취 교정: 나오십시오 지시형 종결 → 관제. 노포 출고선 1005열차 상회차선 신호진로 확인 지시",
|
||
},
|
||
}
|
||
|
||
# ════════════════════════════════════════════════════════
|
||
# ② STT 오인식 패턴 → correction 사전 추가 항목
|
||
# ════════════════════════════════════════════════════════
|
||
|
||
NEW_CORRECTIONS: dict[str, str] = {
|
||
# 전철 범일 변이형
|
||
"전차범인": "전철 범일", # 전차범위와 별개
|
||
"전체 보면": "전철 범일",
|
||
|
||
# 신평 관련
|
||
"신평일반선": "신평 1번선",
|
||
"심평출범산회": "신평 출고선에",
|
||
|
||
# 열차/편성 오인식
|
||
"7105열 ": "7105열차 ", # "열" 뒤 공백 → "열차"
|
||
|
||
# 행동 오인식
|
||
"조차했습니다": "승차했습니다",
|
||
|
||
# 노포 출고선 오인식
|
||
"이제 포출고선의": "노포 출고선에",
|
||
"포출고선의": "노포 출고선에",
|
||
"부출고선의": "노포 출고선에",
|
||
"부출부선의": "노포 출고선에",
|
||
|
||
# 열차번호 관련
|
||
"첫 노열차": "1005열차", # 구체적이지만 빈번한 패턴
|
||
"천 공고": "1005", # 노포 1005열차 발음 패턴
|
||
|
||
# 신호진로 오인식
|
||
"진호진로": "신호 진로",
|
||
"잘 보관하십시오": "잘 보고 나오십시오",
|
||
}
|
||
|
||
# ════════════════════════════════════════════════════════
|
||
# 실행 로직
|
||
# ════════════════════════════════════════════════════════
|
||
|
||
def apply():
|
||
print(f"\n{'='*56}")
|
||
print(f" 직접 교정 일괄 반영")
|
||
print(f" {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f"{'='*56}\n")
|
||
|
||
# ── 1) 개별 청크 JSON 업데이트 ─────────────────────────────
|
||
print("① 개별 청크 JSON 업데이트")
|
||
updated_chunks: list[str] = []
|
||
for chunk_name, patch in VERIFIED_CORRECTIONS.items():
|
||
jpath = OUTPUT_DIR / f"{chunk_name}.json"
|
||
if not jpath.exists():
|
||
print(f" ⚠️ {chunk_name}.json 없음 → 스킵")
|
||
continue
|
||
|
||
with open(jpath, encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
|
||
old_tx = data.get("transcription", "")
|
||
data["transcription"] = patch["transcription"]
|
||
data["speaker"] = patch["speaker"]
|
||
data["confidence"] = "high"
|
||
data["reason"] = patch["reason"]
|
||
data["verified"] = True
|
||
if "note" in patch:
|
||
data["note"] = patch["note"]
|
||
elif "note" in data:
|
||
del data["note"] # 이전 배치 note 초기화
|
||
|
||
with open(jpath, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
|
||
print(f" ✅ {chunk_name} [{patch['speaker']}]")
|
||
print(f" Before: {old_tx[:60]}")
|
||
print(f" After : {patch['transcription'][:60]}")
|
||
updated_chunks.append(chunk_name)
|
||
|
||
# ── 2) result_summary.json 업데이트 ────────────────────────
|
||
print(f"\n② result_summary.json 업데이트")
|
||
if SUMMARY_PATH.exists():
|
||
with open(SUMMARY_PATH, encoding="utf-8") as f:
|
||
summary = json.load(f)
|
||
|
||
updated_count = 0
|
||
for utt in summary.get("utterances", []):
|
||
cname = utt.get("chunk_name", "")
|
||
if cname in VERIFIED_CORRECTIONS:
|
||
patch = VERIFIED_CORRECTIONS[cname]
|
||
utt["transcription"] = patch["transcription"]
|
||
utt["speaker"] = patch["speaker"]
|
||
utt["confidence"] = "high"
|
||
utt["reason"] = patch["reason"]
|
||
utt["verified"] = True
|
||
if "note" in patch:
|
||
utt["note"] = patch["note"]
|
||
updated_count += 1
|
||
|
||
summary["_last_manual_correction"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
with open(SUMMARY_PATH, "w", encoding="utf-8") as f:
|
||
json.dump(summary, f, ensure_ascii=False, indent=2)
|
||
print(f" ✅ {updated_count}건 반영 완료")
|
||
else:
|
||
print(f" ⚠️ {SUMMARY_PATH} 없음 (파이프라인 실행 후 재시도)")
|
||
|
||
# ── 3) railway_config.json correction 사전 업데이트 ────────
|
||
print(f"\n③ correction 사전 업데이트")
|
||
with open(CONFIG_PATH, encoding="utf-8") as f:
|
||
cfg = json.load(f)
|
||
|
||
correction = cfg.get("correction", {})
|
||
added = []
|
||
for wrong, right in NEW_CORRECTIONS.items():
|
||
if wrong not in correction:
|
||
correction[wrong] = right
|
||
added.append((wrong, right))
|
||
print(f" ➕ '{wrong}' → '{right}'")
|
||
else:
|
||
print(f" ℹ️ 이미 존재: '{wrong}' → '{correction[wrong]}'")
|
||
|
||
cfg["correction"] = correction
|
||
cfg["_updated"] = datetime.now().strftime("%Y-%m-%d")
|
||
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
|
||
json.dump(cfg, f, ensure_ascii=False, indent=2)
|
||
print(f" ✅ correction 사전 {len(added)}건 추가")
|
||
|
||
# ── 최종 리포트 ──────────────────────────────────────────
|
||
print(f"\n{'='*56}")
|
||
print(f" 완료: 청크 {len(updated_chunks)}개 / correction {len(added)}건 추가")
|
||
print(f"{'='*56}\n")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
apply()
|