from __future__ import annotations import json import os import subprocess import tempfile import unittest from pathlib import Path from unittest import mock TMP = tempfile.TemporaryDirectory() os.environ["ASR_STATE_DIR"] = str(Path(TMP.name) / "state") os.environ["ASR_VAULT"] = str(Path(TMP.name) / "vault") os.environ["XDG_RUNTIME_DIR"] = TMP.name from asrpipe import cli, context, flagger, graph, lexicon, numbers, review, revise, vault # noqa: E402 from asrpipe.config import CFG # noqa: E402 from asrpipe.refine import _apply_settled, _queue_ambiguous # noqa: E402 from asrpipe import refine # noqa: E402 from asrpipe.segment import Chunk, Turn # noqa: E402 class CoreTests(unittest.TestCase): def setUp(self): CFG.vault.mkdir(parents=True, exist_ok=True) def test_context_compacts_duplicate_recall(self): raw = [ "刘冀是张科领弋的基金投资人，目前处于活跃合作状态。", "刘冀是张科领弋的基金投资人，最近联系日期为2026年6月26日。", "刘冀是张科领弋的基金投资人，目前处于活跃合作状态。", ] out = context.compact(["刘冀"], raw, per_entity=2) self.assertEqual(out.count("目前处于活跃合作状态"), 1) self.assertLessEqual(out.count("\n- "), 2) def test_settled_mapping_is_deterministic_and_common_word_is_safe(self): flags = [ {"span": "老万", "settled": True, "common_word": False, "candidates": [{"target": "万亚坤"}]}, {"span": "公司", "settled": True, "common_word": True, "candidates": [{"target": "工时"}]}, ] out, decisions = _apply_settled("老万说公司暂时不投。", flags) self.assertEqual(out, "万亚坤说公司暂时不投。") self.assertEqual(decisions[0]["surface"], "老万") def test_review_answer_becomes_reusable_alias(self): qid = review.enqueue("fixture-1", "老万是不是万亚坤？", { "surface": "老万", "candidates": ["万亚坤"], "sentence": "老万说洛启的事情"}) result = review.answer(qid, "万亚坤") self.assertTrue(result["learned"]) self.assertIn("万亚坤", [x[0] for x in lexicon.alias_map()["老万"]]) def test_review_answer_is_idempotent(self): qid = review.enqueue("fixture-idempotent", "是否确认？", { "surface": "老张", "candidates": ["张成海"]}) first = review.answer(qid, "张成海") second = review.answer(qid, "另一个人") self.assertTrue(first["learned"]) self.assertTrue(second["already_answered"]) self.assertEqual(second["answer"], "张成海") def test_draft_delivery_queues_questions_and_persists_answers(self): CFG.drafts.mkdir(parents=True, exist_ok=True) path = CFG.drafts / "web-draft.md" path.write_text( "# 草稿\n\n## 通话背景（待确认）\n\n双方讨论项目安排。\n\n" "## 待你确认\n\n" "1. 刘总是 [[刘冀]] 还是 [[刘大涛]]？\n" "2. speaker_0 是谁？\n", encoding="utf-8") count = review.enqueue_draft("web-draft-recording", path) self.assertEqual(count, 4) rows = [r for r in review.pending(20) if r["recording"] == "web-draft-recording"] self.assertEqual(len(rows), 4) self.assertEqual(json.loads(rows[0]["payload"])["kind"], "background_check") question_payload = json.loads(rows[1]["payload"]) self.assertEqual(question_payload["candidates"], ["刘冀", "刘大涛"]) with mock.patch.object(vault, "sync", return_value="disabled"), \ mock.patch.object(review, "_request_revision", return_value=True): for row, answer_value in zip( rows, ["背景基本准确", "刘冀", "谭凯", "草稿整体可用"]): review.answer(row["id"], answer_value) text = path.read_text(encoding="utf-8") self.assertIn("Web 人工确认记录", text) self.assertIn("**回答**：刘冀", text) self.assertIn("**回答**：草稿整体可用", text) def test_draft_delivery_prompts_only_substantive_missing_voiceprints(self): CFG.drafts.mkdir(parents=True, exist_ok=True) transcripts = Path(TMP.name) / "voiceprint-transcript.txt" transcripts.write_text( "【speaker_0】" + "这是高瑶对协议条款和保密范围的详细说明。" * 8 + "\n" "【speaker_2】嗯。\n" "【谭凯】好的。\n", encoding="utf-8") path = CFG.drafts / "voiceprint-draft.md" path.write_text( "---\nstatus: 草稿待确认\n" f"transcript: {transcripts}\n" "voiceprint_session: test-session\n" "voiceprint_speakers:\n speaker_1: 谭凯\n---\n\n" "# 草稿\n\n## 通话背景（待确认）\n\n协议修改沟通。\n\n" "## 待你确认\n\n1. speaker_0 身份是谁？\n", encoding="utf-8") review.enqueue_draft("voiceprint-recording", path) rows = [r for r in review.pending(20) if r["recording"] == "voiceprint-recording"] kinds = [json.loads(r["payload"]).get("kind") for r in rows] self.assertEqual(kinds.count("voiceprint"), 1) voice = next(json.loads(r["payload"]) for r in rows if json.loads(r["payload"]).get("kind") == "voiceprint") self.assertEqual(voice["speaker_id"], "speaker_0") self.assertFalse(any("speaker_2" in r["question"] for r in rows)) def test_entity_card_suggestion_counts_distinct_formal_sources(self): con = lexicon.connect() con.execute( "INSERT INTO entity_mention(entity,kind,source,occurred_at) " "VALUES(?,?,?,?)", ("高瑶", "人名", "minute:older.md", 1)) con.commit() con.close() CFG.drafts.mkdir(parents=True, exist_ok=True) path = CFG.drafts / "entity-suggestion.md" path.write_text( "---\npeople: [高瑶]\ncompanies: []\n---\n\n" "# 草稿\n\n## 通话背景（待确认）\n\n协议沟通。\n\n" "## 待你确认\n\n", encoding="utf-8") review.enqueue_draft("entity-suggestion-recording", path) rows = [r for r in review.pending(20) if r["recording"] == "entity-suggestion-recording"] cards = [json.loads(r["payload"]) for r in rows if json.loads(r["payload"]).get("kind") == "entity_card"] self.assertEqual(len(cards), 1) self.assertEqual(cards[0]["entity"], "高瑶") self.assertEqual(cards[0]["entity_kind"], "人名") def test_first_review_prepares_preview_without_publishing(self): CFG.drafts.mkdir(parents=True, exist_ok=True) path = CFG.drafts / "publish-fixture.md" path.write_text( "---\ndate: 2026-07-26\ntype: minutes\n" "tags: [会议纪要, 草稿, 待确认]\nsource: 录音转录\n" "status: 草稿待确认\npeople: [谭凯]\ncompanies: []\n" "transcript: /archive/publish.txt\n---\n\n# 粗整理草稿\n\n" f"{review.FEEDBACK_START}\n## Web 人工确认记录\n\n" "1. **问题**：背景？\n - **回答**：准确\n\n" f"{review.FEEDBACK_END}\n", encoding="utf-8") formal = """--- date: 2026-07-26 type: minutes tags: [会议纪要] source: 录音转录 people: [谭凯] companies: [] transcript: /archive/publish.txt --- # 谭凯通话测试 > [!summary] Summary > 测试正式发布。 """ with mock.patch.object(revise, "chat", return_value=formal), \ mock.patch("asrpipe.review_web.request_start", return_value=True): self.assertTrue(revise.revise(path)) self.assertTrue(path.exists()) self.assertIn("status: 待发布确认", path.read_text()) approvals = [ r for r in review.pending(100) if json.loads(r["payload"]).get("draft_path") == str(path) and json.loads(r["payload"]).get("kind") == "publish_approval" ] self.assertEqual(len(approvals), 1) destination = CFG.minutes / "2026-07-26-谭凯通话测试.md" self.assertFalse(destination.exists()) with mock.patch.object(review, "_request_revision", return_value=False): review.answer(approvals[0]["id"], "发布正式稿") with mock.patch.object(vault, "sync", return_value="pushed"), \ mock.patch.object(vault, "notify"), \ mock.patch.object(graph, "sync", return_value={}): self.assertTrue(revise.revise(path)) self.assertTrue(destination.exists()) self.assertNotIn("草稿待确认", destination.read_text()) self.assertNotIn("待发布确认", destination.read_text()) self.assertFalse(path.exists()) self.assertTrue((CFG.state_dir / "published" / path.name).exists()) def test_normalize_can_disable_review_side_effects(self): chunk = Chunk(0, [Turn("", "老万说洛启项目。", 0)]) flags = [{ "span": "老万", "settled": False, "common_word": False, "sentence": "老万说洛启项目。", "candidates": [{ "target": "万亚坤", "kind": "人名", "score": 0.9, "rule": "alias", "source": "alias-card", }], }] usage = refine.Usage() with mock.patch.object(refine, "_queue_ambiguous") as queue, \ mock.patch.object(refine, "chat_many", return_value=["老万说洛启项目。"]): _parts, _decisions, stats = refine.normalize_chunks( [chunk], [chunk.body], flags, "", "dry-run-fixture", usage, queue_review=False) queue.assert_not_called() self.assertEqual(stats["review_queued"], 0) def test_atomic_write_and_recovery_lookup(self): CFG.drafts.mkdir(parents=True, exist_ok=True) p = CFG.drafts / "fixture.md" vault.write_draft_file(p, "---\ntranscript: /archive/a.txt\n---\nbody") self.assertEqual(vault.find_draft_for_transcript("/archive/a.txt"), p) leftovers = list(CFG.drafts.glob(".fixture.md.*")) self.assertEqual(leftovers, []) def test_relation_card_is_structured(self): lexicon.upsert_relation("刘冀", "投资", "张科领弋", confidence=0.9, source="test") card = lexicon.relationship_card({"刘冀"}) self.assertTrue(any("刘冀 投资 张科领弋" in x for x in card)) def test_confirmed_correction_enters_personal_confusion_channel(self): lexicon.learn_correction("张泰海", "张成海", confirmed=1, note="ASR fixture") rows = lexicon.confusion_map()["张泰海"] self.assertTrue(any(canon == "张成海" and confirmed for canon, _n, confirmed in rows)) def test_confirmed_correction_teaches_directional_syllable_error(self): lexicon.learn_correction("张泰海", "张成海", confirmed=1) flagger._PERSONAL_SYL = None heard, intended = flagger.to_syls("泰")[0], flagger.to_syls("成")[0] self.assertGreater(flagger.syl_sim(heard, intended, personal=True), 0.90) self.assertLess(flagger.syl_sim(heard, intended, personal=False), 0.90) def test_graph_indexes_minutes_and_recency(self): for sub in ("People", "Companies", "Projects", "minutes"): (CFG.vault / sub).mkdir(parents=True, exist_ok=True) (CFG.vault / "People" / "刘冀.md").write_text( "---\ncompany: 张科领弋\nrole: 投资人\n---\n", encoding="utf-8") (CFG.vault / "Companies" / "张科领弋.md").write_text( "---\n---\n", encoding="utf-8") (CFG.vault / "Projects" / "杭州基金.md").write_text( "---\n---\n", encoding="utf-8") (CFG.vault / "minutes" / "2026-07-25-test.md").write_text( "---\ndate: 2026-07-25\npeople: [刘冀]\ncompanies: [张科领弋]\n" "---\n[[刘冀]]参与[[杭州基金]]，并与[[张科领弋]]合作。\n", encoding="utf-8") result = graph.sync(force=True) self.assertGreaterEqual(result["relations_written"], 3) weights = graph.context_weights({"刘冀"}) self.assertIn("张科领弋", weights) self.assertGreater(weights["张科领弋"], 0) def test_number_normalization_is_typed_and_conservative(self): text = "七月十日打款一千万元，BF四幺八二项目占百分之十五，共三十人，两万五九份待核。" out, decisions = numbers.normalize(text) self.assertIn("7月10日", out) self.assertIn("1000万元", out) self.assertIn("BF4182", out) self.assertIn("15%", out) self.assertIn("30人", out) self.assertIn("两万五九份", out) self.assertTrue(any(d["raw"] == "两万五九" and d["confidence"] < 0.8 for d in decisions)) def test_generic_discourse_counter_stays_chinese(self): out, _ = numbers.normalize("我同步两个事，第一个是条款，第二个是公告。") self.assertEqual(out, "我同步两个事，第一个是条款，第二个是公告。") def test_financial_scale_without_currency_is_normalized_from_context(self): out, decisions = numbers.normalize("八月份投资两千万，后续再退出。") self.assertIn("投资2000万元", out) self.assertTrue(any(d["kind"] == "money" for d in decisions)) def test_process_lock_remains_held(self): handle = cli._lock() lock_path = str(Path(TMP.name) / "asr-pipeline.lock") code = ( "import fcntl,sys;" f"f=open({lock_path!r},'w');" "\ntry: fcntl.flock(f,fcntl.LOCK_EX|fcntl.LOCK_NB)" "\nexcept OSError: sys.exit(23)" "\nsys.exit(0)" ) r = subprocess.run([os.sys.executable, "-c", code]) self.assertEqual(r.returncode, 23) handle.close() cli._LOCK_HANDLE = None def test_staged_pipeline_smoke_without_network(self): old = (CFG.use_termbank, CFG.use_hindsight, CFG.use_keyword_recall, CFG.use_normalize, CFG.use_evidence) CFG.use_termbank = False CFG.use_hindsight = False CFG.use_keyword_recall = False CFG.use_normalize = False CFG.use_evidence = True def fake_many(_model, jobs, *, system=None, **_kw): if system == refine.prompts.CLEAN_SYS: return ["【谭凯】讨论洛启项目，下一步确认投资条款。" for _ in jobs] return ["- [片段 0] 决策｜洛启项目｜下一步确认投资条款｜置信度:明确" for _ in jobs] try: with mock.patch.object(refine, "chat_many", side_effect=fake_many), \ mock.patch.object(refine, "chat", return_value="# 粗整理\n\n确认投资条款。"): result = refine.refine( "【谭凯】嗯，讨论洛启项目，然后下一步确认投资条款。", date="2026-07-26", transcript_path="/archive/test.txt") self.assertIn("确认投资条款", result.draft) self.assertIn("[片段 0]", result.evidence) self.assertEqual(result.stats["n_chunks"], 1) finally: (CFG.use_termbank, CFG.use_hindsight, CFG.use_keyword_recall, CFG.use_normalize, CFG.use_evidence) = old def test_pure_phonetic_guess_does_not_interrupt_human(self): groups = [{ "span": "也说", "settled": False, "common_word": False, "sentence": "但是他也说他肯定不会反驳", "candidates": [{ "target": "叶硕", "kind": "人名", "score": 0.91, "rule": "full", "source": "lexicon-entity", "boost": 0, "support": "", }], }] self.assertEqual(_queue_ambiguous("fixture-junk", groups), 0) def test_context_adjudicator_must_explicitly_request_human_review(self): groups = [{ "span": "老万", "settled": False, "common_word": False, "sentence": "老万说洛启项目。", "candidates": [{ "target": "万亚坤", "kind": "人名", "score": 0.92, "rule": "alias", "source": "alias-card", "boost": 1, "support": "关系图", }], }] self.assertEqual( _queue_ambiguous("fixture-context-decided", groups, "老万说洛启项目。"), 0) self.assertEqual( _queue_ambiguous("fixture-context-uncertain", groups, "[?万亚坤]说洛启项目。"), 1) def test_common_words_block_embedded_name_guesses(self): cases = [ ("小问", "UPS是某个小问题", ["万亚坤", "万津"]), ("用介", "包括所有的这些公用介质报告", ["邓永杰"]), ("们总", "我上次问他们总监", ["卢萌"]), ("经签", "所有方案已经签掉", ["徐劲前"]), ("统统", "三批之间统统计一下", ["薛彤彤"]), ("峰总", "雷峰总这边的开发部", ["冯昆仑"]), ] for i, (surface, sentence, choices) in enumerate(cases): groups = [{ "span": surface, "settled": False, "common_word": False, "sentence": sentence, "candidates": [{ "target": target, "kind": "人名", "score": 0.92, "rule": "alias", "source": "alias-card", "boost": 1, "support": "关系图", } for target in choices], }] marker = f"[?{'/'.join(choices)}]" self.assertEqual( _queue_ambiguous(f"fixture-common-{i}", groups, marker), 0) def test_archive_resolution_uses_content_when_timestamp_collides(self): old_obsidian = CFG.obsidian root = Path(TMP.name) / "archive-resolution" archive = root / "recordings" / "transcripts" archive.mkdir(parents=True, exist_ok=True) inbox = root / "2026-07-27-095902-录音-2.txt" inbox.write_text("wav transcript", encoding="utf-8") mp3 = archive / "aaa-2026-07-27_09_59_02.txt" wav = archive / "bbb-2026-07-27_09_59_02.txt" mp3.write_text("mp3 transcript", encoding="utf-8") wav.write_text("wav transcript", encoding="utf-8") CFG.obsidian = root try: self.assertEqual( cli.resolve_archive("2026-07-27-095902-录音-2", inbox), str(wav)) finally: CFG.obsidian = old_obsidian def test_observed_neighbour_needs_more_than_recent_weight(self): groups = [{ "span": "们七", "settled": False, "common_word": False, "sentence": "你们七月二十号签完", "candidates": [{ "target": "范梦奇", "kind": "人名", "score": 0.96, "rule": "observed", "source": "observed-correction", "via": "范梦奇", "boost": 1.2, "support": "近期权重4.0", }], }] self.assertEqual(_queue_ambiguous("fixture-observed-junk", groups), 0) if __name__ == "__main__": unittest.main()