如何实现俄语重音标注：从 Anki 插件到 51 万词形词库的工程实践

本文记录我们在项目中实现俄语重音（ударение）标注 API 的完整技术路径。不讨论产品功能列表，只谈：问题怎么拆、数据怎么建、代码怎么写、踩了哪些坑。

消费端示例见 鹅语菌 app

1. 要解决的核心工程问题

俄语重音在文本层面的表示并不复杂——在元音上附加 Unicode 组合锐音 U+0301 即可，例如 сегодня → сего́дня。真正困难的是：

• 重音位置没有算法可推导，必须依赖词典或外部服务；
• 同一词形可能有多种读法（同形异读 homograph），如 реки → ре́ки / реки́；
• 
• Unicode 规范化做不好会导致词库明明有条目却查不到；
• 第三方在线服务不可靠，不能直接当唯一数据源；
• 客户端高并发场景（变格表 12 格同时渲染）会把单查 API 打爆。

我们的解法可以概括为三句话：

• 离线词库兜底（OpenRussian 51 万+ 词形）
• RussianGram 在线补全（继承 anki-russian-stressmark 逻辑）
• 客户端 in-flight 去重 + 批量预取

2. 整体架构

系统架构

┌──────────────┐     POST /api/stress      ┌─────────────────────────┐
│ Flutter 客户端 │ ────────────────────────→ │ Flask app.py            │
│ StressService │     POST /api/stress/batch │ RussianStressEngine     │
└──────────────┘                            │  ├─ LRU cache           │
                                            │  ├─ common_words_dict  │
                                            │  ├─ homographs_dict    │
                                            │  └─ RussianGram (fallback)│
                                            └─────────────────────────┘

引擎是单进程内存字典查询，没有数据库，没有 ES。词库在 init 时一次性加载进 dict，查询复杂度 O(1)。

3. 起点：anki-russian-stressmark 给了我们什么

anki-russian-stressmark 是一个 Anki 插件，核心逻辑只有一件事：向 RussianGram.com POST 表单，从返回 HTML 的 textarea 里抠出带重音的文本。

我们保留了它的两个关键组件：

但插件代码不能直接用于生产，原因后面会讲。我们的改造是把这段逻辑封进 RussianStressEngine，并在它前面加了一层本地词库查询。

4. Unicode 层：一切问题的一半

4.1 组合锐音不是「多一个字符」那么简单

сего́дня 在 Python 里是 8 个 code point（е 后面跟了 U+0301），但视觉上仍是 7 个字母。所有大小写回写、重音检测逻辑都必须按**字素簇（grapheme cluster）**处理，不能 len(str) 硬比。

Unicode 组合锐音

引擎里的字素拆分：

@staticmethod
def _stress_grapheme_clusters(s: str) -> List[str]:
    """将基字符与紧随其后的组合用音标（Mn）拆成可视字素段。"""
    out: List[str] = []
    i = 0
    n = len(s)
    while i < n:
        if unicodedata.category(s[i]) == "Mn":
            i += 1
            continue
        start = i
        i += 1
        while i < n and unicodedata.category(s[i]) == "Mn":
            i += 1
        out.append(s[start:i])
    return out

大小写回写 _apply_case_from_cleaned 就是靠这个函数把用户输入的 Москва 和词库里的 москва́ 逐字素对齐，再按用户输入的大小写还原为 Москва́。

4.2 查询前必须剥重音 + NFC 规范化

词库键是小写、无重音、NFC 规范化的 bare 形。如果用户输入已经带了重音，或者输入法产生了 NFD 形式，直接查会 miss。

@staticmethod
def _normalize_word_for_lookup(word: str) -> str:
    w = unicodedata.normalize("NFC", word)
    w = re.sub(r"[\u200b\u200c\u200d\ufeff]", "", w)
    return w

