ef3df73166
清洗管线重构为严格三层架构：
- tango_analyser.py（底层：单词分析）
- task_processor.py（中层：文件 I/O、桶管理）
- workflow.py（顶层：状态机、任务推进）
- 移除旧的 batch_processor.py
新增配置文件系统：
- config.py：TOML 配置，相对路径相对配置文件目录解析
- 查找优先级 --config > cwd > 项目根 > ~ > 默认值
- count=None 语义为处理到文件末尾
项目目录整理：
- 根脚本归档到 scripts/analysis 与 scripts/legacy
- 文档归档到 docs/{design,history,analysis}
- 临时报告移到 reports/（已 gitignore）
文档质量：
- 新增 .markdownlint.json 与 scripts/mdlint.cmd
- 修复全部 14 个 md 文件的 markdownlint 警告
测试：74 passed（8 cleaner + 31 workflow + 6 tango + 15 validator + 14 config）
444 lines
17 KiB
Python
444 lines
17 KiB
Python
"""
|
||
半自动流水线：处理 xinbiaori_tobe.txt 的分割 + 拼音校正。
|
||
|
||
输入行格式（tobe）: 汉字词:假名: 或 汉字词:假名:（第三段拼音为空）
|
||
输出目标格式: 汉字|分段:假名|分段:拼音|分段
|
||
|
||
处理策略（分级隔离，拿不准的交人工）:
|
||
1. auto_done.txt 🟢 高置信：单音字 + 词典唯一分割 + 无动词/特殊
|
||
2. review_pinyin.txt 🟡 含多音字，已填最常见读音，标 ⚠
|
||
3. review_split.txt 🟡 假名分割失败或多解，靠猜
|
||
4. review_verb.txt 🔴 する/します/でした 等结尾，拿不准砍还是留
|
||
5. review_special.txt🔴 片假名/字母/~/符号 等特殊格式
|
||
|
||
用法: python pipeline.py <tobe文件> [起始行] [条数]
|
||
"""
|
||
import sys
|
||
import re
|
||
from pathlib import Path
|
||
|
||
from pypinyin import pinyin, Style
|
||
from jamdict import Jamdict
|
||
|
||
DB = r"C:\Users\panli\jamdict_local.db"
|
||
jam = Jamdict(db_file=DB)
|
||
|
||
HANZI = re.compile(r'[\u4e00-\u9fff]')
|
||
KATA = re.compile(r'[\u30A0-\u30FF]')
|
||
HIRA = re.compile(r'[\u3040-\u309F]')
|
||
LATIN = re.compile(r'[A-Za-zＡ-Ｚａ-ｚ]')
|
||
# 动词/敬语结尾标记
|
||
VERB_ENDINGS = ('します', 'しました', 'します。', 'でした', 'ください', 'です', 'ます',
|
||
'する', 'させる', 'される')
|
||
|
||
# ます形 i段→u段映射（五段动词变位）
|
||
I_TO_U_MAP = {'き':'く','ぎ':'ぐ','し':'す','ち':'つ','に':'ぬ','ひ':'ふ','び':'ぶ','み':'む','り':'る','い':'う'}
|
||
|
||
def is_verb_kana(kana):
|
||
"""查词典该读音是否对应动词"""
|
||
try:
|
||
r = jam.lookup(kata2hira(kana))
|
||
for e in r.entries:
|
||
for s in e.senses:
|
||
for pos in s.pos:
|
||
if 'verb' in str(pos).lower():
|
||
return True
|
||
except:
|
||
pass
|
||
return False
|
||
|
||
def masu_to_dict_form(kana):
|
||
"""ます形转辞書形，返回 (原型假名, 类型)。类型：'ichidan', 'godan', 'ambiguous', 'failed'"""
|
||
if not kana.endswith('ます'):
|
||
return None, 'not-masu'
|
||
stem = kana[:-2]
|
||
cands = []
|
||
# 一段: stem + る
|
||
ichidan = stem + 'る'
|
||
if is_verb_kana(ichidan):
|
||
cands.append((ichidan, 'ichidan'))
|
||
# 五段: 末尾i音→u音
|
||
if stem and stem[-1] in I_TO_U_MAP:
|
||
godan = stem[:-1] + I_TO_U_MAP[stem[-1]]
|
||
if is_verb_kana(godan):
|
||
cands.append((godan, 'godan'))
|
||
if len(cands) == 1:
|
||
return cands[0]
|
||
elif len(cands) > 1:
|
||
return cands, 'ambiguous'
|
||
# 词典未找到，返回猜测
|
||
guess = ichidan if stem else None
|
||
return guess, 'failed'
|
||
|
||
# 高频多音字黑名单（只有这些才标记为"需确认拼音"，其他直接用最常见读音）
|
||
COMMON_POLYPHONIC = {
|
||
'行', '长', '重', '传', '哪', '得', '的', '了', '着', '地', '为', '将',
|
||
'还', '要', '觉', '处', '朝', '数', '背', '调', '应', '教', '担', '差',
|
||
'量', '种', '落', '便', '曲', '分', '发', '乐', '干', '系', '降', '和',
|
||
'号', '当', '校', '正', '只', '见', '中', '切', '相', '宁', '强', '观',
|
||
'间', '给', '弹', '更', '散', '率', '空', '少', '处', '累', '都', '听',
|
||
'会', '转', '单', '场', '几', '宿', '角', '划', '解', '假', '任', '载',
|
||
'答', '难', '参', '省', '血', '恶', '佛', '供', '尽', '识', '扎', '扇',
|
||
'折', '劲', '吓', '模', '藏', '兴', '似', '奔', '刨', '曾', '泊', '炸',
|
||
'创', '壳', '吐', '喝', '伺', '宁', '蒙', '薄', '咽', '露', '塞', '晕'
|
||
}
|
||
|
||
def kata2hira(s):
|
||
return ''.join(chr(ord(c) - 0x60) if 0x30A1 <= ord(c) <= 0x30F6 else c for c in s)
|
||
|
||
def clean(s):
|
||
s = s.replace('-', '').replace('.', '')
|
||
return kata2hira(s).strip()
|
||
|
||
# ---- 缓存 ----
|
||
_read_cache = {}
|
||
_py_cache = {}
|
||
|
||
def readings(ch):
|
||
if ch in _read_cache:
|
||
return _read_cache[ch]
|
||
rs = set()
|
||
try:
|
||
for c in jam.lookup(ch).chars:
|
||
for g in c.rm_groups:
|
||
for x in g.on_readings:
|
||
rs.add(clean(str(x)))
|
||
for x in g.kun_readings:
|
||
rs.add(clean(str(x)))
|
||
except Exception:
|
||
pass
|
||
rs.discard('')
|
||
_read_cache[ch] = rs
|
||
return rs
|
||
|
||
def valid_pinyins(ch):
|
||
if ch in _py_cache:
|
||
return _py_cache[ch]
|
||
res = pinyin(ch, style=Style.NORMAL, heteronym=True)
|
||
s = res[0] if res else []
|
||
_py_cache[ch] = s
|
||
return s
|
||
|
||
DAK = {'か':'が','き':'ぎ','く':'ぐ','け':'げ','こ':'ご','さ':'ざ','し':'じ','す':'ず',
|
||
'せ':'ぜ','そ':'ぞ','た':'だ','ち':'ぢ','つ':'づ','て':'で','と':'ど','は':'ば',
|
||
'ひ':'び','ふ':'ぶ','へ':'べ','ほ':'ぼ'}
|
||
HAN = {'は':'ぱ','ひ':'ぴ','ふ':'ぷ','へ':'ぺ','ほ':'ぽ'}
|
||
SMALL = 'ゃゅょぁぃぅぇぉ'
|
||
|
||
def variants(r):
|
||
base = {r}
|
||
if r:
|
||
f = r[0]
|
||
if f in DAK:
|
||
base.add(DAK[f] + r[1:])
|
||
if f in HAN:
|
||
base.add(HAN[f] + r[1:])
|
||
out = set()
|
||
for b in base:
|
||
out.add(b)
|
||
if b.endswith(('つ', 'く', 'ち', 'き')):
|
||
out.add(b[:-1] + 'っ')
|
||
if b.endswith('う'):
|
||
out.add(b[:-1])
|
||
i = 1
|
||
while i < len(b):
|
||
if b[i] not in SMALL:
|
||
out.add(b[:i])
|
||
i += 1
|
||
out.discard('')
|
||
return out
|
||
|
||
def align(kanji_tokens, kana):
|
||
"""把 kana 分配到每个 token（汉字或假名或字母）。返回所有可能解的集合。"""
|
||
# 统一转平假名（kanjidic 读音都是平假名，所以目标 kana 也要转）
|
||
kana_hira = kata2hira(kana)
|
||
n = len(kanji_tokens)
|
||
results = []
|
||
|
||
def rec(i, pos, acc):
|
||
if i == n:
|
||
if pos == len(kana):
|
||
results.append(tuple(acc))
|
||
return
|
||
tok = kanji_tokens[i]
|
||
tok_hira = kata2hira(tok) # token 也可能含片假名
|
||
if HANZI.search(tok):
|
||
cands = set()
|
||
for r in readings(tok):
|
||
cands |= variants(r)
|
||
for v in sorted(cands, key=len, reverse=True):
|
||
if kana_hira.startswith(v, pos):
|
||
acc.append(kana[pos:pos + len(v)]) # 用原始 kana（保留片假名）
|
||
rec(i + 1, pos + len(v), acc)
|
||
acc.pop()
|
||
elif LATIN.search(tok):
|
||
# 字母段：尝试所有可能长度
|
||
remain = len(kana) - pos
|
||
min_len = 1
|
||
max_len = max(1, remain - (n - i - 1))
|
||
for l in range(max_len, min_len - 1, -1):
|
||
seg = kana[pos:pos + l]
|
||
acc.append(seg)
|
||
rec(i + 1, pos + l, acc)
|
||
acc.pop()
|
||
else:
|
||
# 假名 token：原样匹配（转平假名后比较）
|
||
if kana_hira.startswith(tok_hira, pos):
|
||
acc.append(kana[pos:pos + len(tok)])
|
||
rec(i + 1, pos + len(tok), acc)
|
||
|
||
rec(0, 0, [])
|
||
return set(results)
|
||
|
||
def split_kanji_tokens(kanji_part):
|
||
"""把汉字词拆成 token 序列：每个汉字单独，连续假名合并成一段，连续字母合并。"""
|
||
tokens = []
|
||
buf = ''
|
||
buf_type = None # 'kana', 'latin', None
|
||
|
||
for ch in kanji_part:
|
||
if HANZI.search(ch):
|
||
if buf:
|
||
tokens.append(buf)
|
||
buf = ''
|
||
buf_type = None
|
||
tokens.append(ch)
|
||
elif LATIN.search(ch) or ch in 'ＡＢＣＤＥＦＧＨＩＪＫＬＭＮＯＰＱＲＳＴＵＶＷＸＹＺａｂｃｄｅｆｇｈｉｊｋｌｍｎｏｐｑｒｓｔｕｖｗｘｙｚ':
|
||
if buf_type == 'latin':
|
||
buf += ch
|
||
else:
|
||
if buf:
|
||
tokens.append(buf)
|
||
buf = ch
|
||
buf_type = 'latin'
|
||
else:
|
||
# 假名
|
||
if buf_type == 'kana':
|
||
buf += ch
|
||
else:
|
||
if buf:
|
||
tokens.append(buf)
|
||
buf = ch
|
||
buf_type = 'kana'
|
||
if buf:
|
||
tokens.append(buf)
|
||
return tokens
|
||
|
||
def make_pinyin(tokens):
|
||
"""为 token 序列生成拼音段；返回 (拼音列表, 是否含高频多音字)。"""
|
||
pys = []
|
||
multi = False
|
||
for tok in tokens:
|
||
if len(tok) == 1 and HANZI.search(tok):
|
||
cands = valid_pinyins(tok)
|
||
if not cands:
|
||
pys.append('')
|
||
else:
|
||
pys.append(cands[0])
|
||
# 只有在高频多音字黑名单里才标记
|
||
if len(cands) > 1 and tok in COMMON_POLYPHONIC:
|
||
multi = True
|
||
elif LATIN.search(tok):
|
||
# 字母段：整段转小写作为拼音（如 CS -> cs, Q -> q）
|
||
pys.append(tok.lower())
|
||
else:
|
||
pys.append('') # 假名段留空
|
||
return pys, multi
|
||
|
||
def classify_and_process(kanji_part, kana_part):
|
||
"""返回 (bucket, 处理后字符串, 备注)。"""
|
||
note = []
|
||
|
||
# ~ 标记
|
||
has_tilde = '～' in kanji_part or '~' in kanji_part
|
||
kanji_clean = kanji_part.replace('～', '').replace('~', '')
|
||
|
||
# 规则1：纯英文/纯字母词 → 跳过不处理（汉字段无任何汉字）
|
||
if not HANZI.search(kanji_clean):
|
||
return 'skip', f"{kanji_part}:{kana_part}:", "纯英文/无汉字,跳过"
|
||
|
||
# 处理々（同字重复符号）：替换成前一个字
|
||
kanji_expanded = []
|
||
for i, ch in enumerate(kanji_clean):
|
||
if ch == '々' and i > 0:
|
||
kanji_expanded.append(kanji_expanded[-1]) # 重复前一个字
|
||
else:
|
||
kanji_expanded.append(ch)
|
||
kanji_clean = ''.join(kanji_expanded)
|
||
|
||
# 动词/敬语结尾
|
||
is_verbish = has_tilde or any(kana_part.endswith(e) for e in VERB_ENDINGS)
|
||
|
||
tokens = split_kanji_tokens(kanji_clean)
|
||
|
||
# 含字母：按正常流程对齐（字母作为独立 token）
|
||
has_latin = LATIN.search(kanji_clean)
|
||
has_kata = KATA.search(kanji_clean)
|
||
|
||
if is_verbish:
|
||
if has_tilde:
|
||
# 规则3：含~：去掉 ~ 和 します/する，保留名词词干，直接处理
|
||
stem_kana = kana_part
|
||
for e in sorted(VERB_ENDINGS, key=len, reverse=True):
|
||
if stem_kana.endswith(e):
|
||
stem_kana = stem_kana[:-len(e)]
|
||
break
|
||
sols = align(tokens, stem_kana)
|
||
if len(sols) == 1:
|
||
kana_tokens = list(next(iter(sols)))
|
||
pys, multi = make_pinyin(tokens)
|
||
out = f"{'|'.join(tokens)}:{'|'.join(kana_tokens)}:{'|'.join(pys)}"
|
||
# 分割成功，按正常流程分类
|
||
if multi:
|
||
return 'pinyin', out, "含~+多音字,请确认"
|
||
return 'auto', out, ""
|
||
else:
|
||
# 分割失败，进 split
|
||
return 'split', f"{kanji_part}:{kana_part}:", "含~但假名无法分割"
|
||
else:
|
||
# ます形动词：转原型（辞書形）
|
||
dict_result, vtype = masu_to_dict_form(kana_part)
|
||
if vtype == 'ambiguous':
|
||
# 歧义：列出所有可能
|
||
opts = ' 或 '.join([f"{k}({t})" for k, t in dict_result])
|
||
return 'verb', f"[原] {kanji_part}:{kana_part}: [ます→原型有歧义] {opts}，请人工确认", "ます形歧义"
|
||
elif vtype in ('ichidan', 'godan'):
|
||
# 唯一解：汉字段的送り仮名也要同步转换
|
||
# 根据变位类型处理
|
||
if kanji_clean.endswith('ます'):
|
||
if vtype == 'ichidan':
|
||
# 一段：ます→る，直接替换
|
||
kanji_dict = kanji_clean[:-2] + 'る'
|
||
else:
|
||
# 五段：ます前的i段音→u段音，例如 かり→かる
|
||
stem = kanji_clean[:-2]
|
||
if stem and stem[-1] in I_TO_U_MAP:
|
||
kanji_dict = stem[:-1] + I_TO_U_MAP[stem[-1]]
|
||
else:
|
||
kanji_dict = kanji_clean # 异常情况，保持原样
|
||
else:
|
||
kanji_dict = kanji_clean
|
||
|
||
tokens_dict = split_kanji_tokens(kanji_dict)
|
||
sols = align(tokens_dict, dict_result)
|
||
if len(sols) == 1:
|
||
kana_tokens = list(next(iter(sols)))
|
||
pys, _ = make_pinyin(tokens_dict)
|
||
suggest = f"{'|'.join(tokens_dict)}:{'|'.join(kana_tokens)}:{'|'.join(pys)}"
|
||
return 'verb', f"[原] {kanji_part}:{kana_part}: [建议ます→{vtype}原型] {suggest}", f"ます→{vtype}"
|
||
else:
|
||
return 'verb', f"[原] {kanji_part}:{kana_part}: [建议ます→原型] {kanji_dict}:{dict_result}，但分割失败", f"ます→{vtype}但分割失败"
|
||
else:
|
||
# 词典未找到或不是ます形，按原来逻辑砍词尾
|
||
stem_kana = kana_part
|
||
for e in sorted(VERB_ENDINGS, key=len, reverse=True):
|
||
if stem_kana.endswith(e):
|
||
stem_kana = stem_kana[:-len(e)]
|
||
break
|
||
sols = align(tokens, stem_kana)
|
||
if len(sols) == 1:
|
||
kana_tokens = list(next(iter(sols)))
|
||
pys, _ = make_pinyin(tokens)
|
||
suggest = f"{'|'.join(tokens)}:{'|'.join(kana_tokens)}:{'|'.join(pys)}"
|
||
else:
|
||
suggest = f"{'|'.join(tokens)}:{stem_kana}:?"
|
||
return 'verb', f"[原] {kanji_part}:{kana_part}: [建议砍词尾] {suggest}", "动词/敬语结尾"
|
||
|
||
# 普通词（含字母/片假名也走这里）：对齐
|
||
sols = align(tokens, kana_part)
|
||
if len(sols) == 0:
|
||
return 'split', f"{kanji_part}:{kana_part}:", "假名无法分割(词典缺读音)"
|
||
if len(sols) > 1:
|
||
# 多解，取第一个但标记
|
||
kana_tokens = list(sorted(sols)[0])
|
||
pys, multi = make_pinyin(tokens)
|
||
out = f"{'|'.join(tokens)}:{'|'.join(kana_tokens)}:{'|'.join(pys)}"
|
||
return 'split', out, f"分割有{len(sols)}种可能,取其一,请确认"
|
||
|
||
kana_tokens = list(next(iter(sols)))
|
||
pys, multi = make_pinyin(tokens)
|
||
out = f"{'|'.join(tokens)}:{'|'.join(kana_tokens)}:{'|'.join(pys)}"
|
||
|
||
# 含字母/片假名标记到 special，但已完成分割和拼音
|
||
if has_latin or has_kata:
|
||
tag = "含字母" if has_latin else "含片假名"
|
||
if multi:
|
||
return 'special', out, f"{tag}+多音字,请确认"
|
||
return 'special', out, f"{tag},已自动处理,请确认"
|
||
|
||
if multi:
|
||
return 'pinyin', out, "含多音字,拼音取最常见,请确认"
|
||
return 'auto', out, ""
|
||
|
||
def main():
|
||
if len(sys.argv) < 2:
|
||
print("usage: python pipeline.py <tobe文件> [起始行] [条数]")
|
||
return
|
||
src = Path(sys.argv[1])
|
||
start = int(sys.argv[2]) if len(sys.argv) > 2 else 1
|
||
count = int(sys.argv[3]) if len(sys.argv) > 3 else 10**9
|
||
|
||
buckets = {'auto': [], 'pinyin': [], 'split': [], 'verb': [], 'special': [], 'skip': []}
|
||
processed = 0
|
||
|
||
with open(src, 'r', encoding='utf-8') as f:
|
||
for lineno, raw in enumerate(f, 1):
|
||
if lineno < start:
|
||
continue
|
||
if processed >= count:
|
||
break
|
||
line = raw.rstrip('\n').strip()
|
||
if not line:
|
||
continue
|
||
processed += 1
|
||
norm = line.replace('：', ':')
|
||
parts = norm.split(':')
|
||
if len(parts) < 2:
|
||
buckets['special'].append(f"L{lineno} {line} | 格式异常")
|
||
continue
|
||
kanji_part = parts[0].strip()
|
||
kana_part = parts[1].strip()
|
||
if not kana_part:
|
||
buckets['special'].append(f"L{lineno} {line} | 无假名")
|
||
continue
|
||
bucket, out, note = classify_and_process(kanji_part, kana_part)
|
||
tag = f"L{lineno}"
|
||
if note:
|
||
buckets[bucket].append(f"{tag} {out} # {note}")
|
||
else:
|
||
buckets[bucket].append(out)
|
||
|
||
outdir = src.parent
|
||
files = {
|
||
'auto': ('auto_done.txt', '高置信度,可直接用'),
|
||
'pinyin': ('review_pinyin.txt', '含多音字,校对拼音'),
|
||
'split': ('review_split.txt', '假名分割待确认'),
|
||
'verb': ('review_verb.txt', 'する/敬语结尾,拿不准砍还是留'),
|
||
'special': ('review_special.txt', '片假名/字母/特殊格式'),
|
||
'skip': ('skip.txt', '纯英文/无汉字,跳过不处理'),
|
||
}
|
||
summary = []
|
||
for key, (fname, desc) in files.items():
|
||
fpath = outdir / fname
|
||
# 追加模式：先读取已有内容
|
||
existing = []
|
||
if fpath.exists():
|
||
existing = [ln.rstrip('\n') for ln in fpath.read_text(encoding='utf-8').splitlines() if ln.strip()]
|
||
new_content = existing + buckets[key]
|
||
fpath.write_text('\n'.join(new_content) + ('\n' if new_content else ''), encoding='utf-8')
|
||
summary.append(f" {fname:22s} {len(buckets[key]):5d} 条新增 (共{len(new_content)}条, {desc})")
|
||
|
||
print(f"处理 {processed} 条 (从第 {start} 行起)")
|
||
print('\n'.join(summary))
|
||
|
||
# 严格条数校验（铁律）
|
||
total_output = sum(len(buckets[k]) for k in buckets)
|
||
if total_output != processed:
|
||
print(f"\n[!] 条数校验失败！输入 {processed} 条，输出 {total_output} 条，差异 {total_output - processed} 条")
|
||
print(" 条数不对结果一定不对，请检查流水线逻辑")
|
||
else:
|
||
print(f"\n[OK] 条数校验通过：输入 {processed} 条 = 输出 {total_output} 条")
|
||
|
||
if __name__ == '__main__':
|
||
main()
|