1f64c87d0e
- 任务化清洗流程：Task/TaskManager + BatchProcessor 三方法
- 数据目录规范化：data/{db,sources,backup}
- CLI 入口移进包，注册 jclean 命令
- 工作流测试驱动（tests/test_cleaner_workflow.py）
- 40 测试通过
235 lines
8.7 KiB
Python
235 lines
8.7 KiB
Python
"""
|
||
校验手工维护的日语词表数据。
|
||
- 拼音校验：用 pypinyin 检查每个汉字拼音是否在其合法读音集合内（考虑多音字）
|
||
- 日语读音校验：
|
||
1) 整词级：用 jamdict 真词典查整词，看假名是否为该词已知读音
|
||
2) 单字级：用 kanjidic 的音读/训读，检查每个汉字的假名是否合理
|
||
- 结构校验：三段是否齐全、汉字数/假名数/拼音数是否对齐
|
||
|
||
用法: python validate_data.py <file1> [file2 ...]
|
||
结果写入 validation_report.txt（UTF-8），避免控制台乱码。
|
||
"""
|
||
import sys
|
||
import re
|
||
from pathlib import Path
|
||
from collections import defaultdict
|
||
|
||
from pypinyin import pinyin, Style
|
||
from jamdict import Jamdict
|
||
|
||
import os
|
||
# 优先用本地副本（网络路径上的 sqlite 查询极慢）
|
||
_LOCAL_DB = r"C:\Users\panli\jamdict_local.db"
|
||
_NET_DB = r"\\192.168.3.200\work\workspace\python\japanese\src\pl_japanese\jamdict.db\jamdict.db"
|
||
DB = _LOCAL_DB if os.path.isfile(_LOCAL_DB) else _NET_DB
|
||
jam = Jamdict(db_file=DB)
|
||
|
||
HANZI = re.compile(r'[\u4e00-\u9fff]')
|
||
|
||
# 片假名 -> 平假名
|
||
def kata_to_hira(s: str) -> str:
|
||
out = []
|
||
for ch in s:
|
||
code = ord(ch)
|
||
if 0x30A1 <= code <= 0x30F6:
|
||
out.append(chr(code - 0x60))
|
||
else:
|
||
out.append(ch)
|
||
return ''.join(out)
|
||
|
||
def clean_reading(s: str) -> str:
|
||
# 去掉 kanjidic 训读里的送り仮名标记：- . 及其后缀
|
||
s = s.replace('-', '')
|
||
if '.' in s:
|
||
s = s.split('.')[0]
|
||
return kata_to_hira(s).strip()
|
||
|
||
# ---- 缓存 ----
|
||
_pinyin_cache = {}
|
||
_kanji_reading_cache = {}
|
||
_word_cache = {}
|
||
|
||
def valid_pinyins(char: str):
|
||
if char in _pinyin_cache:
|
||
return _pinyin_cache[char]
|
||
res = pinyin(char, style=Style.NORMAL, heteronym=True)
|
||
s = set(res[0]) if res else set()
|
||
_pinyin_cache[char] = s
|
||
return s
|
||
|
||
def kanji_readings(char: str):
|
||
"""返回该汉字所有合法假名读音（平假名，含音读+训读，含常见连浊近似）"""
|
||
if char in _kanji_reading_cache:
|
||
return _kanji_reading_cache[char]
|
||
readings = set()
|
||
try:
|
||
r = jam.lookup(char)
|
||
for c in r.chars:
|
||
for g in c.rm_groups:
|
||
for on in g.on_readings:
|
||
readings.add(clean_reading(str(on)))
|
||
for kun in g.kun_readings:
|
||
readings.add(clean_reading(str(kun)))
|
||
except Exception:
|
||
pass
|
||
readings.discard('')
|
||
_kanji_reading_cache[char] = readings
|
||
return readings
|
||
|
||
def word_readings(word: str):
|
||
"""整词在词典里的已知假名读音集合"""
|
||
if word in _word_cache:
|
||
return _word_cache[word]
|
||
readings = set()
|
||
try:
|
||
r = jam.lookup(word)
|
||
for e in r.entries:
|
||
for k in e.kana_forms:
|
||
readings.add(kata_to_hira(k.text))
|
||
except Exception:
|
||
pass
|
||
_word_cache[word] = readings
|
||
return readings
|
||
|
||
# 连浊/促音的宽松匹配：允许清浊互换、结尾促音
|
||
DAKUTEN = {
|
||
'か':'が','き':'ぎ','く':'ぐ','け':'げ','こ':'ご',
|
||
'さ':'ざ','し':'じ','す':'ず','せ':'ぜ','そ':'ぞ',
|
||
'た':'だ','ち':'ぢ','つ':'づ','て':'で','と':'ど',
|
||
'は':'ば','ひ':'び','ふ':'ぶ','へ':'べ','ほ':'ぼ',
|
||
}
|
||
HANDAKU = {'は':'ぱ','ひ':'ぴ','ふ':'ぷ','へ':'ぺ','ほ':'ぽ'}
|
||
|
||
def reading_variants(r: str):
|
||
"""给一个字典读音生成宽松变体：首音连浊/半浊、结尾长音、结尾促音化"""
|
||
vs = {r}
|
||
if r:
|
||
first = r[0]
|
||
if first in DAKUTEN:
|
||
vs.add(DAKUTEN[first] + r[1:])
|
||
if first in HANDAKU:
|
||
vs.add(HANDAKU[first] + r[1:])
|
||
# 结尾促音（っ）：如 いち->いっ
|
||
if len(r) >= 2:
|
||
vs.add(r[:-1] + 'っ')
|
||
# 结尾长音：こう 之类，允许省略う
|
||
if r.endswith('う'):
|
||
vs.add(r[:-1])
|
||
return vs
|
||
|
||
def kana_matches_kanji(kana_seg: str, char: str) -> bool:
|
||
"""该假名片段是否匹配该汉字的某个合法读音（含宽松变体）"""
|
||
valid = kanji_readings(char)
|
||
if not valid:
|
||
return None # 词典无此字信息，无法判断
|
||
candidates = set()
|
||
for r in valid:
|
||
candidates |= reading_variants(r)
|
||
return kana_seg in candidates
|
||
|
||
def analyze_file(path: str, issues: dict):
|
||
p = Path(path)
|
||
name = p.name
|
||
with open(p, 'r', encoding='utf-8') as f:
|
||
for lineno, raw in enumerate(f, 1):
|
||
line = raw.rstrip('\n').strip()
|
||
if not line or line.startswith('#'):
|
||
continue
|
||
# 归一全角冒号
|
||
norm = line.replace('：', ':')
|
||
parts = norm.split(':')
|
||
if len(parts) < 2:
|
||
continue
|
||
kanji_part = parts[0]
|
||
kana_part = parts[1]
|
||
pinyin_part = parts[2] if len(parts) >= 3 else ''
|
||
|
||
kanji_tokens = kanji_part.split('|')
|
||
kana_tokens = kana_part.split('|')
|
||
pinyin_tokens = pinyin_part.split('|') if pinyin_part else []
|
||
|
||
word = ''.join(kanji_tokens)
|
||
word_kana = ''.join(kana_tokens)
|
||
|
||
# ---- 结构校验 ----
|
||
if pinyin_tokens and len({len(kanji_tokens), len(kana_tokens), len(pinyin_tokens)}) != 1:
|
||
issues['struct'].append(
|
||
f"{name}:{lineno} 分段数不齐 汉字{len(kanji_tokens)}/假名{len(kana_tokens)}/拼音{len(pinyin_tokens)} | {line}")
|
||
|
||
# ---- 拼音校验（逐字，仅汉字）----
|
||
if pinyin_tokens and len(kanji_tokens) == len(pinyin_tokens):
|
||
for ch, py in zip(kanji_tokens, pinyin_tokens):
|
||
if not HANZI.search(ch):
|
||
if py.strip():
|
||
issues['pinyin'].append(
|
||
f"{name}:{lineno} 非汉字'{ch}'却标了拼音'{py}' | {line}")
|
||
continue
|
||
if not py.strip():
|
||
issues['pinyin'].append(
|
||
f"{name}:{lineno} 汉字'{ch}'缺拼音 | {line}")
|
||
continue
|
||
valid = valid_pinyins(ch)
|
||
if valid and py.strip().lower() not in {v.lower() for v in valid}:
|
||
issues['pinyin'].append(
|
||
f"{name}:{lineno} '{ch}' 拼音 '{py}' 疑误，正确应为 {sorted(valid)} | {line}")
|
||
|
||
# ---- 日语整词校验 ----
|
||
wr = word_readings(word)
|
||
if wr and word_kana not in wr:
|
||
issues['word'].append(
|
||
f"{name}:{lineno} 词'{word}'读音'{word_kana}'不在词典 {sorted(wr)} | {line}")
|
||
|
||
# ---- 日语单字校验（逐字）----
|
||
if len(kanji_tokens) == len(kana_tokens):
|
||
for ch, ka in zip(kanji_tokens, kana_tokens):
|
||
if not HANZI.search(ch):
|
||
continue # 假名对假名，跳过
|
||
m = kana_matches_kanji(ka, ch)
|
||
if m is False:
|
||
valid = sorted(kanji_readings(ch))
|
||
issues['kanji'].append(
|
||
f"{name}:{lineno} '{ch}'读'{ka}'疑误，音训读为 {valid} | {line}")
|
||
|
||
def main():
|
||
files = sys.argv[1:]
|
||
if not files:
|
||
print("usage: python validate_data.py <file...>")
|
||
return
|
||
issues = defaultdict(list)
|
||
for f in files:
|
||
analyze_file(f, issues)
|
||
|
||
out = Path(r"\\192.168.3.200\work\workspace\python\japanese\validation_report.txt")
|
||
lines = []
|
||
lines.append("=" * 70)
|
||
lines.append("日语词表数据校验报告")
|
||
lines.append("=" * 70)
|
||
lines.append("")
|
||
lines.append("说明：以下为疑似错误，需人工确认。词典(jamdict/kanjidic)未必收录所有")
|
||
lines.append("专有名词、姓氏、古语读法，故'疑误'不等于一定错，请结合语境判断。")
|
||
lines.append("")
|
||
|
||
order = [
|
||
('struct', '一、结构问题（汉字/假名/拼音分段数不一致）'),
|
||
('pinyin', '二、汉语拼音疑似错误'),
|
||
('word', '三、整词日语读音疑似错误（词典有该词但读音不符）'),
|
||
('kanji', '四、单字日语读音疑似错误（音读/训读均不匹配）'),
|
||
]
|
||
for key, title in order:
|
||
lst = issues[key]
|
||
lines.append("-" * 70)
|
||
lines.append(f"【{title}】 共 {len(lst)} 条")
|
||
lines.append("-" * 70)
|
||
if not lst:
|
||
lines.append(" （无）")
|
||
else:
|
||
lines.extend(" " + x for x in lst)
|
||
lines.append("")
|
||
|
||
out.write_text('\n'.join(lines), encoding='utf-8')
|
||
print(f"done. issues: struct={len(issues['struct'])} pinyin={len(issues['pinyin'])} word={len(issues['word'])} kanji={len(issues['kanji'])}")
|
||
print(f"report -> {out}")
|
||
|
||
if __name__ == '__main__':
|
||
main()
|