458 lines
17 KiB
Python
458 lines
17 KiB
Python
import numpy as np
|
|
import re
|
|
|
|
try:
|
|
from pypinyin import pinyin, Style
|
|
except ImportError:
|
|
pinyin = None
|
|
|
|
class ASRMappingText:
|
|
@staticmethod
|
|
def sub_cost(a: str, r: str) -> float:
|
|
return 0.0 if a == r else 1.0
|
|
|
|
def map_text(self, source_text: str, asr_data: str) -> list[int]:
|
|
n, m = len(source_text), len(asr_data)
|
|
dp = [[0.0] * (m + 1) for _ in range(n + 1)]
|
|
bt = [[0] * (m + 1) for _ in range(n + 1)]
|
|
|
|
# 初始化邊界與 DP (同前)
|
|
for i in range(1, n + 1): dp[i][0] = i; bt[i][0] = 1
|
|
for j in range(1, m + 1): dp[0][j] = j; bt[0][j] = 2
|
|
|
|
for i in range(1, n + 1):
|
|
for j in range(1, m + 1):
|
|
cost = 0.0 if source_text[i-1] == asr_data[j-1] else 1.0
|
|
c_diag = dp[i-1][j-1] + cost
|
|
c_up = dp[i-1][j] + 1
|
|
c_left = dp[i][j-1] + 1
|
|
|
|
best = c_diag; dir_ = 0
|
|
if c_up < best: best = c_up; dir_ = 1
|
|
if c_left < best: best = c_left; dir_ = 2
|
|
dp[i][j] = best; bt[i][j] = dir_
|
|
|
|
# 1. 第一次回溯：僅記錄「完全匹配」的索引
|
|
mapA = [None] * n
|
|
curr_i, curr_j = n, m
|
|
while curr_i > 0 or curr_j > 0:
|
|
direction = bt[curr_i][curr_j]
|
|
if curr_i > 0 and curr_j > 0 and direction == 0:
|
|
if source_text[curr_i - 1] == asr_data[curr_j - 1]:
|
|
mapA[curr_i - 1] = curr_j - 1
|
|
curr_i -= 1; curr_j -= 1
|
|
elif curr_i > 0 and (curr_j == 0 or direction == 1):
|
|
curr_i -= 1
|
|
else:
|
|
curr_j -= 1
|
|
|
|
# 2. 第二次掃描：補足缺失的索引 (前一個 + 1)
|
|
for i in range(n):
|
|
if mapA[i] is None:
|
|
if i == 0:
|
|
mapA[i] = 0 # 第一個字如果不匹配，從 0 開始
|
|
else:
|
|
# 這裡採用「前一個字的索引 + 1」
|
|
mapA[i] = mapA[i-1] + 1
|
|
|
|
# 確保索引不會超過 ASR 數據的最大長度
|
|
return [min(idx, m - 1) for idx in mapA]
|
|
|
|
def map_text_old(self, source_text: str, asr_data: str) -> list[int]:
|
|
n = len(source_text)
|
|
m = len(asr_data)
|
|
|
|
# 初始化 DP 表與 Backtrack 表
|
|
# dp[n+1][m+1], bt[n+1][m+1]
|
|
dp = [[0.0] * (m + 1) for _ in range(n + 1)]
|
|
bt = [[0] * (m + 1) for _ in range(n + 1)]
|
|
|
|
# 初始化邊界條件
|
|
for i in range(1, n + 1):
|
|
dp[i][0] = dp[i - 1][0] + 1
|
|
bt[i][0] = 1 # up (刪除 source)
|
|
|
|
for j in range(1, m + 1):
|
|
dp[0][j] = dp[0][j - 1] + 1
|
|
bt[0][j] = 2 # left (插入 asr)
|
|
|
|
# 填寫 DP 表
|
|
for i in range(1, n + 1):
|
|
for j in range(1, m + 1):
|
|
# 0: diag, 1: up, 2: left
|
|
c_diag = dp[i-1][j-1] + self.sub_cost(source_text[i-1], asr_data[j-1])
|
|
c_up = dp[i-1][j] + 1
|
|
c_left = dp[i][j-1] + 1
|
|
|
|
best = c_diag
|
|
dir_ = 0
|
|
|
|
if c_up < best:
|
|
best = c_up
|
|
dir_ = 1
|
|
if c_left < best:
|
|
best = c_left
|
|
dir_ = 2
|
|
|
|
dp[i][j] = best
|
|
bt[i][j] = dir_
|
|
|
|
# 回溯路徑 (Backtracking)
|
|
# Python 沒有 INT_MAX，通常用 float('inf') 或 -1 表示未對應
|
|
mapA = [float('inf')] * n
|
|
curr_i, curr_j = n, m
|
|
|
|
while curr_i > 0 or curr_j > 0:
|
|
direction = bt[curr_i][curr_j]
|
|
if curr_i > 0 and curr_j > 0 and direction == 0:
|
|
mapA[curr_i - 1] = curr_j - 1
|
|
curr_i -= 1
|
|
curr_j -= 1
|
|
elif curr_i > 0 and (curr_j == 0 or direction == 1):
|
|
mapA[curr_i - 1] = float('inf')
|
|
curr_i -= 1
|
|
else:
|
|
curr_j -= 1
|
|
|
|
return mapA
|
|
|
|
class ASRMappingTextV2:
|
|
def __init__(self):
|
|
# 設定權重：完全匹配=0, 同音字=0.2, 不匹配=1.0
|
|
self.match_cost = 0.0
|
|
self.pinyin_match_cost = 0.2
|
|
self.mismatch_cost = 1.0
|
|
|
|
def get_pinyin(self, char):
|
|
if pinyin:
|
|
res = pinyin(char, style=Style.NORMAL)
|
|
return res[0][0] if res else None
|
|
return None
|
|
|
|
def sub_cost(self, a, r):
|
|
if a == r:
|
|
return self.match_cost
|
|
# 同音字檢查 (處理 姓物/信物)
|
|
if self.get_pinyin(a) == self.get_pinyin(r):
|
|
return self.pinyin_match_cost
|
|
return self.mismatch_cost
|
|
|
|
def map_text(self, source_text: str, asr_data: str) -> list[int]:
|
|
n, m = len(source_text), len(asr_data)
|
|
if n == 0: return []
|
|
|
|
# 1. DP 表初始化
|
|
dp = [[0.0] * (m + 1) for _ in range(n + 1)]
|
|
bt = [[0] * (m + 1) for _ in range(n + 1)]
|
|
|
|
for i in range(1, n + 1):
|
|
dp[i][0] = i
|
|
bt[i][0] = 1 # Up
|
|
for j in range(1, m + 1):
|
|
dp[0][j] = j
|
|
bt[0][j] = 2 # Left
|
|
|
|
# 2. 填寫 DP 表 (計算編輯距離路徑)
|
|
for i in range(1, n + 1):
|
|
for j in range(1, m + 1):
|
|
cost = self.sub_cost(source_text[i-1], asr_data[j-1])
|
|
c_diag = dp[i-1][j-1] + cost
|
|
c_up = dp[i-1][j] + 1
|
|
c_left = dp[i][j-1] + 1
|
|
|
|
best = c_diag
|
|
dir_ = 0 # Diagonal
|
|
if c_up < best:
|
|
best = c_up
|
|
dir_ = 1
|
|
if c_left < best:
|
|
best = c_left
|
|
dir_ = 2
|
|
dp[i][j] = best
|
|
bt[i][j] = dir_
|
|
|
|
# 3. 第一次回溯：找出精確或近似匹配的點
|
|
mapA = [None] * n
|
|
curr_i, curr_j = n, m
|
|
while curr_i > 0 or curr_j > 0:
|
|
direction = bt[curr_i][curr_j]
|
|
if curr_i > 0 and curr_j > 0 and direction == 0:
|
|
# 只有當代價小於 1.0 (完全相同或同音) 時才算對應成功
|
|
if self.sub_cost(source_text[curr_i-1], asr_data[curr_j-1]) < 1.0:
|
|
mapA[curr_i - 1] = curr_j - 1
|
|
curr_i -= 1
|
|
curr_j -= 1
|
|
elif curr_i > 0 and (curr_j == 0 or direction == 1):
|
|
curr_i -= 1
|
|
else:
|
|
curr_j -= 1
|
|
|
|
# 4. 第二次掃描：補足缺失的索引 (處理斷尾與漏字)
|
|
known_indices = [idx for idx, val in enumerate(mapA) if val is not None]
|
|
|
|
if not known_indices:
|
|
# 極端情況：完全沒對上，給出均勻分佈
|
|
return [int(i * (m-1) / max(1, n-1)) for i in range(n)]
|
|
|
|
# A. 補齊開頭
|
|
first_known = known_indices[0]
|
|
for i in range(first_known - 1, -1, -1):
|
|
mapA[i] = max(0, mapA[i+1] - 1)
|
|
|
|
# B. 補齊中間與結尾
|
|
for k in range(len(known_indices)):
|
|
curr_i = known_indices[k]
|
|
|
|
if k == len(known_indices) - 1: # 處理斷尾處
|
|
for i in range(curr_i + 1, n):
|
|
# 遞增但不超過 ASR 長度
|
|
mapA[i] = min(m - 1, mapA[i-1] + 1)
|
|
else: # 處理兩個已知點之間的空隙 (插值)
|
|
next_i = known_indices[k+1]
|
|
gap_len = next_i - curr_i
|
|
val_diff = mapA[next_i] - mapA[curr_i]
|
|
for i in range(curr_i + 1, next_i):
|
|
# 使用線性分佈補齊缺失索引
|
|
mapA[i] = mapA[curr_i] + int((i - curr_i) * val_diff / gap_len)
|
|
|
|
return mapA
|
|
|
|
class ASRMappingTextV3:
|
|
def __init__(self):
|
|
# 權重設定
|
|
self.COST_MATCH = 0.0 # 完全匹配
|
|
self.COST_PINYIN = 0.4 # 拼音相同 (如: 引离/云璃, 性物/信物)
|
|
self.COST_MISMATCH = 1.2 # 完全不同 (調高一點，強迫走對角線)
|
|
|
|
def _get_pinyin(self, char):
|
|
"""取得單個漢字的拼音"""
|
|
if not re.match(r'[\u4e00-\u9fa5]', char):
|
|
return char
|
|
res = pinyin(char, style=Style.NORMAL)
|
|
return res[0][0] if res else char
|
|
|
|
def _is_valid(self, char):
|
|
"""判斷是否為有效字元（排除標點符號）"""
|
|
return re.match(r'[\u4e00-\u9fa5a-zA-Z0-9]', char) is not None
|
|
|
|
def sub_cost(self, a, r):
|
|
"""計算替換代價"""
|
|
if a == r:
|
|
return self.COST_MATCH
|
|
if self._get_pinyin(a) == self._get_pinyin(r):
|
|
return self.COST_PINYIN
|
|
return self.COST_MISMATCH
|
|
|
|
def map_text(self, source_text: str, asr_data: str) -> list[int]:
|
|
# 1. 預處理：分離標點，只對有效字元做 DP
|
|
src_info = [(i, c) for i, c in enumerate(source_text) if self._is_valid(c)]
|
|
src_chars = [x[1] for x in src_info]
|
|
|
|
n = len(src_chars)
|
|
m = len(asr_data)
|
|
|
|
if n == 0: return [0] * len(source_text)
|
|
if m == 0: return [0] * len(source_text)
|
|
|
|
# 2. DP 表初始化
|
|
dp = [[0.0] * (m + 1) for _ in range(n + 1)]
|
|
bt = [[0] * (m + 1) for _ in range(n + 1)]
|
|
|
|
for i in range(1, n + 1):
|
|
dp[i][0] = i * 1.0
|
|
bt[i][0] = 1 # Up
|
|
for j in range(1, m + 1):
|
|
dp[0][j] = j * 1.0
|
|
bt[0][j] = 2 # Left
|
|
|
|
# 3. 填寫 DP 表
|
|
for i in range(1, n + 1):
|
|
for j in range(1, m + 1):
|
|
cost = self.sub_cost(src_chars[i-1], asr_data[j-1])
|
|
c_diag = dp[i-1][j-1] + cost
|
|
c_up = dp[i-1][j] + 1.0
|
|
c_left = dp[i][j-1] + 1.0
|
|
|
|
best = c_diag
|
|
dir_ = 0
|
|
if c_up < best:
|
|
best = c_up
|
|
dir_ = 1
|
|
if c_left < best:
|
|
best = c_left
|
|
dir_ = 2
|
|
dp[i][j] = best
|
|
bt[i][j] = dir_
|
|
|
|
# 4. 第一次回溯：找出對應點
|
|
raw_map = [None] * n
|
|
curr_i, curr_j = n, m
|
|
while curr_i > 0 or curr_j > 0:
|
|
direction = bt[curr_i][curr_j]
|
|
if curr_i > 0 and curr_j > 0 and direction == 0:
|
|
# 只要不是完全不匹配(1.2)，就視為對應成功
|
|
if self.sub_cost(src_chars[curr_i-1], asr_data[curr_j-1]) < self.COST_MISMATCH:
|
|
raw_map[curr_i - 1] = curr_j - 1
|
|
curr_i -= 1
|
|
curr_j -= 1
|
|
elif curr_i > 0 and (curr_j == 0 or direction == 1):
|
|
curr_i -= 1
|
|
else:
|
|
curr_j -= 1
|
|
|
|
# 5. 線性插值補全（解決開頭錯誤與結尾斷字）
|
|
known = [i for i, v in enumerate(raw_map) if v is not None]
|
|
|
|
if not known:
|
|
filled_map = [int(i * (m-1) / (n-1)) if n > 1 else 0 for i in range(n)]
|
|
else:
|
|
filled_map = list(raw_map)
|
|
# 補頭
|
|
first_k = known[0]
|
|
for i in range(first_k - 1, -1, -1):
|
|
filled_map[i] = max(0, filled_map[i+1] - 1)
|
|
# 補中間
|
|
for k in range(len(known) - 1):
|
|
idx_start, idx_end = known[k], known[k+1]
|
|
val_start, val_end = filled_map[idx_start], filled_map[idx_end]
|
|
for i in range(idx_start + 1, idx_end):
|
|
ratio = (i - idx_start) / (idx_end - idx_start)
|
|
filled_map[i] = int(val_start + ratio * (val_end - val_start))
|
|
# 補尾
|
|
last_k = known[-1]
|
|
for i in range(last_k + 1, n):
|
|
filled_map[i] = min(m - 1, filled_map[i-1] + 1)
|
|
|
|
# 6. 映射回原始帶標點的字串
|
|
final_result = [0] * len(source_text)
|
|
valid_idx = 0
|
|
for i, char in enumerate(source_text):
|
|
if self._is_valid(char):
|
|
final_result[i] = filled_map[valid_idx]
|
|
valid_idx += 1
|
|
else:
|
|
# 標點符號跟隨前一個有效字的索引
|
|
final_result[i] = final_result[i-1] if i > 0 else 0
|
|
|
|
return final_result
|
|
|
|
class ASRMappingTextV4:
|
|
def __init__(self):
|
|
# 權重設定：同音字 0.4, 完全不同 1.5
|
|
self.COST_MATCH = 0.0
|
|
self.COST_PINYIN = 0.4
|
|
self.COST_MISMATCH = 1.5
|
|
|
|
def _get_pinyin(self, char):
|
|
if not re.match(r'[\u4e00-\u9fa5]', char): return char
|
|
res = pinyin(char, style=Style.NORMAL)
|
|
return res[0][0] if res else char
|
|
|
|
def _is_valid(self, char):
|
|
return re.match(r'[\u4e00-\u9fa5a-zA-Z0-9]', char) is not None
|
|
|
|
def sub_cost(self, a, r):
|
|
if a == r: return self.COST_MATCH
|
|
if self._get_pinyin(a) == self._get_pinyin(r): return self.COST_PINYIN
|
|
return self.COST_MISMATCH
|
|
|
|
def map_text(self, source_text: str, asr_segments: list) -> list:
|
|
# 1. 串聯 ASR 片段並建立「字元 -> 時間」映射表
|
|
asr_full_text = ""
|
|
char_times = [] # 儲存每個 ASR 字元對應的 (start, end)
|
|
|
|
for seg in asr_segments:
|
|
val = seg['value'].strip()
|
|
if not val: continue
|
|
# 計算該片段內每個字的預估時間 (均分)
|
|
duration = (seg['end'] - seg['start']) / len(val)
|
|
for i, char in enumerate(val):
|
|
asr_full_text += char
|
|
char_times.append({
|
|
'start': seg['start'] + i * duration,
|
|
'end': seg['start'] + (i + 1) * duration
|
|
})
|
|
|
|
# 2. 提取 Source 有效字元位置
|
|
src_valid_chars = [c for c in source_text if self._is_valid(c)]
|
|
n, m = len(src_valid_chars), len(asr_full_text)
|
|
if n == 0 or m == 0: return []
|
|
|
|
# 3. DP 矩陣計算
|
|
dp = [[0.0] * (m + 1) for _ in range(n + 1)]
|
|
bt = [[0] * (m + 1) for _ in range(n + 1)]
|
|
|
|
for i in range(1, n + 1): dp[i][0] = i * 1.0; bt[i][0] = 1
|
|
for j in range(1, m + 1): dp[0][j] = j * 1.0; bt[0][j] = 2
|
|
|
|
for i in range(1, n + 1):
|
|
for j in range(1, m + 1):
|
|
cost = self.sub_cost(src_valid_chars[i-1], asr_full_text[j-1])
|
|
c_diag = dp[i-1][j-1] + cost
|
|
c_up = dp[i-1][j] + 1.0
|
|
c_left = dp[i][j-1] + 1.0
|
|
|
|
best = c_diag; dir_ = 0
|
|
if c_up < best: best = c_up; dir_ = 1
|
|
if c_left < best: best = c_left; dir_ = 2
|
|
dp[i][j] = best; bt[i][j] = dir_
|
|
|
|
# 4. 回溯：找出初步對齊 (僅保留代價低於 1.0 的對應點)
|
|
raw_map = [None] * n
|
|
curr_i, curr_j = n, m
|
|
while curr_i > 0 or curr_j > 0:
|
|
direction = bt[curr_i][curr_j]
|
|
if curr_i > 0 and curr_j > 0 and direction == 0:
|
|
if self.sub_cost(src_valid_chars[curr_i-1], asr_full_text[curr_j-1]) < 1.0:
|
|
raw_map[curr_i - 1] = curr_j - 1
|
|
curr_i -= 1; curr_j -= 1
|
|
elif curr_i > 0 and (curr_j == 0 or direction == 1):
|
|
curr_i -= 1
|
|
else:
|
|
curr_j -= 1
|
|
|
|
# 5. 強力插值補全 (處理結尾「死」或開頭「景軒」)
|
|
known = [i for i, v in enumerate(raw_map) if v is not None]
|
|
if not known:
|
|
# 沒對上任何字，強制線性分佈
|
|
final_indices = [int(i * (m-1) / (n-1)) if n > 1 else 0 for i in range(n)]
|
|
else:
|
|
final_indices = list(raw_map)
|
|
# 補頭：若第一個字沒對上，從對上的點往回推
|
|
for i in range(known[0] - 1, -1, -1):
|
|
final_indices[i] = max(0, final_indices[i+1] - 1)
|
|
# 補中間：線性插值
|
|
for k in range(len(known) - 1):
|
|
s, e = known[k], known[k+1]
|
|
v_s, v_e = final_indices[s], final_indices[e]
|
|
for i in range(s + 1, e):
|
|
final_indices[i] = v_s + int((i - s) * (v_e - v_s) / (e - s))
|
|
# 補尾：重點修正！若最後一個字沒對上，強制對應到 ASR 最後一個索引
|
|
for i in range(known[-1] + 1, n):
|
|
# 這裡不再只是 +1，而是確保最後一個字一定指向 m-1
|
|
dist_to_end = n - 1 - known[-1]
|
|
val_dist = (m - 1) - final_indices[known[-1]]
|
|
final_indices[i] = final_indices[i-1] + max(1, int(val_dist / dist_to_end))
|
|
final_indices[i] = min(final_indices[i], m - 1)
|
|
|
|
# 強制最後一個有效字元指向 ASR 的末尾字
|
|
final_indices[-1] = m - 1
|
|
|
|
# 6. 組裝回原本包含標點符號的清單
|
|
final_mapping = []
|
|
valid_ptr = 0
|
|
for char in source_text:
|
|
if self._is_valid(char):
|
|
idx = final_indices[valid_ptr]
|
|
final_mapping.append({
|
|
'char': char,
|
|
'asr_idx': idx,
|
|
'time': char_times[idx]
|
|
})
|
|
valid_ptr += 1
|
|
else:
|
|
# 標點符號跟隨前一字時間
|
|
prev_time = final_mapping[-1]['time'] if final_mapping else char_times[0]
|
|
final_mapping.append({'char': char, 'asr_idx': -1, 'time': prev_time})
|
|
|
|
return final_mapping |