294 lines
14 KiB
Python
294 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""Build the domain term bank used to seed ASR misrecognition candidates.
|
||
|
||
Why not ChEMBL / DrugBank directly (evaluated 2026-07-25):
|
||
ChEMBL's 4,225 approved molecules and DrugBank's dumps are **English** chemical
|
||
names (PRAZOSIN, OFLOXACIN). Kai's transcripts are Chinese speech, so a Chinese
|
||
ASR never emits "OFLOXACIN" — it emits 氧氟沙星. An English chemical dictionary
|
||
therefore contributes ~nothing to phonetic correction of Chinese audio, while
|
||
costing a multi-GB download. Skipped deliberately.
|
||
|
||
What actually pays off, in descending order of value:
|
||
1. **INN stem generation** — the WHO INN system gives Chinese biologics highly
|
||
regular suffixes (单抗/珠单抗/替尼/西普/他滨...). Any Han n-gram ending in one
|
||
of those stems is almost certainly a drug name, which lets us *recognize* drug
|
||
mentions we have never seen, not just ones in a list.
|
||
2. **zhwiki intitle: search** — 237 hits for 单抗, 49 for 替尼, all real INN
|
||
Chinese generic names (帕博利珠单抗, 奥希替尼...). Small, free, exactly on-domain.
|
||
3. **Curated CDMO/antibody/CMC jargon** — the vocabulary that actually dominates
|
||
Kai's meetings (原液/下游/批记录/中试/放行/一次性反应器/GMP 证...). ASR mangles
|
||
these constantly (批记录→披露/P图) and no public corpus contains them as a set.
|
||
4. **THUOCL medical/IT/caijing** — general high-frequency words. Weak on their own
|
||
(医院/检查) but useful as a "this is a real word, do not 'correct' it" allowlist,
|
||
which is how we suppress the 主要→朱岩 class of false positives.
|
||
|
||
Output: termbank/terms.json { term: {sources: [...], kind: str} }
|
||
termbank/allowlist.txt common words that must never be rewritten
|
||
|
||
python3 -m lab.termbank build # download + assemble (network)
|
||
python3 -m lab.termbank stats
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import sys
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
from .config import CFG
|
||
|
||
LAB = CFG.root
|
||
BANK = CFG.termbank_dir
|
||
UA = {"User-Agent": "asr-lab/1.0 (personal research; contact: local)"}
|
||
|
||
# ---------------------------------------------------------------- INN stems
|
||
# WHO INN suffixes, Chinese renderings. A Han string ending in one of these is
|
||
# a drug/biologic with very high probability.
|
||
INN_SUFFIXES = [
|
||
"单抗", "珠单抗", "尤单抗", "利珠单抗", "妥珠单抗", "西单抗", "木单抗", "库单抗",
|
||
"替尼", "布替尼", "沙替尼", "래替尼", "洛替尼", "非替尼",
|
||
"西普", "普肽", "格鲁肽", "鲁肽", "那肽",
|
||
"他滨", "他赛", "紫杉醇", "沙星", "霉素", "环素", "西林", "唑胺", "磺胺",
|
||
"干扰素", "白介素", "促红素", "生长素",
|
||
"酶", "疫苗", "佐剂",
|
||
]
|
||
# Suffix -> what kind of thing it names, for the prompt hint.
|
||
INN_KIND = {"单抗": "抗体药", "替尼": "小分子抑制剂", "肽": "多肽药", "疫苗": "疫苗"}
|
||
|
||
# --------------------------------------------------- curated CDMO / CMC jargon
|
||
# Hand-built from Kai's own minutes + glossary. These are the terms that actually
|
||
# break: each line is the CORRECT form; the flagger finds phonetic neighbours of
|
||
# them in the transcript. Grouped only for readability.
|
||
CDMO_TERMS = """
|
||
原液 制剂 下游 上游 中试 放行 灌装 冻干 除菌过滤 层析 亲和层析 离子交换 疏水层析
|
||
超滤 透析 病毒灭活 病毒去除 纳米过滤 收率 纯度 电荷异构体 糖型 聚体 碎片
|
||
批记录 批生产记录 批分析记录 工艺验证 清洁验证 培养基模拟灌装 无菌工艺模拟
|
||
偏差 变更 CAPA 超标 超趋势 纠正预防措施 质量回顾 年度质量回顾
|
||
稳定性 加速稳定性 长期稳定性 强制降解 影响因素试验
|
||
细胞库 主细胞库 工作细胞库 细胞株 稳定池 单克隆 克隆筛选 传代 传代稳定性
|
||
反应器 生物反应器 一次性反应器 摇瓶 摇床 补料 流加 灌流 收获 澄清
|
||
质谱 肽图 N端测序 效价 生物活性 亲和力 抗体偶联 载药量 游离药物
|
||
工艺表征 工艺性能确认 关键质量属性 关键工艺参数 设计空间 控制策略
|
||
技术转移 工艺开发 分析方法开发 方法验证 方法转移 对照品 标准品
|
||
临床前 毒理 非GLP 药代 药效 首次人体试验 剂量爬坡 剂量探索
|
||
申报 临床试验申请 新药上市申请 药品注册 药品上市许可持有人 受托生产
|
||
现场检查 迎检 迎审 缺陷项 主要缺陷 严重缺陷 一般缺陷 整改
|
||
生产许可证 药品生产许可证 B证 C证 D证 委托生产 受托方 委托方
|
||
洁净区 洁净级别 压差 环境监测 沉降菌 浮游菌 表面微生物 尘埃粒子
|
||
计算机化系统 数据完整性 审计追踪 电子签名 权限管理 系统验证
|
||
双抗 双特异性抗体 三特异性抗体 纳米抗体 单域抗体 融合蛋白 重组蛋白
|
||
抗体偶联药物 细胞治疗 基因治疗 慢病毒 腺病毒 质粒 转染 瞬转 稳转
|
||
产能 产能利用率 排产 排期 交付 交付周期 项目里程碑 阶段付款
|
||
报价 服务包 里程碑款 技术服务费 居间费 保密协议 主服务协议 工作说明书
|
||
"""
|
||
|
||
# Company / platform names common in Kai's domain that generic corpora lack.
|
||
DOMAIN_ORGS = """
|
||
药明生物 药明康德 药明合联 康龙化成 泰格医药 昭衍新药 金斯瑞 义翘神州
|
||
百奥赛图 和铂医药 信达生物 君实生物 恒瑞医药 复宏汉霖 三生国健 神州细胞
|
||
迈威生物 洛启生物 麦济生物 维眸生物 白帆生物 鼎康生物 恒元生物 愚公生物
|
||
奥浦迈 澳斯康 东富龙 楚天科技 森松 赛多利斯 思拓凡 默克 丹纳赫 颇尔
|
||
药审中心 国家药监局 药品审评中心 核查中心 中检院
|
||
"""
|
||
|
||
THUOCL = {
|
||
"medical": "https://raw.githubusercontent.com/thunlp/THUOCL/master/data/THUOCL_medical.txt",
|
||
"it": "https://raw.githubusercontent.com/thunlp/THUOCL/master/data/THUOCL_IT.txt",
|
||
"caijing": "https://raw.githubusercontent.com/thunlp/THUOCL/master/data/THUOCL_caijing.txt",
|
||
}
|
||
|
||
# The allowlist's job is to stop entity candidates from displacing ordinary words
|
||
# (主要→朱岩, 分钟→冯昆仑, 具体→聚体). THUOCL's domain lists cannot do this: they
|
||
# contain 医院/检查 but not 具体/自己/硬件. jieba's dictionary is a general-purpose
|
||
# frequency table over 330k Han words, and measurement showed freq>=500 covers the
|
||
# entire observed false-positive class (具体 自己 硬件 不了 主要 时间 分钟) while
|
||
# still excluding the真 targets we must keep flagging (元烨 药名 中式 姚明).
|
||
JIEBA_DICT = "https://raw.githubusercontent.com/fxsjy/jieba/master/jieba/dict.txt"
|
||
JIEBA_MIN_FREQ = 500
|
||
|
||
HAN = re.compile(r"^[\u4e00-\u9fff]+$")
|
||
|
||
|
||
_last_wiki = [0.0]
|
||
|
||
|
||
def fetch(url: str, timeout=60, throttle=0.0) -> str:
|
||
# zhwiki returns 429 after a handful of rapid api.php calls; pace them and
|
||
# retry once on 429 so a term-bank build is reproducible rather than
|
||
# silently losing whole drug-name families.
|
||
if throttle:
|
||
import time
|
||
wait = throttle - (time.time() - _last_wiki[0])
|
||
if wait > 0:
|
||
time.sleep(wait)
|
||
_last_wiki[0] = time.time()
|
||
req = urllib.request.Request(url, headers=UA)
|
||
for attempt in range(4):
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||
return r.read().decode("utf-8", "ignore")
|
||
except urllib.error.HTTPError as e:
|
||
if e.code != 429 or attempt == 3:
|
||
raise
|
||
import time
|
||
time.sleep(6 * (attempt + 1))
|
||
raise RuntimeError("unreachable")
|
||
|
||
|
||
def wiki_intitle(keyword: str, limit=500) -> list[str]:
|
||
"""zhwiki titles containing `keyword`. INN generic names live here."""
|
||
out, offset = [], 0
|
||
while offset < limit:
|
||
q = urllib.parse.urlencode({
|
||
"action": "query", "list": "search",
|
||
"srsearch": f"intitle:{keyword}", "srlimit": 50, "sroffset": offset,
|
||
"srnamespace": 0, "format": "json",
|
||
})
|
||
try:
|
||
d = json.loads(fetch(f"https://zh.wikipedia.org/w/api.php?{q}", 40, throttle=1.5))
|
||
except Exception as e:
|
||
print(f" wiki {keyword} @{offset} failed: {e}", file=sys.stderr)
|
||
break
|
||
hits = d.get("query", {}).get("search", [])
|
||
if not hits:
|
||
break
|
||
out += [h["title"] for h in hits]
|
||
if "continue" not in d:
|
||
break
|
||
offset += 50
|
||
return out
|
||
|
||
|
||
def build():
|
||
BANK.mkdir(parents=True, exist_ok=True)
|
||
terms: dict[str, dict] = {}
|
||
|
||
def add(t: str, source: str, kind: str):
|
||
t = t.strip()
|
||
if not (2 <= len(t) <= 20):
|
||
return
|
||
e = terms.setdefault(t, {"sources": [], "kind": kind})
|
||
if source not in e["sources"]:
|
||
e["sources"].append(source)
|
||
|
||
print("[1/4] curated CDMO/CMC jargon")
|
||
for t in CDMO_TERMS.split():
|
||
add(t, "cdmo", "工艺术语")
|
||
for t in DOMAIN_ORGS.split():
|
||
add(t, "org", "机构")
|
||
print(f" {len(terms)} terms")
|
||
|
||
print("[2/4] zhwiki INN generic drug names")
|
||
n0 = len(terms)
|
||
for kw in ("单抗", "替尼", "西普", "格鲁肽", "干扰素", "他滨", "疫苗", "沙星"):
|
||
titles = wiki_intitle(kw)
|
||
good = 0
|
||
for t in titles:
|
||
# Keep only names that genuinely END with a drug stem; the search also
|
||
# returns 朱塞佩·塔替尼 (a violinist) and 塔替尼广场.
|
||
if HAN.match(t) and any(t.endswith(s) for s in INN_SUFFIXES):
|
||
add(t, "inn-wiki", "药品通用名")
|
||
good += 1
|
||
print(f" {kw}: {len(titles)} hits -> {good} kept")
|
||
print(f" +{len(terms) - n0} drug names")
|
||
|
||
print("[3/5] jieba general-frequency dictionary -> allowlist")
|
||
allow: set[str] = set()
|
||
try:
|
||
raw = fetch(JIEBA_DICT, 120)
|
||
n = 0
|
||
for line in raw.splitlines():
|
||
parts = line.split()
|
||
if len(parts) < 2 or not parts[1].isdigit():
|
||
continue
|
||
w, f = parts[0], int(parts[1])
|
||
if HAN.match(w) and 2 <= len(w) <= 4:
|
||
n += 1
|
||
if f >= JIEBA_MIN_FREQ:
|
||
allow.add(w)
|
||
print(f" {n} han words scanned, {len(allow)} with freq>={JIEBA_MIN_FREQ}")
|
||
except Exception as e:
|
||
print(f" jieba dict FAILED: {e}", file=sys.stderr)
|
||
|
||
print("[4/5] THUOCL domain wordlists (domain vocabulary)")
|
||
for name, url in THUOCL.items():
|
||
try:
|
||
raw = fetch(url, 90)
|
||
except Exception as e:
|
||
print(f" {name}: FAILED {e}", file=sys.stderr)
|
||
continue
|
||
words = []
|
||
for line in raw.splitlines():
|
||
parts = line.split()
|
||
if not parts:
|
||
continue
|
||
w = parts[0]
|
||
freq = int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else 0
|
||
if HAN.match(w) and 2 <= len(w) <= 8:
|
||
words.append((w, freq))
|
||
# Medical/IT words are domain vocabulary. They deliberately do NOT feed
|
||
# the allowlist: THUOCL frequency is corpus-relative (医院 ranks huge) and
|
||
# using it as a "this is a common word" signal both over-blocks real terms
|
||
# and under-covers the general words that actually cause false positives.
|
||
if name in ("medical", "it"):
|
||
for w, f in words:
|
||
if 3 <= len(w) <= 8 and f < 10000:
|
||
add(w, f"thuocl-{name}", "领域词")
|
||
print(f" {name}: {len(words)} words")
|
||
|
||
print("[5/5] vault glossary + entity cards")
|
||
n0 = len(terms)
|
||
gl = CFG.glossary
|
||
if gl.exists():
|
||
for line in gl.read_text(errors="ignore").splitlines():
|
||
m = re.match(r"^-\s*(.+?)\s*[→⇒➔]\s*(.+)$", line.strip())
|
||
if m:
|
||
right = re.sub(r"（.*?）|\[\[|\]\]", "", m.group(2))
|
||
right = right.split("；")[0].split("/")[0].strip()
|
||
if right:
|
||
add(right, "glossary", "已确认纠错目标")
|
||
else:
|
||
m2 = re.match(r"^-\s*([^\s—:：]+)\s*[—:：]", line.strip())
|
||
if m2:
|
||
add(re.sub(r"\[\[|\]\]", "", m2.group(1)), "glossary", "术语表")
|
||
for sub, kind in (("People", "人名"), ("Companies", "公司")):
|
||
d = CFG.vault / sub
|
||
for f in d.glob("*.md") if d.exists() else []:
|
||
add(f.stem, f"card-{sub.lower()}", kind)
|
||
print(f" +{len(terms) - n0}")
|
||
|
||
# An allowlisted common word must not also be offered as a correction target.
|
||
for w in allow:
|
||
if w in terms and terms[w]["sources"] == ["thuocl-medical"]:
|
||
del terms[w]
|
||
|
||
(BANK / "terms.json").write_text(
|
||
json.dumps(terms, ensure_ascii=False, indent=0), encoding="utf-8")
|
||
(BANK / "allowlist.txt").write_text(
|
||
"\n".join(sorted(allow)), encoding="utf-8")
|
||
(BANK / "inn_suffixes.json").write_text(
|
||
json.dumps({"suffixes": INN_SUFFIXES, "kinds": INN_KIND}, ensure_ascii=False, indent=2),
|
||
encoding="utf-8")
|
||
print(f"\nwrote {len(terms)} terms, {len(allow)} allowlist -> {BANK}")
|
||
|
||
|
||
def stats():
|
||
terms = json.loads((BANK / "terms.json").read_text())
|
||
from collections import Counter
|
||
c = Counter(s for e in terms.values() for s in e["sources"])
|
||
k = Counter(e["kind"] for e in terms.values())
|
||
print(f"total {len(terms)} terms")
|
||
print("by source:", dict(c))
|
||
print("by kind: ", dict(k))
|
||
allow = (BANK / "allowlist.txt").read_text().split()
|
||
print(f"allowlist {len(allow)}: {allow[:15]}")
|
||
drugs = [t for t, e in terms.items() if e["kind"] == "药品通用名"]
|
||
print(f"drug names {len(drugs)}: {drugs[:12]}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
(build if len(sys.argv) < 2 or sys.argv[1] == "build" else stats)()
|