@staticmethod
def _strip_stress_marks(word: str) -> str:
    return re.sub(r"[\u0301\u00B4\u0485\u0311\u0313\u0027\u2019\u02BC\u0060]", "", word)

get_stress_mark 入口：

cleaned_word = self._strip_stress_marks(
    self._normalize_word_for_lookup(self._clean_word(word))
)
lookup_key = cleaned_word.lower()

4.3 Flutter 端：为什么不用 RegExp

Dart 端 RussianStressTextUtils 与 Python 语义对齐，但实现上刻意不用 Unicode 正则——iOS 模拟器 Dart Worker 编译复杂 RegExp 时会 EXC_BAD_ACCESS。改为逐 code unit 判断：

static bool hasStressMark(String source) {
  for (int i = 0; i < source.length; i++) {
    final int codeUnit = source.codeUnitAt(i);
    if (_isKnownStressCodeUnit(codeUnit) ||
        _isApostropheStressCodeUnit(codeUnit)) {
      return true;
    }
  }
  return false;
}

同一份业务规则，两端各用各自平台最稳的实现方式——这是典型的跨端工程折中。

5. 引擎核心：get_stress_mark 走读

查询生命周期

5.1 分层回退

数据源优先级

def get_stress_mark(self, word: str) -> StressResult:
    cleaned_word = self._strip_stress_marks(
        self._normalize_word_for_lookup(self._clean_word(word))
    )
    if not cleaned_word:
        return StressResult(..., source='validation_error')

# 缓存键含词库签名，部署新词库后旧缓存自动失效
    cache_key = f"{self._lexicon_signature}:{cleaned_word.lower()}"
    if cache_key in self.cache:
        return self._finalize_stress_result(cleaned_word, cached_result)

result = None
    lookup_key = cleaned_word.lower()

# ① 本地词库
    if lookup_key in self.common_words_dict:
        stressed_word = self.common_words_dict[lookup_key]
        has_marker = self._stressed_form_has_marker(stressed_word)
        homograph_forms = self.homographs_dict.get(lookup_key, [])
        alternatives = self._build_stress_alternatives(stressed_word, homograph_forms)
        if has_marker:
            result = StressResult(
                original_word=word,
                stressed_word=stressed_word,
                has_stress=True,
                confidence=0.95,
                source='verified_lexicon',
                alternatives=alternatives,
            )
        # 词库有条目但无重音标记 → 不采用，继续走在线补全

# ② RussianGram
    if result is None:
        stressed_word, alternatives = self._fetch_from_russiangram(cleaned_word)
        if stressed_word and self._stressed_form_has_marker(stressed_word):
            result = StressResult(..., source='russiangram', confidence=0.8)

# ③ 全部失败
    if result is None:
        result = StressResult(..., source='no_source', has_stress=False)

self._update_cache(cache_key, result)
    return self._finalize_stress_result(cleaned_word, result)

几个值得单独说的设计点：

词库签名嵌入缓存键

def _compute_lexicon_signature() -> str:
    parts: List[str] = []
    for path in (_OPENRUSSIAN_LEXICON_PATH, _MANUAL_LEXICON_PATH, ...):
        st = path.stat()
        parts.append(f"{path.name}:{st.st_mtime_ns}:{st.st_size}")
    return "|".join(parts)

不手动清缓存，词库文件一更新，mtime 变，缓存键变，旧结果自然失效。

只缓存有重音的结果

def _update_cache(self, key: str, result: StressResult):
    if not result.has_stress or result.source in (
        'no_source', 'validation_error', 'error', 'verified_lexicon_no_marker',
    ):
        return
    # LRU 写入...

我们曾经缓存过 no_source，导致 RussianGram 一次超时后该词永远查不到重音。负结果不缓存是硬性规则。

ё 也算重音

def _stressed_form_has_marker(stressed: str) -> bool:
    if "\u0301" in stressed or "\u00B4" in stressed:
        return True
    return "ё" in stressed or "Ё" in stressed

