MB-X Bilibili Pipeline
6 days ago 873dca205129f123d5ea9dd768bfa1c2e5452830
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
from __future__ import annotations
 
import json
import sys
import tempfile
import unittest
from pathlib import Path
 
 
MODULE_DIR = Path(__file__).resolve().parents[1] / "meeting_minutes_tool"
sys.path.insert(0, str(MODULE_DIR))
 
import meeting_minutes_tool as tool  # noqa: E402
 
 
class TranscriptIndexTests(unittest.TestCase):
    def test_index_extracts_navigation_and_flags_regression(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            transcript = root / "sample.txt"
            transcript.write_text(
                "\n".join(
                    [
                        "甲 共享音频(00:00:10): 正式开始,政治局政策与算力网投资约21.4%。",
                        "甲 共享音频(00:00:20): 请问华为和半导体公司怎么看?",
                        "甲 共享音频(00:00:05): 这个数字大概需要核对,不构成投资建议。",
                        "甲 共享音频(00:00:30): 一段足够长并且会重复出现的测试内容用于重复检测。",
                        "甲 共享音频(00:00:40): 一段足够长并且会重复出现的测试内容用于重复检测。",
                    ]
                ),
                encoding="utf-8",
            )
            result = tool.build_transcript_index(transcript, root / "index", max_items=20)
 
            self.assertEqual(result["source"]["line_count"], 5)
            self.assertEqual(result["sequence_check"]["timestamp_regression_count"], 1)
            self.assertEqual(
                result["sequence_check"]["exact_normalized_duplicate_group_count"], 1
            )
            self.assertGreater(
                result["category_total_counts"]["numbers_and_time_windows"], 0
            )
            self.assertTrue((root / "index" / "transcript-index.json").is_file())
            self.assertTrue((root / "index" / "transcript-index.md").is_file())
 
    def test_missing_transcript_fails_fast(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            with self.assertRaises(FileNotFoundError):
                tool.build_transcript_index(root / "missing.txt", root / "index")
 
 
class PdfPipelineTests(unittest.TestCase):
    SAMPLE_MARKDOWN = """# 2026年度超长标题会议纪要【第6场:政策与产业链深度解读和后续跟踪安排】
 
> **免责声明:内容为会议观点记录,不构成投资建议。**
 
## 一、核心结论
 
1. 这是第一条核心结论。
2. 这是第二条核心结论。
 
## 二、主体内容
 
### 2.1 政策与产业
 
- 算力网与新型电网需要结合订单验证。
- 无法确认的数据需要标记为待核对。
 
| 行业 | 观点 | 风险 |
|---|---|---|
| 算力 | 关注订单和交付 | 估值与技术迭代 |
| 电网 | 关注招标与资本开支 | 项目节奏 |
 
## 三、问答环节重点
 
### 3.1 如何判断?
 
- 回答:结合基本面、价格和风险承受能力判断。
 
## 四、后续跟踪清单
 
1. 跟踪订单。
2. 跟踪正式政策。
"""
 
    def test_pdf_and_qa_roundtrip(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            markdown = root / "minutes.md"
            pdf = root / "minutes.pdf"
            qa_dir = root / "qa"
            markdown.write_text(self.SAMPLE_MARKDOWN, encoding="utf-8")
 
            pdf_result = tool.render_markdown_pdf(markdown, pdf)
            qa_result = tool.run_pdf_qa(markdown, pdf, qa_dir, dpi=96)
 
            self.assertGreater(pdf_result["pdf_bytes"], 10_000)
            self.assertTrue(pdf.is_file())
            self.assertTrue(qa_result["automated_checks_passed"])
            self.assertTrue(qa_result["manual_visual_review_required"])
            self.assertEqual(
                qa_result["pdf"]["page_count"],
                len(qa_result["render"]["page_images"]),
            )
            self.assertTrue((qa_dir / "qa-report.json").is_file())
            self.assertTrue((qa_dir / "qa-report.md").is_file())
            payload = json.loads((qa_dir / "qa-report.json").read_text(encoding="utf-8"))
            self.assertEqual(payload["schema_version"], "meeting-minutes-pdf-qa.v1")
 
    def test_invalid_markdown_does_not_leave_pdf(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            markdown = root / "invalid.md"
            pdf = root / "invalid.pdf"
            markdown.write_text("没有一级标题。", encoding="utf-8")
            with self.assertRaises(ValueError):
                tool.render_markdown_pdf(markdown, pdf)
            self.assertFalse(pdf.exists())
 
 
if __name__ == "__main__":
    unittest.main()