Merge pull request #868 from lukhnos/postprocess

postprocess: 允許在詞庫後製過程中，手動調整常見問題詞的分數

Author: lukhnos

.github/workflows/continuous-build-data.yml

@@ -21,6 +21,9 @@ jobs:
       - uses: actions/setup-python@v6
         with:
           python-version: '3.12'
+      - name: Unit test the curation library
+        run: python3 -m unittest
+        working-directory: Source/Data
       - name: Test data files
         run: make check
         working-directory: Source/Data

Source/Data/.gitignore

@@ -3,3 +3,4 @@ associated-phrases-v2.txt
 cand.occ
 data-plain-bpmf.txt
 data.txt
+data-raw.txt

Source/Data/Makefile

@@ -6,15 +6,19 @@ all: data.txt data-plain-bpmf.txt associated-phrases-v2.txt
 
 install: all
 
-data-plain-bpmf.txt: curation/compilers/plain_bpmf_compiler.py BPMFBase.txt BPMFPunctuations.txt
+data-plain-bpmf.txt: curation/compilers/plain_bpmf_compiler.py \
+	curation/compilers/compiler_utils.py \
+	BPMFBase.txt BPMFPunctuations.txt
 	$(PYTHON) -m curation.compilers.plain_bpmf_compiler \
 		--bpmf_base BPMFBase.txt \
 		--punctuations BPMFPunctuations.txt \
 		--output data-plain-bpmf.txt
 
-data.txt: curation/compilers/main_compiler.py BPMFBase.txt BPMFMappings.txt BPMFPunctuations.txt \
-          PhraseFreq.txt phrase.occ Symbols.txt Macros.txt\
-          heterophony1.list heterophony2.list heterophony3.list
+data-raw.txt: curation/compilers/main_compiler.py \
+	curation/compilers/compiler_utils.py \
+	BPMFBase.txt BPMFMappings.txt BPMFPunctuations.txt \
+	PhraseFreq.txt phrase.occ Symbols.txt Macros.txt \
+	heterophony1.list heterophony2.list heterophony3.list
 	$(PYTHON) -m curation.compilers.main_compiler \
 		--heterophony1 heterophony1.list \
 		--heterophony2 heterophony2.list \
@@ -25,6 +29,12 @@ data.txt: curation/compilers/main_compiler.py BPMFBase.txt BPMFMappings.txt BPMF
 		--punctuations BPMFPunctuations.txt \
 		--symbols Symbols.txt \
 		--macros Macros.txt \
+		--output data-raw.txt
+
+data.txt: curation/compilers/postprocess.py curation/mandarin/grid.py data-raw.txt Postprocess.txt
+	$(PYTHON) -m curation.compilers.postprocess \
+		--input data-raw.txt \
+		--directive Postprocess.txt \
 		--output data.txt
 
 associated-phrases-v2.txt: data.txt curation/builders/phrase_deriver.py associated-punctuation.txt
