ila
b1b2e6c6fe
- ECDICT 常用子集（82,721 条，含中文释义）与 CMUdict ARPAbet→IPA 转写音标， 由 scripts/dict_prepare.py 从 sha256 pin 的源显式准备，运行时不联网。 - schema v12：放开 lexgo_dictionaries 单行约束并增加 provider 列，英英与英汉词典 可分别启用；查词合并时中文释义在前、英英释义随后。 - 音标只取可验证来源：优先 CMUdict 转写，否则保留 ECDICT 记法并标注来源， 两者都不可靠时不显示；含 ^ 等丢失首音的记法整条丢弃。 - 屈折形经 WordNet lemma 解析后回查中文词典，但不把词目音标复制到屈折形上。 - 管理端可区分两种词典并分别启停；学习端查词面板显示音标与中文释义。
351 lines
14 KiB
Python
351 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""Build the prepared 英汉词典 resource for LexGo (#40).
|
||
|
||
This is tooling, not product runtime: the Go server only imports the prepared ZIP.
|
||
Two pinned sources are combined:
|
||
|
||
* ECDICT StarDict (MIT) -> Chinese glosses and a fallback phonetic notation
|
||
* CMUdict (BSD-2) -> ARPAbet phonemes transcribed to IPA
|
||
|
||
The subset is limited to lowercase single-word headwords that either exist in the
|
||
WordNet lemma list (pinned in server/wordnet-resource.json) or carry an ECDICT
|
||
exam tag, which keeps the resource at a few MiB instead of the full 340 万条.
|
||
|
||
Everything is verified: source sha256, subset size within the pinned tolerance and
|
||
the phonetic character set (no character outside the IPA alphabet may survive).
|
||
|
||
Usage:
|
||
python scripts/dict_prepare.py # download + build into .local/dictionaries
|
||
python scripts/dict_prepare.py --cache-dir DIR # reuse already downloaded sources
|
||
python scripts/dict_prepare.py --keep-going # still write the artifact if counts drift
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import gzip
|
||
import hashlib
|
||
import json
|
||
import re
|
||
import struct
|
||
import sys
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
import zipfile
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
PIN_FILE = ROOT / "server" / "zh-dictionary-resource.json"
|
||
DEFAULT_OUT = ROOT / ".local" / "dictionaries" / "zh-dict-v1.zip"
|
||
DEFAULT_CACHE = ROOT / ".local" / "dictionaries" / "cache"
|
||
|
||
EXAM_TAG_RE = re.compile(r"\((?:[^)]*?(研|四|六|托|专|雅|高|中|初))[^)]*?\)")
|
||
PHONETIC_RE = re.compile(r"^\*?\[([^\]\u4e00-\u9fff]*)\]")
|
||
CJK_RE = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff\uf900-\ufaff]")
|
||
SPACE_RE = re.compile(r"\s+")
|
||
|
||
# ARPAbet (CMUdict, American English) -> IPA, a fixed documented convention.
|
||
ARPABET_CONSONANTS = {
|
||
"B": "b", "CH": "t͡ʃ", "D": "d", "DH": "ð", "F": "f", "G": "ɡ", "HH": "h",
|
||
"JH": "d͡ʒ", "K": "k", "L": "l", "M": "m", "N": "n", "NG": "ŋ", "P": "p",
|
||
"R": "ɹ", "S": "s", "SH": "ʃ", "T": "t", "TH": "θ", "V": "v", "W": "w",
|
||
"Y": "j", "Z": "z", "ZH": "ʒ",
|
||
}
|
||
ARPABET_VOWELS = {
|
||
"AA": ("ɑ", "ː"), "AE": ("æ", ""), "AH": ("ʌ", ""), "AO": ("ɔ", "ː"),
|
||
"AW": ("a", "ʊ"), "AY": ("a", "ɪ"), "EH": ("e", ""), "ER": ("ɜ", "ː"),
|
||
"EY": ("e", "ɪ"), "IH": ("ɪ", ""), "IY": ("i", "ː"), "OW": ("o", "ʊ"),
|
||
"OY": ("ɔ", "ɪ"), "UH": ("ʊ", ""), "UW": ("u", "ː"),
|
||
}
|
||
STRESS_MARKS = {"1": "ˈ", "2": "ˌ", "0": ""}
|
||
# Only character-level substitutions whose meaning is verifiable from the data.
|
||
ECDICT_CHAR_MAP = {
|
||
"'": "ˈ", ",": "ˌ", ".": "ˌ", "ˊ": "ˈ", ":": "ː", "ә": "ə", "ε": "e", # ә is U+04D9, ε U+03B5
|
||
" ": "", "-": "", "=": "", ";": "", "^": "", # ^ is a stray separator mid-word
|
||
}
|
||
IPA_ALPHABET = set(
|
||
"abcdefghijklmnopqrstuvwxyz"
|
||
"æɑɒɔəɜɛɪʊʌʃʒθðŋɹɡɚɝt͡ʃd͡ʒ"
|
||
"ːˈˌ"
|
||
)
|
||
|
||
|
||
def log(message: str) -> None:
|
||
print(message, flush=True)
|
||
|
||
|
||
def digest(path: Path) -> str:
|
||
hasher = hashlib.sha256()
|
||
with path.open("rb") as handle:
|
||
for block in iter(lambda: handle.read(1024 * 1024), b""):
|
||
hasher.update(block)
|
||
return hasher.hexdigest()
|
||
|
||
|
||
def download(source: dict, cache_dir: Path) -> Path:
|
||
"""Download a pinned source, trying every mirror, and verify sha256 + size."""
|
||
name = source["name"]
|
||
target = cache_dir / source_filename(source["source"])
|
||
urls = [source["source"]] + list(source.get("mirrors", []))
|
||
if target.exists() and digest(target) == source["sha256"]:
|
||
log(f"[cache] {name}: {target.name} ({target.stat().st_size} 字节, sha256 匹配)")
|
||
return target
|
||
last_error = None
|
||
for url in urls:
|
||
log(f"[download] {name}: {url}")
|
||
try:
|
||
request = urllib.request.Request(url, headers={"User-Agent": "lexgo-dict-prepare/1"})
|
||
with urllib.request.urlopen(request, timeout=300) as response, target.open("wb") as handle:
|
||
while True:
|
||
block = response.read(1024 * 256)
|
||
if not block:
|
||
break
|
||
handle.write(block)
|
||
except (urllib.error.URLError, TimeoutError, OSError) as error:
|
||
last_error = error
|
||
log(f"[download] {name}: 失败 {type(error).__name__}: {error}")
|
||
continue
|
||
found = digest(target)
|
||
if found == source["sha256"]:
|
||
log(f"[download] {name}: 完成 {target.stat().st_size} 字节, sha256 匹配")
|
||
return target
|
||
log(f"[download] {name}: sha256 不匹配 ({found[:16]}… != {source['sha256'][:16]}…)，尝试下一个通道")
|
||
raise SystemExit(f"{name}: 所有通道都无法取得通过校验的资源（最后错误 {last_error}）")
|
||
|
||
|
||
def source_filename(url: str) -> str:
|
||
return Path(urllib.parse.urlparse(url).path).name
|
||
|
||
|
||
def wordnet_lemmas(archive: Path) -> set:
|
||
lemmas = set()
|
||
with zipfile.ZipFile(archive) as bundle:
|
||
for name in bundle.namelist():
|
||
if "/index." not in name:
|
||
continue
|
||
for line in bundle.read(name).decode("latin-1").splitlines():
|
||
if not line.strip() or line.startswith(" "):
|
||
continue
|
||
lemmas.add(line.split(" ", 1)[0].lower().replace("_", " "))
|
||
return lemmas
|
||
|
||
|
||
def cmudict_entries(path: Path) -> dict:
|
||
entries = {}
|
||
for line in path.read_text(encoding="latin-1").splitlines():
|
||
if line.startswith(";;;") or not line.strip():
|
||
continue
|
||
head, _, rest = line.partition(" ")
|
||
head = head.split("(", 1)[0].lower()
|
||
if head and rest.strip() and head not in entries:
|
||
entries[head] = rest.split()
|
||
return entries
|
||
|
||
|
||
def stardict_entries(archive: Path):
|
||
with zipfile.ZipFile(archive) as bundle:
|
||
names = bundle.namelist()
|
||
ifo_name = next((n for n in names if n.endswith(".ifo")), None)
|
||
idx_name = next((n for n in names if n.endswith(".idx")), None)
|
||
dict_name = next((n for n in names if n.endswith(".dict")), None)
|
||
if not (ifo_name and idx_name and dict_name):
|
||
raise SystemExit("ECDICT 归档缺少 .ifo/.idx/.dict")
|
||
ifo = bundle.read(ifo_name).decode("utf-8", "replace")
|
||
sequence = ""
|
||
for line in ifo.splitlines():
|
||
if line.startswith("sametypesequence="):
|
||
sequence = line.split("=", 1)[1].strip()
|
||
if sequence != "m":
|
||
raise SystemExit(f"未预期的 StarDict sametypesequence: {sequence!r}（只处理纯文本 m）")
|
||
index = bundle.read(idx_name)
|
||
payload = bundle.read(dict_name)
|
||
offset = 0
|
||
while offset < len(index):
|
||
end = index.index(b"\x00", offset)
|
||
word = index[offset:end].decode("utf-8", "replace")
|
||
start, size = struct.unpack(">II", index[end + 1:end + 9])
|
||
yield word, payload[start:start + size].decode("utf-8", "replace")
|
||
offset = end + 9
|
||
|
||
|
||
def arpabet_to_ipa(phonemes) -> str:
|
||
vowel_count = sum(1 for phoneme in phonemes if phoneme.rstrip("012").strip() in ARPABET_VOWELS)
|
||
parts = []
|
||
for phoneme in phonemes:
|
||
stress = ""
|
||
base = phoneme
|
||
if phoneme and phoneme[-1].isdigit():
|
||
base, digit = phoneme[:-1], phoneme[-1]
|
||
stress = STRESS_MARKS.get(digit, "")
|
||
if base in ARPABET_VOWELS and vowel_count == 1:
|
||
# A monosyllable carries no stress mark in IPA (cat is /kæt/, not /kˈæt/).
|
||
stress = ""
|
||
if base in ARPABET_CONSONANTS:
|
||
parts.append(ARPABET_CONSONANTS[base])
|
||
elif base in ARPABET_VOWELS:
|
||
head, tail = ARPABET_VOWELS[base]
|
||
if stress == "":
|
||
# Unstressed vowels are short: curiosity ends /əti/, not /ətiː/.
|
||
tail = ""
|
||
if base == "AH":
|
||
head = "ə"
|
||
elif base == "ER":
|
||
head = "ɚ"
|
||
parts.append(stress + head + tail)
|
||
else:
|
||
return ""
|
||
return "".join(parts)
|
||
|
||
|
||
def normalise_ecdict_phonetic(raw: str) -> str:
|
||
text = raw.strip()
|
||
# A phonetic that starts with the stray caret has lost its first sound (grok is stored as
|
||
# "^rɔk"): showing "rɔk" would be a wrong transcription, so the whole value is dropped.
|
||
if text.startswith("^"):
|
||
return ""
|
||
for old, new in ECDICT_CHAR_MAP.items():
|
||
text = text.replace(old, new)
|
||
return SPACE_RE.sub("", text)
|
||
|
||
|
||
def build_subset(lemmas: set, exam_entries) -> dict:
|
||
subset = {}
|
||
for word, body in exam_entries:
|
||
lower = word.lower()
|
||
if " " in word or not lower.isalpha() or not lower.islower() or lower in subset:
|
||
continue
|
||
if lower not in lemmas and not EXAM_TAG_RE.search(body):
|
||
continue
|
||
subset[lower] = body
|
||
return subset
|
||
|
||
|
||
def translation_of(body: str) -> str:
|
||
lines = []
|
||
for line in body.splitlines():
|
||
stripped = PHONETIC_RE.sub("", line).strip()
|
||
if not stripped or not CJK_RE.search(stripped):
|
||
continue
|
||
# "(研四六托 4518/5059)" is an exam-list rank and "[时态] guarded, guarding" is a word-form
|
||
# list: neither is a meaning, so they stay out of the gloss.
|
||
if stripped.startswith("(") and stripped.endswith(")"):
|
||
continue
|
||
if stripped.startswith("[时态]"):
|
||
continue
|
||
lines.append(stripped)
|
||
return "\n".join(lines)
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="构建 LexGo 英汉词典资源 (#40)")
|
||
parser.add_argument("--out", default=str(DEFAULT_OUT))
|
||
parser.add_argument("--cache-dir", default=str(DEFAULT_CACHE))
|
||
parser.add_argument("--keep-going", action="store_true", help="计数漂移时仍产出，只在 manifest 标记")
|
||
args = parser.parse_args()
|
||
|
||
pin = json.loads(PIN_FILE.read_text(encoding="utf-8"))
|
||
sources = {source["role"]: source for source in pin["sources"]}
|
||
cache_dir = Path(args.cache_dir)
|
||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
ecdict_zip = download(sources["chinese-gloss"], cache_dir)
|
||
cmudict_file = download(sources["ipa-phonetic"], cache_dir)
|
||
wordnet_zip = download(sources["subset-wordlist"], cache_dir)
|
||
|
||
log("[parse] WordNet lemma 词表")
|
||
lemmas = wordnet_lemmas(wordnet_zip)
|
||
log(f"[parse] WordNet lemma: {len(lemmas)}")
|
||
log("[parse] CMUdict 音素")
|
||
cmu = cmudict_entries(cmudict_file)
|
||
log(f"[parse] CMUdict 词条: {len(cmu)}")
|
||
|
||
log("[build] 裁剪 ECDICT 子集")
|
||
subset = build_subset(lemmas, stardict_entries(ecdict_zip))
|
||
log(f"[build] 子集词条: {len(subset)}")
|
||
|
||
records = []
|
||
stats = {"cmudict": 0, "ecdict": 0, "none": 0, "droppedPhonetic": 0, "empty": 0}
|
||
offenders = {}
|
||
for word in sorted(subset):
|
||
body = subset[word]
|
||
translation = translation_of(body)
|
||
phonetic = ""
|
||
source_name = ""
|
||
if word in cmu:
|
||
phonetic = arpabet_to_ipa(cmu[word])
|
||
source_name = "cmudict" if phonetic else ""
|
||
if not phonetic:
|
||
match = PHONETIC_RE.match(body)
|
||
candidate = normalise_ecdict_phonetic(match.group(1)) if match else ""
|
||
if candidate:
|
||
phonetic = candidate
|
||
source_name = "ecdict"
|
||
if phonetic:
|
||
unknown = sorted({char for char in phonetic if char not in IPA_ALPHABET})
|
||
if unknown:
|
||
stats["droppedPhonetic"] += 1
|
||
for char in unknown:
|
||
offenders[char] = offenders.get(char, 0) + 1
|
||
phonetic = ""
|
||
source_name = ""
|
||
if not translation and not phonetic:
|
||
# An entry that lost both its gloss and its phonetic answers nothing, so it is dropped
|
||
# instead of occupying a lookup slot with a bare headword.
|
||
stats["empty"] += 1
|
||
continue
|
||
stats[source_name or "none"] += 1
|
||
records.append({
|
||
"w": word,
|
||
"t": translation,
|
||
"p": phonetic,
|
||
"ps": source_name,
|
||
})
|
||
|
||
if offenders:
|
||
log(f"[check] 丢弃的非 IPA 字符: {sorted((c, n) for c, n in offenders.items())}")
|
||
if stats["droppedPhonetic"]:
|
||
log(f"[check] 因字符不合法丢弃音标 {stats['droppedPhonetic']} 条（不做语义猜测）")
|
||
if stats["empty"]:
|
||
log(f"[check] 既无释义也无音标丢弃 {stats['empty']} 条")
|
||
|
||
expected = pin["subset"]["expectedEntries"]
|
||
tolerance = pin["subset"]["toleranceEntries"]
|
||
drift = abs(len(records) - expected)
|
||
log(f"[check] 条目数 {len(records)}（pin 期望 {expected} ± {tolerance}）")
|
||
if drift > tolerance and not args.keep_going:
|
||
raise SystemExit("子集条目数超出 pin 容差，拒绝产出；确认数据源变化后用 --keep-going 重新评估")
|
||
if len(records) == 0:
|
||
raise SystemExit("子集为空，拒绝产出")
|
||
|
||
manifest = {
|
||
"format": pin["format"],
|
||
"version": pin["version"],
|
||
"language": pin["language"],
|
||
"provider": pin["provider"],
|
||
"entries": len(records),
|
||
"phonetic": {"cmudict": stats["cmudict"], "ecdict": stats["ecdict"], "none": stats["none"]},
|
||
"droppedPhonetic": stats["droppedPhonetic"],
|
||
"droppedEmpty": stats["empty"],
|
||
"subsetRule": pin["subset"]["rule"],
|
||
"sources": [
|
||
{"name": s["name"], "role": s["role"], "sha256": s["sha256"], "license": s["license"]}
|
||
for s in pin["sources"]
|
||
],
|
||
"preparedBy": "scripts/dict_prepare.py",
|
||
}
|
||
destination = Path(args.out)
|
||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||
with zipfile.ZipFile(destination, "w", zipfile.ZIP_DEFLATED) as bundle:
|
||
payload = "\n".join(json.dumps(record, ensure_ascii=False, sort_keys=True) for record in records)
|
||
bundle.writestr("entries.jsonl.gz", gzip.compress(payload.encode("utf-8")))
|
||
bundle.writestr("manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2))
|
||
log(f"[write] {destination} ({destination.stat().st_size} 字节)")
|
||
log(f"[write] sha256 {digest(destination)}")
|
||
log(f"[stats] CMUdict IPA {stats['cmudict']} / ECDICT 记法 {stats['ecdict']} / 无音标 {stats['none']}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|