6. 词库构建：从 OpenRussian CSV 到内存 dict

6.1 为什么不用运行时查数据库

51 万词形，每次查询走 SQLite/Redis 都多余。启动时 dict 加载进内存，查询就是哈希查找。代价是进程启动时要读一个大 JSON（约 30~45 MB 紧凑格式），用 orjson 加速：

@staticmethod
def _parse_lexicon_json_bytes(data: bytes) -> Any:
    try:
        import orjson
        return orjson.loads(data)
    except ImportError:
        return json.loads(data.decode("utf-8"))

Gunicorn 配置 preload_app=True，master 进程加载一次，worker fork 后共享只读内存（copy-on-write）。

6.2 CSV 导入：撇号 → U+0301

OpenRussian 原始数据用 ' 标记重音，如 сего'дня。导入脚本 import_openrussian_stress.py 负责转换：

def _apostrophe_to_combining_acute(text: str) -> str:
    if not text or "'" not in text:
        return unicodedata.normalize("NFC", text)
    parts: List[str] = []
    i = 0
    while i < len(text):
        ch = text[i]
        if ch == "'":
            i += 1
            continue
        parts.append(ch)
        if i + 1 < len(text) and text[i + 1] == "'":
            parts.append("\u0301")
        i += 1
    return unicodedata.normalize("NFC", "".join(parts))

键的生成：带重音词形 → NFC → lower → 剥重音。值：保留重音的完整词形。

6.3 分文件存储与覆盖优先级

--split 模式输出：

openrussian_stress_lexicon.json   # ~51 万 imported
russian_stress_lexicon_manual.json # 人工修正，同键覆盖

引擎加载顺序：

def _load_common_words(self) -> Dict[str, str]:
    result: Dict[str, str] = {}
    if _OPENRUSSIAN_LEXICON_PATH.is_file():
        result.update(self._load_entries_from_json_path(_OPENRUSSIAN_LEXICON_PATH))
    if _MANUAL_LEXICON_PATH.is_file():
        result.update(self._load_entries_from_json_path(_MANUAL_LEXICON_PATH))
    return result

dict.update 后者覆盖前者——人工修正不需要改 51 万条的大文件，只维护一个小 JSON。

6.4 同形异读表：与主词库分离

主词库每个键只能存一个「主读音」。同形异读（同一 bare 形对应多种重音）单独维护在 russian_stress_homographs.json，由 build_homograph_lexicon.py 从 CSV 扫描生成：

# 仅当同一 bare 键有 >= 2 种不同重音形式时写入
if len(forms) >= 2:
    result[nk] = forms

查询时合并：

homograph_forms = self.homographs_dict.get(lookup_key, [])
alternatives = self._build_stress_alternatives(stressed_word, homograph_forms)

_build_stress_alternatives 把主读音和 homograph 列表去重合并，仅当 ≥2 种读法时才返回 alternatives，否则返回空列表。

7. RussianGram 在线补全：生产环境必须修的 bug

RussianGram 流程

插件原版最大的坑：硬编码 __VIEWSTATE。ASP.NET WebForms 的 viewstate 与会话绑定，几小时后全部 403/500。

正确做法：每次查询先 GET 首页，解析隐藏域，再 POST：

def _fetch_from_russiangram(self, word: str) -> Tuple[Optional[str], List[str]]:
    session = requests.Session()
    get_resp = session.get("https://russiangram.com/", headers=headers, timeout=15)
    html = get_resp.text
    view_state = self._find_aspnet_hidden_value(html, "__VIEWSTATE")
    view_gen = self._find_aspnet_hidden_value(html, "__VIEWSTATEGENERATOR")
    event_val = self._find_aspnet_hidden_value(html, "__EVENTVALIDATION")