@@ -101,3 +111,13 @@ _mycodecheck:
 	@for myfile in curation/*/*.py scripts/*.py; do \
             pycodestyle --ignore=E501,E722,E701,E221,W605 $$myfile; \
         done
+
+format:
+	black \
+		curation/mandarin/*.py \
+		curation/compilers/postprocess.py \
+		tests/*.py
+
+typecheck:
+	mypy \
+		curation/compilers/postprocess.py

Source/Data/Postprocess.txt

@@ -0,0 +1,82 @@
+# 詞庫後製設定檔
+
+# epsilon 用來設定 override 對數分數的 ε 值。web 版精確度只有小數點後三位，因
+# 此建議 epsilon 設到這個數字
+
+epsilon 0.001
+
+
+# before 用來確認後製修改之前的詞庫狀態
+
+before ㄓㄜˋ-ㄧㄤˋ-ㄇㄟˊ-ㄕˋ-ㄐㄧㄡˋ-ㄏㄠˇ 這樣-沒-是-就好
+
+
+# promote-over-single-syllables 的意思是，給定多字詞，這個多字詞的分數，應該
+# 高於同數量同音單字的加總。例如打「沒事」但出現「沒是」，是因為在 unigram
+# model 下，P_log(沒) + P_log(是) > P_log(沒事)
+
+promote-over-single-syllables 沒事 ㄇㄟˊ-ㄕˋ
+
+
+# assert 用於確認修改之後的詞庫狀態
+
+assert ㄇㄟˊ-ㄕˋ 沒事
+assert ㄓㄜˋ-ㄧㄤˋ-ㄇㄟˊ-ㄕˋ-ㄐㄧㄡˋ-ㄏㄠˇ 這樣-沒事-就好
+
+promote-over-single-syllables 好險 ㄏㄠˇ-ㄒㄧㄢˇ
+assert ㄏㄠˇ-ㄒㄧㄢˇ-ㄇㄟˊ-ㄕˋ-ㄦˊ-ㄑㄧㄝˇ-ㄏㄠˇ-ㄒㄧㄢˇ-ㄒㄧㄢˇ-ㄕˋ-ㄑㄧˋ-ㄇㄟˊ-ㄏㄨㄞˋ 好險-沒事-而且-好險-顯示器-沒-壞
+
+
+# promote-over-peers 的意思是，如果有許多同字數的同音詞，把指定的詞給列為最高
+# 順位。例如首字理論上可念成一聲（但實際則否）的「一顆」「一棵」「一科」這三
+# 個詞跟「醫科」有衝突，在語用上「ㄧ ㄎㄜ」應該出「醫科」為首。
+
+promote-over-peers 醫科 ㄧ-ㄎㄜ
+
+# 這個修改讓「一顆」的「一」只能用四聲讀，否則結果永遠出現「醫科」為先
+assert ㄧˋ-ㄎㄜ-ㄊㄤˊ-ㄍㄨㄛˇ 一顆-糖果
+assert ㄧ-ㄎㄜ-ㄉㄚˋ-ㄒㄩㄝˊ 醫科-大學
+assert ㄉㄨˊ-ㄧ-ㄎㄜ 讀-醫科
+assert ㄧ-ㄎㄜ-ㄕㄥ 醫科-生
+
+# ㄎㄜ 與右鄰成詞（科目）時，「一」保留，promote 不覆蓋
+assert ㄧ-ㄎㄜ-ㄇㄨˋ 一-科目
+
+# ㄧ 與左鄰成詞（第一、這一）時，「一」亦保留
+assert ㄉㄧˋ-ㄧ-ㄎㄜ 第一-科
+assert ㄓㄜˋ-ㄧ-ㄎㄜ 這一-科
+
+# 副作用：量詞「一顆」「一棵」左右皆無法成詞時，誤判為「醫科」；
+# 日後若能依上下文正確判斷量詞，此二 assert 將失敗以示提醒。
+assert ㄔ-ㄧ-ㄎㄜ 吃-醫科
+assert ㄧ-ㄎㄜ-ㄘㄞˋ 醫科-菜
+
+# 若念為四聲則不衝突，但此處「ㄧˋ ㄎㄜ」因為組詞模型欠缺判別下文的能力，
+# 首選仍為「一顆」
+assert ㄔ-ㄧˋ-ㄎㄜ 吃-一顆
+assert ㄧˋ-ㄎㄜ-ㄘㄞˋ 一顆-菜
+
+
+# 「裡裡外外」已經成詞，所以不需要再 promote 「裡裡」
+# promote-over-peers 裡裡 ㄌㄧˇ-ㄌㄧˇ
+# assert ㄌㄧˇ-ㄌㄧˇ-ㄨㄞˋ-ㄨㄞˋ 裡裡外外
+
+promote-over-single-syllables 依舊 ㄧ-ㄐㄧㄡˋ
+assert ㄓㄜˋ-ㄨㄣˋ-ㄊㄧˊ-ㄧ-ㄐㄧㄡˋ-ㄘㄨㄣˊ-ㄗㄞˋ 這-問題-依舊-存在
+
+# 這個問題目前還不能解決
+#
+# promote-over-single-syllables 依舊在 ㄧ-ㄐㄧㄡˋ-ㄗㄞˋ
+#
+# 提升後，斷詞仍然是「青山-一-就在」
+# assert ㄑㄧㄥ-ㄕㄢ-ㄧ-ㄐㄧㄡˋ-ㄗㄞˋ 青山-依舊在
+
+promote-over-single-syllables 中醫 ㄓㄨㄥ-ㄧ
+promote-over-single-syllables 西醫 ㄒㄧ-ㄧ
+assert ㄓㄨㄥ-ㄧ-ㄩˇ-ㄒㄧ-ㄧ 中醫-與-西醫
+
+# 以下兩例用來確保修改後，不會影響其他同音但詞界不同詞
+# 台中一中、中一中為完整長詞
+assert ㄓㄨㄥ-ㄧ-ㄓㄨㄥ 中一中
+assert ㄊㄞˊ-ㄓㄨㄥ-ㄧ-ㄓㄨㄥ 台中一中
+

Source/Data/curation/compilers/postprocess.py

@@ -0,0 +1,219 @@
+import argparse
+from .compiler_utils import HEADER
+from ..mandarin.grid import most_plausible_walk
+import sys
+import unittest
+
+errors: list[tuple[int, str]] = []
+warnings: list[tuple[int, str]] = []
+epsilon = 0.0001
+
+
+def set_epsilon(e):
+    global epsilon
+    epsilon = e
+
+
+def accrue_error(lineno, err):
+    errors.append((lineno, err))
+
+
+def accrue_warning(lineno, warning):
+    warnings.append((lineno, warning))
+
+
+def show_errors_and_warnings():
+    combined = [(lineno, f"error (line {lineno}): {msg}\n") for lineno, msg in errors]
+    combined += [
+        (lineno, f"warning (line {lineno}): {msg}\n") for lineno, msg in warnings
+    ]
+
+    combined = sorted(combined, key=lambda x: x[0])
+
+    for _, msg in combined:
+        sys.stderr.write(msg)
+
+
+def segmented_values(nodes):
+    return "-".join(n.value for n in nodes)
+
+
+def find_top_unigram_in_lm(lm, reading):
+    if reading not in lm:
+        return None
+
+    unigrams = lm[reading]
+    v, s = unigrams[0]
+    return (v, float(s))
+
+
+def find_score_in_lm(lm, reading, value):
+    if reading not in lm:
+        return None
+
+    unigrams = lm[reading]
+    for unigram in unigrams:
+        if unigram[0] == value:
+            return float(unigram[1])
+
+    return None
+
+
+def replace_score_in_lm(lm, reading, value, new_score):
+    if reading not in lm:
+        raise ValueError(f"reading {reading} not in language model!")
+
+    unigrams = lm[reading]
+
+    has_replacement = False
+    new_unigrams = []
+    for unigram in unigrams:
+        uv, us = unigram
+        if uv == value:
+            if type(us) == str:
+                new_unigrams.append((uv, str(new_score)))
+            else:
+                new_unigrams.append((uv, new_score))
+            has_replacement = True
+        else:
+            new_unigrams.append((uv, us))
+
+    if not has_replacement:
+        raise ValueError("reading:value %s:%s not in LM" % (reading, value))
+
+    unigrams = sorted(new_unigrams, key=lambda x: float(x[1]), reverse=True)
+    lm[reading] = unigrams
+    return True
+
+
+def promote_over_single_syllables(lineno, lm, value, reading):
+
+    readings = reading.split("-")
+
+    if len(value) != len(readings):
+        return accrue_error(lineno, "number of codepoints don't match readings")
+
+    our_score = find_score_in_lm(lm, reading, value)
+    if not our_score:
+        return accrue_error(lineno, f"reading:value {reading}:{value} not in LM")
+
+    # validate data
+    unigrams = [find_top_unigram_in_lm(lm, r) for r in readings]
+    if not all(unigrams):
+        return accrue_error(lineno, "cannot find all single-syllable readings")
+
+    their_scores = sum(u[1] for u in unigrams)
+
+    if their_scores <= our_score:
+        return accrue_error(lineno, "no need to promote")
+
+    our_score = their_scores + epsilon
+    return replace_score_in_lm(lm, reading, value, our_score)
+
+
+def promote_over_peers(lineno, lm, value, reading):
+    our_score = find_score_in_lm(lm, reading, value)
+    if not our_score:
+        return accrue_error(lineno, f"reading:value {reading}:{value} not in LM")
+
+    top_gram = find_top_unigram_in_lm(lm, reading)
+    if not top_gram:
+        return accrue_error(lineno, f"no unigrams found for reading: {reading}")
+
+    top_value, top_score = top_gram
+    if top_value == value:
+        return accrue_error(lineno, f"value {value} already top among peers")
+
+    our_score = float(top_score) + epsilon
+    return replace_score_in_lm(lm, reading, value, our_score)
+
+
+def run_assert(lineno, lm, readings, expected, warn_only=False):
+    nodes = most_plausible_walk(readings.split("-"), lm)
+    result = segmented_values(nodes)
+
+    if result != expected:
+        if warn_only:
+            return accrue_warning(lineno, f"expected: {expected}, actual: {result}")
+        else:
+            return accrue_error(lineno, f"expected: {expected}, actual: {result}")
+
+
+def postprocess(input, directive, output):
+    lm = {}
+
+    with open(input) as f:
+        lines = f.readlines()
+
+        for line in lines[1:]:
+            # don't use bare split() since it also splits full-width spaces
+            r, v, s = line.strip().split(" ")
+
+            if r in lm:
+                lm[r].append((v, s))
+            else:
+                lm[r] = [(v, s)]
+
+    with open(directive) as f:
+        lineno = 0
+
+        for line in f:
+            lineno += 1
+
+            line = line.strip()
+            if not line:
+                continue
+            if line.startswith("#"):
+                continue
+
+            elements = line.split()
+            if elements[0] == "assert":
+                readings = elements[1]
+                expected = elements[2]
+                run_assert(lineno, lm, readings, expected)
+            elif elements[0] == "before":
+                readings = elements[1]
+                expected = elements[2]
+                run_assert(lineno, lm, readings, expected, warn_only=True)
+            elif elements[0] == "promote-over-single-syllables":
+                value = elements[1]
+                reading = elements[2]
+                promote_over_single_syllables(lineno, lm, value, reading)
+            elif elements[0] == "promote-over-peers":
+                value = elements[1]
+                reading = elements[2]
+                promote_over_peers(lineno, lm, value, reading)
+            elif elements[0] == "epsilon":
+                set_epsilon(float(elements[1]))
+            else:
+                accrue_error(lineno, f"unknown command: {elements[0]}")
+
+    if errors or warnings:
+        show_errors_and_warnings()
+
+    if warnings:
+        print("%d warning(s) found" % len(warnings))
+
+    if errors:
+        print("%d error(s) found" % len(errors))
+        sys.exit(1)
+
+    with open(output, "w") as f:
+        f.write(HEADER)
+
+        for r in sorted(lm.keys(), key=lambda x: x.encode()):
+            for v, s in lm[r]:
+                f.write("%s %s %s\n" % (r, v, s))
+
+
+def main():
+    parser = argparse.ArgumentParser(description="postprocess compiled phrase database")
+    parser.add_argument("--input", required=True, help="path to source data")
+    parser.add_argument("--directive", required=True, help="path to directive file")
+    parser.add_argument("--output", required=True, help="path to postprocessed output")
+    args = parser.parse_args()
+    postprocess(input=args.input, directive=args.directive, output=args.output)
+
+
+if __name__ == "__main__":
+    main()