"""TTS 텍스트 전처리/WAV 병합 유틸리티 (legacy_gradio.py 이식).""" from __future__ import annotations import re import tempfile import wave from pathlib import Path # 멘트/REST 경로(enqueue_text)의 텍스트 청크 분할 기준 (한국어 글자수). # split_text 는 이 길이까지 문장을 모아 한 블록으로 합성하는데, 블록 하나는 워커 # 생성 토큰 상한(TTS_MAX_NEW_TOKENS=2048 ≈ ~19s 오디오)을 넘으면 끝이 잘린다. # 한국어 차분한 멘트는 ~0.2s/글자 → 100자 ≈ ~22s 로 상한을 초과해, 3문장짜리 인사 # (예: "…보름이의 사인 코너와 함께합니다")의 마지막 문장이 합성에서 통째로 잘리던 # 회귀(사용자 보고 2026-06-29, /api/tts/generate 측정으로 19.3s 컷 확인). 70자 ≈ # ~15s 로 상한 아래 여유를 두면 긴 멘트가 2+ 블록으로 쪼개져 합성→merge 로 완창된다. # (자유대화 chat 은 enqueue_chunk 별도 경로라 이 값 무관.) # 참고: scaleout-cosyvoice image 는 worker packing 폐기(2026-05-22) 로 vLLM # max_model_len 상한을 제거(업스트림 default ~32K)했으므로 이 값과의 결합 제약은 # 해소됨. 단, scaleout-dual image 는 여전히 max_model_len=2048 packing patch 를 # 유지하므로 dual 운영 시 CHUNK_SIZE 를 크게 키우면 토큰 초과 가능 — 상세는 # doc/runbook/tts_engine_multi.md. CHUNK_SIZE_DEFAULT = 70 def clean_for_speech(text: str) -> str: """마크다운/URL/특수문자·섹션 라벨을 제거해 TTS 입력에 적합하게 정제. LLM 이 규칙을 어기고 `**1) 상황 설명:**` 같은 구조적 마크업을 넣어도 TTS 로 이상하게 들리지 않게 클린업. 프롬프트 엔지니어링 + 이 후처리 2중으로 방어. CosyVoice 이벤트 마커 14종(req_tts_event_markers)은 **정확 일치 round-trip 보존** — 아래 기호 제거(step 6/7)가 `[laughter]`→"laughter" 로 벗겨 영단어로 낭독되는 사고를 막는다. 실제 발화/제거는 엔진 확정 dispatch 에서 결정(tts_markers.prepare_text_for_engine). ⛔ 임의 괄호는 지금처럼 제거(LLM 마크업 방어 불변) — 14종 정확 일치만 통과. """ from tts_markers import protect_markers, restore_markers text, _saved_markers = protect_markers(text) _section_labels = ( "상황|설명|해석|예시|정답|질문|답변|힌트|주의|참고|비고|팁|요약|결론|" "소개|대사|멘트|이름|나이|직업|성별|배경|" "한국어\\s*상황\\s*설명|한국어\\s*해석|한국어\\s*뜻|" "영어\\s*문장|영어\\s*표현|" "Q|A|q|a|Tip|TIP|Note|note|Hint|hint|Example|example" ) # 1) URL 먼저 제거. text = re.sub(r"https?://\S+", "", text) # 2) 마크다운 강조/코드/헤더/인용 기호 먼저 — 이래야 `**1) 상황:**` 에서 번호·라벨 규칙이 동작. text = re.sub(r"\*\*+", "", text) text = re.sub(r"__+", "", text) text = re.sub(r"(?m)^[ \t]*[#]{1,6}[ \t]+", "", text) text = re.sub(r"(?m)^[ \t]*>[ \t]+", "", text) # 3) 줄 시작의 불릿 / 번호 라벨 / 원문자 제거. text = re.sub(r"(?m)^[ \t]*[-*•][ \t]+", "", text) text = re.sub(r"(?m)^[ \t]*\(?\d+\s*[.)）][ \t]*", "", text) text = re.sub(r"(?m)^[ \t]*[①-⑳⒈-⒛]\s*", "", text) # 4) 줄 시작 + 라인 중간 모두에서 섹션 라벨 (`상황:`, `Q:`, `한국어 해석:`) 제거. # 줄 시작이 가장 일반적인 패턴, 줄 중간이더라도 단독 라벨 뒤 콜론이면 TTS 가 이상하게 읽음. label_re = rf"(?:{_section_labels})[ \t]*[:：]" text = re.sub(rf"(?m)^[ \t]*{label_re}[ \t]*", "", text) # 문장 중간에 " 상황: " 형태로 나와도 제거 — 공백 · 문장 시작 뒤에 오는 경우. text = re.sub(rf"(?:(?<=\s)|(?<=[.!?。]\s)){label_re}[ \t]*", "", text) # 5) 느낌표 제거. TTS 가 과한 흥분/외침으로 읽는 것을 방지. # chat / Voice Moments / 즐겨찾기 재생성 등 모든 경로가 이 함수를 거치므로 # 한 번에 일괄 적용됨. # — 인접한 .?, .? 가 있으면 그냥 삭제 (?!. → ?) # — 그 외엔 마침표로 치환 (응원할게! → 응원할게.) — 문장경계 보존용. text = re.sub(r"[!！]+(?=[.?])", "", text) text = re.sub(r"(?<=[.?])[!！]+", "", text) text = re.sub(r"[!！]+", ".", text) # 6) 남은 기타 기호 제거. text = re.sub(r"[*_`#~\[\](){}>|]", "", text) text = re.sub(r"[^\w\s가-힣a-zA-Z0-9.,?·\-:;]", "", text) # 7) 빈 줄 과도 중첩 → 단일 줄바꿈, 연속 공백 → 단일 공백. text = re.sub(r"\n{2,}", "\n", text) text = re.sub(r"[ \t]{2,}", " ", text) # 8) 보호했던 이벤트 마커 복원 (round-trip). text = restore_markers(text, _saved_markers) return text.strip() def split_text(text: str, max_len: int = CHUNK_SIZE_DEFAULT) -> list[str]: """문장 경계를 우선한 텍스트 청크 분할. 이벤트 마커 paired 스팬(…·…)은 청크 경계에서 찢기지 않게 강제 수용(req_tts_event_markers P1.2) — 여는 태그만 있고 닫는 태그가 없는 버퍼는 max_len 을 넘겨도 flush 하지 않고 스팬이 닫힐 때까지 누적한다. """ from tts_markers import PAIRED_MARKERS clean = clean_for_speech(text) if not clean: return [] def _has_open_paired(s: str) -> bool: # 여는 태그 수 > 닫는 태그 수 = 아직 안 닫힌 스팬 → 자르면 안 됨. return any( s.count(open_t) > s.count(close_t) for open_t, close_t in PAIRED_MARKERS.values() ) sentences = re.split(r"(?<=[.!?。])\s*", clean) sentences = [s.strip() for s in sentences if s.strip()] chunks: list[str] = [] buf = "" for sent in sentences: if buf and len(buf) + len(sent) > max_len and not _has_open_paired(buf): chunks.append(buf.strip()) buf = sent else: buf = (buf + " " + sent) if buf else sent if buf.strip(): chunks.append(buf.strip()) if not chunks and clean: # 문장 경계가 전혀 없는 장문 폴백 — paired 스팬이 있으면 통째 유지(태그 분리 금지). if _has_open_paired(clean) or PAIRED_MARKERS and any( o in clean for o, _c in PAIRED_MARKERS.values()): chunks = [clean] else: chunks = [clean[i : i + max_len] for i in range(0, len(clean), max_len)] return chunks def find_last_sentence_boundary(text: str) -> int: """마지막 문장 끝 인덱스(+1). 없으면 -1.""" last = -1 for m in re.finditer(r"[.!?。~](?=\s|$)", text): last = m.end() return last # --- 한국어 숫자 읽기 정규화 (TTS) ------------------------------------------- # CosyVoice/Qwen3 의 내장 텍스트 정규화(TN)는 한국어 숫자를 못 읽음(중/영 전용) → # 한글+아라비아숫자는 영어로 읽힘("24.6"→"twenty four point six"). 합성 전에 숫자를 # 한글로 풀어써서 보낸다(엔진 무관). 문맥별 읽기: 시각=고유어(네 시), 분/일/월/년·일반 # =한자어, 소수=정수+점+자릿수, 월=유월/시월 불규칙. (라디오/멘트 합성 경로에서 적용.) _SINO = "영일이삼사오육칠팔구" _SINO_SMALL = ["", "십", "백", "천"] # 일의 자리부터 _SINO_BIG = ["", "만", "억", "조", "경"] # 4자리 그룹 단위 _NATIVE_ONES = ["", "한", "두", "세", "네", "다섯", "여섯", "일곱", "여덟", "아홉"] _NATIVE_TENS = ["", "열", "스물", "서른", "마흔", "쉰", "예순", "일흔", "여든", "아흔"] _MONTHS = {1: "일", 2: "이", 3: "삼", 4: "사", 5: "오", 6: "유", 7: "칠", 8: "팔", 9: "구", 10: "시", 11: "십일", 12: "십이"} # 고유어로 세는 대표 단위. 그 외 단위(원·도·년·번 등)는 한자어(일반 정규화가 처리). # 'N번'은 한자어(교향곡 40번=사십 번, 3번 출구=삼 번)라 제외 — 고유어 '한 번/두 번'은 # 보통 한글로 쓰임. '대'도 N대=사십 대(연령) 등 한자어 빈도가 높아 제외. _NATIVE_COUNTERS = "개|명|살|마리|시간|군데|가지|차례|권|잔|채|그루|송이|벌|켤레|쪽" _NUM_RE = r"(\d{1,3}(?:,\d{3})+|\d+)" # 천단위 콤마 허용 def _sino_under_man(n: int) -> str: """0 str: """정수 → 한자어 읽기.""" if n == 0: return "영" if n < 0: return "마이너스 " + _sino(-n) groups = [] while n > 0: groups.append(n % 10000) n //= 10000 parts = [] for i in range(len(groups) - 1, -1, -1): g = groups[i] if g == 0: continue seg = _sino_under_man(g) if g == 1 and _SINO_BIG[i] == "만": seg = "" # 1만 = '만' (일만 생략, 관용) parts.append(seg + _SINO_BIG[i]) return "".join(parts) def _native(n: int): """1..99 → 고유어 관형형 (한·두·네·스무·스물한...). 범위 밖이면 None(→한자어 폴백).""" if n <= 0 or n > 99: return None tens, ones = divmod(n, 10) t = ("스무" if ones == 0 else "스물") if tens == 2 else _NATIVE_TENS[tens] return t + _NATIVE_ONES[ones] def _num_or_digits(s: str) -> str: """일반 정수 — 앞자리 0(전화·코드)이면 자릿수 하나씩(공일공), 아니면 한자어.""" raw = s.replace(",", "") if len(raw) > 1 and raw[0] == "0": return "".join("공" if c == "0" else _SINO[int(c)] for c in raw) return _sino(int(raw)) def normalize_korean_numbers(text: str) -> str: """아라비아 숫자를 한국어 읽기로 정규화. (TTS 가 숫자를 영어로 읽는 것 방지.) 예: "6월 17일 새벽 4시, FM 24.6" → "유월 십칠일 새벽 네 시, FM 이십사 점 육". 순서 중요: 소수→시간→시→월→일/년/분/초→고유어단위→퍼센트→일반정수. """ if not text or not any(c.isdigit() for c in text): return text n = _NUM_RE def _int(m): return int(m.group(1).replace(",", "")) def _hr(m): # 고유어 시각, 범위 밖이면 한자어 v = _int(m) return (_native(v) or _sino(v)) text = re.sub(rf"{n}\.(\d+)", lambda m: _sino(int(m.group(1).replace(",", ""))) + " 점 " + "".join(_SINO[int(c)] for c in m.group(2)), text) text = re.sub(rf"{n}\s*시간", lambda m: _hr(m) + " 시간", text) text = re.sub(rf"{n}\s*시", lambda m: _hr(m) + " 시", text) text = re.sub(rf"{n}\s*월", lambda m: (_MONTHS.get(_int(m), _sino(_int(m)))) + "월", text) text = re.sub(rf"{n}\s*일", lambda m: _sino(_int(m)) + "일", text) text = re.sub(rf"{n}\s*년", lambda m: _sino(_int(m)) + "년", text) text = re.sub(rf"{n}\s*분", lambda m: _sino(_int(m)) + " 분", text) text = re.sub(rf"{n}\s*초", lambda m: _sino(_int(m)) + " 초", text) text = re.sub(rf"{n}\s*({_NATIVE_COUNTERS})", lambda m: (_native(_int(m)) or _sino(_int(m))) + " " + m.group(2), text) text = re.sub(rf"{n}\s*%", lambda m: _sino(_int(m)) + " 퍼센트", text) text = re.sub(n, lambda m: _num_or_digits(m.group(1)), text) return text def concat_wav(paths: list[str | Path]) -> str | None: """WAV 파일들을 한 파일로 이어붙여 임시 경로 반환.""" paths = [str(p) for p in paths if p] if not paths: return None if len(paths) == 1: return paths[0] tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) with wave.open(paths[0], "rb") as first: params = first.getparams() with wave.open(tmp.name, "wb") as out: out.setparams(params) out.writeframes(first.readframes(first.getnframes())) for p in paths[1:]: with wave.open(p, "rb") as w: out.writeframes(w.readframes(w.getnframes())) return tmp.name