form_data = {
        "__VIEWSTATE": view_state,
        "__VIEWSTATEGENERATOR": view_gen,
        "ctl00$MainContent$UserSentenceTextbox": word,
        "ctl00$MainContent$SubmitButton": "Annotate",
    }
    if event_val is not None:
        form_data["__EVENTVALIDATION"] = event_val

post_resp = session.post("https://russiangram.com/", data=form_data, ...)
    rgparser = RussiangramParser()
    rgparser.feed(post_resp.content.decode("utf-8", errors="ignore"))
    stressed_text = rgparser.data.strip()
    return self._extract_stressed_word(stressed_text, word)

HTML 解析器处理 textarea 里可能出现的 \x 十六进制编码：

def handle_data(self, data):
    if self.intextarea:
        if '\\x' in data:
            hex_parts = []
            for part in data.split('\\x'):
                clean_part = ''.join(c for c in part if c in '0123456789abcdefABCDEF')
                if clean_part:
                    hex_parts.append(clean_part)
            bs = bytes(int(x, 16) for x in hex_parts)
            self.data = bs.decode('utf-8', errors='ignore')
        else:
            self.data = data

_extract_stressed_word 处理 RussianGram 返回的 | 分隔多候选：

if '|' in stressed_text:
    options = [opt.strip() for opt in stressed_text.split('|')]
for option in options:
    words = re.findall(r'[А-Яа-яЁё\u0301]+', option)
    for word in words:
        word_without_stress = re.sub(r'\u0301', '', word)
        if word_without_stress.lower() == original_word.lower():
            if '\u0301' in word:
                alternatives.append(word)

性能代价：每次在线补全 600ms~1.6s（含两次 HTTP）。这就是为什么词库命中必须走内存 dict，在线只作 fallback。

8. API 层：薄封装，不做业务

app.py 只做 JSON 序列化和参数校验，核心逻辑全在引擎：

@app.route('/api/stress', methods=['POST'])
def add_stress():
    data = request.get_json()
    word = data['word']
    result = stress_api.get_stress_mark(word)
    return jsonify(result)

RussianStressMarkAPI.get_stress_mark 额外计算两个字段：

stress_count = len(result.alternatives) if result.alternatives else 1
return {
    'success': True,
    'stressed_word': result.stressed_word,
    'has_stress': result.has_stress,
    'confidence': result.confidence,
    'source': result.source,
    'alternatives': result.alternatives,
    'stress_options_count': stress_count,
    'is_multi_stress': stress_count > 1,
}

批量接口 batch_process 就是循环调用 get_stress_mark——没有专门的批量优化，因为词库命中时单次已经是 O(1)，瓶颈只在 RussianGram 在线路径。客户端批量请求的价值在于减少 HTTP 往返次数，不是减少服务端计算。

还有一个辅助接口 has_lexicon_entry，只做词形存在性判断，不调 RussianGram，给后端词典拼写校验用：

def has_lexicon_entry(self, word: str) -> bool:
    key = self._normalize_strict_lexicon_key(word)
    return key in self.common_words_dict

_normalize_strict_lexicon_key 比查询用的 _clean_word 更严格——禁止数字、emoji、拉丁字母，防止 test123 被当成有效词形。

9. Flutter 客户端：并发控制比 UI 更重要

9.1 问题：变格表同时发起 N 次单查

词典形态学展开后，一个名词可能有 12 个变格格位，每个格位渲染一个 StressedRussianText，如果各自独立 getStress，就是 12 个并发 HTTP 请求查询 12 个不同的词形。

9.2 解法一：in-flight 去重

同一 normalized 词形，并发只发一次请求：

static final Map<String, Future> _stressInflight = {};

Future<StressResult> getStress(String word) async {
  final String normalized = RussianStressTextUtils.normalizeForStressQuery(word);
  // ... 缓存检查 ...
  final Future<StressResult>? inflight = _stressInflight[normalized];
  if (inflight != null) return inflight;

final Future<StressResult> future = _fetchStressFromNetwork(normalized);
  _stressInflight[normalized] = future;
  try {
    return await future;
  } finally {
    _stressInflight.remove(normalized);
  }
}

9.3 解法二：批量预取 + revision 通知

StressDisplayCache 在形态学结果返回后，收集全部词形，后台批量查询：

Future<void> _fetchBatches(List<String> words) async {
  const int chunkSize = 100;
  for (int offset = 0; offset < words.length; offset += chunkSize) {
    final List<String> chunk = words.sublist(offset, end);
    final BatchStressResult batch = await _stressService.getBatchStress(chunk);
    for (final StressResult result in batch.results) {
      if (result.success && result.hasStress) {
        _stressedByKey[_cacheKey(result.originalWord)] = result.displayStressedText;
      }
    }
    revision.value = revision.value + 1;  // 通知所有 StressedRussianText 刷新
  }
}

StressedRussianText 监听 revision：

StressDisplayCache.instance.revision.addListener(_onCacheRevision);

这样表格里每个格子不需要自己打 API，等批量预取完成后统一更新。

9.4 解法三：批量与单查共享 in-flight

getBatchStress 为 batch 中每个词注册 in-flight Future，指向同一个 batch 请求：

final Future<BatchStressResult> batchFuture = _fetchBatchStressFromNetwork(uniqueBatchWords);
for (final String word in uniqueBatchWords) {
  _stressInflight[word] = batchFuture.then(
    (BatchStressResult batch) => _pickStressFromBatch(batch, word),
  );
}

避免「单查和 batch 同时发出」的重复请求。

9.5 客户端缓存：只记正结果

客户端缓存层级

// 仅缓存有效重音，避免历史 API 的 has_stress:false 长期挡住正确结果
if (result.success && result.hasStress) {
  await _saveToCache(normalized, result);
} else {
  await _removeCacheEntry(normalized);
  if (result.success && !result.hasStress) {
    _rememberNoStress(normalized);  // 会话级 30min 负缓存
  }
}

持久缓存用 stress_cache_v2_ 前缀，TTL 7 天。旧版 stress_cache_ 曾缓存过负结果（如 месяц），升级后读取时直接删键强制重新请求。

9.6 TTS 与重音符号：展示链路和合成链路分离

Edge TTS 无法正确处理 U+0301，会把 сего́дня 拆成两个词。朗读前必须剥重音：

RussianStressTextUtils.stripStressMarks(paragraph)

这和引擎里的 _strip_stress_marks 是同一套规则，两端各实现一份。

10. 测试：直接测引擎，不测 HTTP

test_stress_corpus.py 绕过 Flask，直接实例化 RussianStressEngine：

engine = RussianStressEngine(cache_size=4000)
r = engine.get_stress_mark("сегодня")
assert r.source == "verified_lexicon"
assert r.stressed_word == "сего́дня"

三种模式：

词库一致性校验对 51 万条全量太慢，超过 800 条时改为抽检：

if len(all_keys) > 800:
    keys_to_check = sorted(set(_LEXICON_SPOT_CHECK))

11. 踩坑清单（技术向）

12. 如果从头做，我会怎么选

回顾整个实现，几个关键取舍：

• 内存 dict vs 数据库：51 万词形、只读、O(1) 查询 → dict 够用，简单可靠。
• 词库 vs 纯在线：在线服务不稳定 → 词库优先，在线补全，confidence 区分来源。
• 同形异读独立表 vs 主词库存列表：主词库保持 Dict[str, str] 简单结构，异读表单独维护，查询时 merge。
• 负结果不缓存：避免一次失败永远失败，客户端和服务端统一规则。
• 批量接口循环单查 vs 专门优化：词库命中时无差别，不值得增加批量专用逻辑；客户端批量是为减 HTTP 往返。
• 跨端 Unicode 处理各写一套：Python 用 unicodedata，Dart 避开 RegExp，语义对齐即可。

附录：核心文件索引