from __future__ import annotations
|
|
import argparse
|
import hashlib
|
import html
|
import json
|
import os
|
import re
|
import shutil
|
import subprocess
|
import sys
|
import tempfile
|
import time
|
from collections import Counter, defaultdict
|
from datetime import datetime, timezone
|
from pathlib import Path
|
from typing import Any, Iterable, Sequence
|
|
|
TOOL_VERSION = "1.0.0"
|
DEFAULT_MAX_ITEMS = 500
|
DEFAULT_RENDER_DPI = 160
|
DISCLAIMER = "内容为会议观点记录，不构成投资建议"
|
|
TIMESTAMP_RE = re.compile(r"\((\d{1,2}):(\d{2}):(\d{2})\)")
|
SPEAKER_PREFIX_RE = re.compile(r"^.*?共享音频\(\d{1,2}:\d{2}:\d{2}\)\s*:\s*")
|
NUMBER_RE = re.compile(
|
r"(?:\d+(?:\.\d+)?\s*(?:%|％|万亿|亿元|亿|万元|万|吨|万吨|倍|点|元)|"
|
r"百分之|三季度|四季度|年底|明年|后年|20\d{2}年?)"
|
)
|
|
CATEGORY_PATTERNS: dict[str, re.Pattern[str]] = {
|
"sequence_and_sections": re.compile(
|
r"正式开始|开始今天|接着讲|下面讲|第一部分|第二部分|第三部分|总结一下|"
|
r"问答|答疑|还有什么问题|下播|今天就到这|会议结束"
|
),
|
"policy_and_macro": re.compile(
|
r"政治局|中央经济工作会议|财政|货币|央行|专项债|特别国债|两重|两新|"
|
r"政策|GDP|投资|消费|内需|房地产|地方债|统一大市场|反内卷"
|
),
|
"industry_and_technology": re.compile(
|
r"AI|人工智能|算力|芯片|半导体|光模块|光芯片|PCB|存储|液冷|EDA|"
|
r"机器人|电网|通信网|物流网|水网|管网|航天|创新药|稀土|光伏"
|
),
|
"entities_and_assets": re.compile(
|
r"公司|科技|集团|股份|银行|指数|ETF|期货|美债|原油|铜|铝|橡胶|"
|
r"稀土|黄金|白酒|创新药|英伟达|华为|海思"
|
),
|
"questions_and_answers": re.compile(
|
r"问答|答疑|还有什么问题|请问|问一下|怎么看|怎么判断|能不能|是否|"
|
r"什么逻辑|什么看法"
|
),
|
"uncertainty_and_verification": re.compile(
|
r"不确定|不知道|没研究|没怎么看|不熟悉|不强答|需核对|需要核对|"
|
r"可能|疑似|大概|应该|我记得|没记错|单位"
|
),
|
"risk_and_disclaimer": re.compile(
|
r"不构成投资建议|不推荐|投资建议|风险|账户负责|不要外传|仅代表个人|"
|
r"个人观点|风险承受能力"
|
),
|
}
|
|
|
def utc_now_iso() -> str:
|
return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
|
|
|
def sha256_file(path: Path) -> str:
|
digest = hashlib.sha256()
|
with path.open("rb") as stream:
|
for block in iter(lambda: stream.read(1024 * 1024), b""):
|
digest.update(block)
|
return digest.hexdigest()
|
|
|
def atomic_write_text(path: Path, text: str) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
handle, temp_name = tempfile.mkstemp(
|
prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent)
|
)
|
temp_path = Path(temp_name)
|
try:
|
with os.fdopen(handle, "w", encoding="utf-8", newline="\n") as stream:
|
stream.write(text)
|
stream.flush()
|
os.fsync(stream.fileno())
|
os.replace(temp_path, path)
|
except BaseException:
|
temp_path.unlink(missing_ok=True)
|
raise
|
|
|
def atomic_write_json(path: Path, payload: Any) -> None:
|
atomic_write_text(path, json.dumps(payload, ensure_ascii=False, indent=2) + "\n")
|
|
|
def require_file(path: Path, suffixes: Sequence[str] | None = None) -> Path:
|
resolved = path.expanduser().resolve()
|
if not resolved.is_file():
|
raise FileNotFoundError(f"required input file not found: {resolved}")
|
if suffixes and resolved.suffix.lower() not in {item.lower() for item in suffixes}:
|
raise ValueError(
|
f"unsupported input suffix {resolved.suffix!r}; expected one of {suffixes}"
|
)
|
return resolved
|
|
|
def timestamp_seconds(text: str) -> int | None:
|
match = TIMESTAMP_RE.search(text)
|
if not match:
|
return None
|
hours, minutes, seconds = (int(value) for value in match.groups())
|
return hours * 3600 + minutes * 60 + seconds
|
|
|
def timestamp_label(text: str) -> str | None:
|
match = TIMESTAMP_RE.search(text)
|
return match.group(0)[1:-1] if match else None
|
|
|
def content_without_prefix(text: str) -> str:
|
return SPEAKER_PREFIX_RE.sub("", text).strip()
|
|
|
def normalized_duplicate_key(text: str) -> str:
|
value = content_without_prefix(text)
|
value = re.sub(r"[\s\W_]+", "", value, flags=re.UNICODE)
|
return value.lower()
|
|
|
def bounded_append(bucket: list[dict[str, Any]], item: dict[str, Any], limit: int) -> None:
|
if len(bucket) < limit:
|
bucket.append(item)
|
|
|
def build_transcript_index(
|
transcript_path: Path, output_dir: Path, max_items: int = DEFAULT_MAX_ITEMS
|
) -> dict[str, Any]:
|
source = require_file(transcript_path, [".txt"])
|
if max_items <= 0:
|
raise ValueError("max_items must be positive")
|
output = output_dir.expanduser().resolve()
|
output.mkdir(parents=True, exist_ok=True)
|
|
started = time.perf_counter()
|
text = source.read_text(encoding="utf-8-sig")
|
lines = text.splitlines()
|
nonempty = [(index + 1, line.strip()) for index, line in enumerate(lines) if line.strip()]
|
|
categories: dict[str, list[dict[str, Any]]] = {
|
name: [] for name in CATEGORY_PATTERNS
|
}
|
categories["numbers_and_time_windows"] = []
|
category_total_counts = Counter()
|
duplicate_locations: dict[str, list[int]] = defaultdict(list)
|
duplicate_samples: dict[str, str] = {}
|
timestamp_regressions: list[dict[str, Any]] = []
|
previous_timestamp: tuple[int, int] | None = None
|
timestamps: list[int] = []
|
|
for line_number, raw_line in nonempty:
|
content = content_without_prefix(raw_line)
|
item = {
|
"line": line_number,
|
"timestamp": timestamp_label(raw_line),
|
"text": content[:1000],
|
}
|
for category, pattern in CATEGORY_PATTERNS.items():
|
if pattern.search(content):
|
category_total_counts[category] += 1
|
bounded_append(categories[category], item, max_items)
|
if NUMBER_RE.search(content):
|
category_total_counts["numbers_and_time_windows"] += 1
|
bounded_append(categories["numbers_and_time_windows"], item, max_items)
|
|
key = normalized_duplicate_key(raw_line)
|
if len(key) >= 20:
|
duplicate_locations[key].append(line_number)
|
duplicate_samples.setdefault(key, content[:300])
|
|
current_timestamp = timestamp_seconds(raw_line)
|
if current_timestamp is not None:
|
timestamps.append(current_timestamp)
|
if previous_timestamp and current_timestamp < previous_timestamp[1]:
|
timestamp_regressions.append(
|
{
|
"previous_line": previous_timestamp[0],
|
"previous_seconds": previous_timestamp[1],
|
"line": line_number,
|
"seconds": current_timestamp,
|
}
|
)
|
previous_timestamp = (line_number, current_timestamp)
|
|
duplicates = []
|
for key, locations in duplicate_locations.items():
|
if len(locations) > 1:
|
duplicates.append(
|
{
|
"count": len(locations),
|
"lines": locations[:30],
|
"sample": duplicate_samples[key],
|
}
|
)
|
duplicates.sort(key=lambda item: (-item["count"], item["lines"][0]))
|
|
result = {
|
"schema_version": "meeting-minutes-transcript-index.v1",
|
"tool_version": TOOL_VERSION,
|
"generated_at": utc_now_iso(),
|
"source": {
|
"path": str(source),
|
"bytes": source.stat().st_size,
|
"sha256": sha256_file(source),
|
"line_count": len(lines),
|
"nonempty_line_count": len(nonempty),
|
"character_count": len(text),
|
"first_timestamp_seconds": min(timestamps) if timestamps else None,
|
"last_timestamp_seconds": max(timestamps) if timestamps else None,
|
},
|
"sequence_check": {
|
"timestamp_count": len(timestamps),
|
"timestamp_regression_count": len(timestamp_regressions),
|
"timestamp_regressions": timestamp_regressions[:100],
|
"exact_normalized_duplicate_group_count": len(duplicates),
|
"duplicate_groups": duplicates[:100],
|
"requires_human_order_review": bool(timestamp_regressions or duplicates),
|
},
|
"category_total_counts": dict(category_total_counts),
|
"categories": categories,
|
"category_item_limit": max_items,
|
"elapsed_seconds": round(time.perf_counter() - started, 3),
|
}
|
atomic_write_json(output / "transcript-index.json", result)
|
atomic_write_text(output / "transcript-index.md", transcript_index_markdown(result))
|
return result
|
|
|
def transcript_index_markdown(index: dict[str, Any]) -> str:
|
source = index["source"]
|
sequence = index["sequence_check"]
|
lines = [
|
"# 逐字稿预检索引",
|
"",
|
f"- 工具版本：`{index['tool_version']}`",
|
f"- 来源：`{source['path']}`",
|
f"- 行数：{source['line_count']:,}；非空行：{source['nonempty_line_count']:,}",
|
f"- 字符数：{source['character_count']:,}；SHA-256：`{source['sha256']}`",
|
f"- 时间戳回退：{sequence['timestamp_regression_count']}；重复组：{sequence['exact_normalized_duplicate_group_count']}",
|
"- 用途：本文件只用于精读导航，不替代人工判断，也不作为正式会议结论。",
|
"",
|
"## 顺序与重复检查",
|
"",
|
]
|
if sequence["timestamp_regressions"]:
|
for item in sequence["timestamp_regressions"]:
|
lines.append(
|
f"- 时间戳回退：第 {item['previous_line']} 行 {item['previous_seconds']} 秒 -> "
|
f"第 {item['line']} 行 {item['seconds']} 秒"
|
)
|
else:
|
lines.append("- 未发现时间戳回退。")
|
if sequence["duplicate_groups"]:
|
lines.append("")
|
lines.append("### 重复候选")
|
lines.append("")
|
for item in sequence["duplicate_groups"][:30]:
|
lines.append(
|
f"- {item['count']} 次；行号 {', '.join(str(value) for value in item['lines'])}："
|
f"{item['sample']}"
|
)
|
else:
|
lines.append("- 未发现长度达到阈值的精确归一化重复。")
|
|
display_names = {
|
"sequence_and_sections": "章节与顺序候选",
|
"policy_and_macro": "政策与宏观",
|
"industry_and_technology": "产业与技术",
|
"entities_and_assets": "公司、指数与资产候选",
|
"questions_and_answers": "问答候选",
|
"uncertainty_and_verification": "不确定与待核对候选",
|
"risk_and_disclaimer": "风险与免责声明",
|
"numbers_and_time_windows": "数字与时间窗口",
|
}
|
for category, items in index["categories"].items():
|
total = index["category_total_counts"].get(category, 0)
|
lines.extend(["", f"## {display_names.get(category, category)}（{total} 条）", ""])
|
for item in items:
|
stamp = f" {item['timestamp']}" if item["timestamp"] else ""
|
lines.append(f"- L{item['line']}{stamp}：{item['text']}")
|
if total > len(items):
|
lines.append(f"- ……已达到索引上限，仅保留前 {len(items)} 条。")
|
lines.append("")
|
return "\n".join(lines)
|
|
|
def load_pdf_modules():
|
try:
|
from reportlab.lib import colors
|
from reportlab.lib.enums import TA_JUSTIFY, TA_LEFT
|
from reportlab.lib.pagesizes import A4
|
from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
|
from reportlab.lib.units import mm
|
from reportlab.pdfbase import pdfmetrics
|
from reportlab.pdfbase.ttfonts import TTFont
|
from reportlab.platypus import (
|
BaseDocTemplate,
|
Frame,
|
LongTable,
|
PageTemplate,
|
Paragraph,
|
Spacer,
|
Table,
|
TableStyle,
|
)
|
except ImportError as exc:
|
raise RuntimeError(
|
"PDF generation requires reportlab from the workspace runtime"
|
) from exc
|
return locals()
|
|
|
def find_chinese_fonts(
|
regular_override: Path | None = None, bold_override: Path | None = None
|
) -> tuple[Path, Path]:
|
if regular_override or bold_override:
|
if not regular_override or not bold_override:
|
raise ValueError("both regular and bold font paths must be provided")
|
return require_file(regular_override), require_file(bold_override)
|
|
windows_dir = Path(os.environ.get("WINDIR", r"C:\Windows"))
|
font_dir = windows_dir / "Fonts"
|
candidates = [
|
(font_dir / "msyh.ttc", font_dir / "msyhbd.ttc"),
|
(font_dir / "simhei.ttf", font_dir / "simhei.ttf"),
|
(font_dir / "simsun.ttc", font_dir / "simhei.ttf"),
|
]
|
for regular, bold in candidates:
|
if regular.is_file() and bold.is_file():
|
return regular.resolve(), bold.resolve()
|
raise FileNotFoundError(
|
"no supported Chinese font found; pass --font-regular and --font-bold"
|
)
|
|
|
def inline_markup(text: str) -> str:
|
placeholders: dict[str, str] = {}
|
|
def stash_code(match: re.Match[str]) -> str:
|
key = f"@@CODE{len(placeholders)}@@"
|
placeholders[key] = (
|
'<font name="MMRegular" color="#384B59" backColor="#EEF2F4">'
|
+ html.escape(match.group(1))
|
+ "</font>"
|
)
|
return key
|
|
value = re.sub(r"`([^`]+)`", stash_code, text)
|
value = html.escape(value)
|
value = re.sub(r"\*\*(.+?)\*\*", r"<b>\1</b>", value)
|
for key, replacement in placeholders.items():
|
value = value.replace(key, replacement)
|
return value
|
|
|
def split_table_row(line: str) -> list[str]:
|
value = line.strip().strip("|")
|
return [cell.strip() for cell in value.split("|")]
|
|
|
def is_table_separator(line: str) -> bool:
|
cells = split_table_row(line)
|
return bool(cells) and all(re.fullmatch(r":?-{3,}:?", cell) for cell in cells)
|
|
|
def table_column_widths(column_count: int, available: float) -> list[float]:
|
if column_count == 2:
|
return [available * 0.28, available * 0.72]
|
if column_count == 3:
|
return [available * 0.22, available * 0.36, available * 0.42]
|
if column_count == 4:
|
return [available * 0.16, available * 0.24, available * 0.30, available * 0.30]
|
return [available / column_count] * column_count
|
|
|
def markdown_title(markdown: str) -> str:
|
for line in markdown.splitlines():
|
if line.startswith("# "):
|
return line[2:].strip()
|
raise ValueError("Markdown must contain an H1 title")
|
|
|
def render_markdown_pdf(
|
markdown_path: Path,
|
output_pdf: Path,
|
regular_font: Path | None = None,
|
bold_font: Path | None = None,
|
) -> dict[str, Any]:
|
modules = load_pdf_modules()
|
colors = modules["colors"]
|
A4 = modules["A4"]
|
mm = modules["mm"]
|
pdfmetrics = modules["pdfmetrics"]
|
TTFont = modules["TTFont"]
|
ParagraphStyle = modules["ParagraphStyle"]
|
getSampleStyleSheet = modules["getSampleStyleSheet"]
|
TA_JUSTIFY = modules["TA_JUSTIFY"]
|
TA_LEFT = modules["TA_LEFT"]
|
BaseDocTemplate = modules["BaseDocTemplate"]
|
Frame = modules["Frame"]
|
PageTemplate = modules["PageTemplate"]
|
Paragraph = modules["Paragraph"]
|
Spacer = modules["Spacer"]
|
Table = modules["Table"]
|
LongTable = modules["LongTable"]
|
TableStyle = modules["TableStyle"]
|
|
source = require_file(markdown_path, [".md"])
|
destination = output_pdf.expanduser().resolve()
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
markdown = source.read_text(encoding="utf-8-sig")
|
title = markdown_title(markdown)
|
regular, bold = find_chinese_fonts(regular_font, bold_font)
|
pdfmetrics.registerFont(TTFont("MMRegular", str(regular)))
|
pdfmetrics.registerFont(TTFont("MMBold", str(bold)))
|
pdfmetrics.registerFontFamily(
|
"MMRegular",
|
normal="MMRegular",
|
bold="MMBold",
|
italic="MMRegular",
|
boldItalic="MMBold",
|
)
|
|
navy = colors.HexColor("#17324D")
|
blue = colors.HexColor("#245A7A")
|
accent = colors.HexColor("#C98B2E")
|
ink = colors.HexColor("#24313A")
|
muted = colors.HexColor("#667784")
|
pale_gold = colors.HexColor("#FBF5E8")
|
grid = colors.HexColor("#CAD5DC")
|
alt_row = colors.HexColor("#F7F9FA")
|
|
sample_styles = getSampleStyleSheet()
|
body = ParagraphStyle(
|
"BodyCN",
|
parent=sample_styles["BodyText"],
|
fontName="MMRegular",
|
fontSize=9.4,
|
leading=15.2,
|
textColor=ink,
|
alignment=TA_JUSTIFY,
|
spaceAfter=5.5,
|
wordWrap="CJK",
|
allowWidows=0,
|
allowOrphans=0,
|
)
|
styles = {
|
"title": ParagraphStyle(
|
"TitleCN",
|
parent=body,
|
fontName="MMBold",
|
fontSize=22,
|
leading=31,
|
textColor=navy,
|
alignment=TA_LEFT,
|
spaceBefore=4,
|
spaceAfter=12,
|
keepWithNext=True,
|
),
|
"h2": ParagraphStyle(
|
"H2CN",
|
parent=body,
|
fontName="MMBold",
|
fontSize=14.5,
|
leading=21,
|
textColor=navy,
|
spaceBefore=11,
|
spaceAfter=7,
|
keepWithNext=True,
|
),
|
"h3": ParagraphStyle(
|
"H3CN",
|
parent=body,
|
fontName="MMBold",
|
fontSize=11.2,
|
leading=17,
|
textColor=blue,
|
spaceBefore=8,
|
spaceAfter=4,
|
keepWithNext=True,
|
),
|
"body": body,
|
"bullet": ParagraphStyle(
|
"BulletCN", parent=body, leftIndent=14, firstLineIndent=-12, spaceAfter=3.5
|
),
|
"numbered": ParagraphStyle(
|
"NumberCN", parent=body, leftIndent=18, firstLineIndent=-16, spaceAfter=4
|
),
|
"quote": ParagraphStyle(
|
"QuoteCN",
|
parent=body,
|
fontSize=9.2,
|
leading=14.8,
|
textColor=colors.HexColor("#394B58"),
|
alignment=TA_LEFT,
|
spaceAfter=0,
|
),
|
"table_header": ParagraphStyle(
|
"TableHeaderCN",
|
parent=body,
|
fontName="MMBold",
|
fontSize=8.1,
|
leading=11.3,
|
textColor=colors.white,
|
alignment=TA_LEFT,
|
spaceAfter=0,
|
),
|
"table_cell": ParagraphStyle(
|
"TableCellCN",
|
parent=body,
|
fontSize=7.75,
|
leading=11.2,
|
alignment=TA_LEFT,
|
spaceAfter=0,
|
),
|
}
|
|
class MinutesDocument(BaseDocTemplate):
|
def __init__(self, filename: str, **kwargs: Any):
|
super().__init__(filename, **kwargs)
|
frame = Frame(
|
self.leftMargin,
|
self.bottomMargin,
|
self.width,
|
self.height,
|
leftPadding=0,
|
rightPadding=0,
|
topPadding=0,
|
bottomPadding=0,
|
)
|
self.addPageTemplates(
|
[PageTemplate(id="minutes", frames=[frame], onPage=self.decorate_page)]
|
)
|
|
def decorate_page(self, canvas: Any, doc: Any) -> None:
|
page_width, page_height = A4
|
canvas.saveState()
|
canvas.setStrokeColor(colors.HexColor("#D7E0E5"))
|
canvas.setLineWidth(0.45)
|
canvas.line(
|
doc.leftMargin,
|
page_height - 14.5 * mm,
|
page_width - doc.rightMargin,
|
page_height - 14.5 * mm,
|
)
|
canvas.setFont("MMRegular", 7.3)
|
canvas.setFillColor(muted)
|
canvas.drawString(doc.leftMargin, page_height - 11.2 * mm, "MB-X 内部资料｜会议纪要")
|
short_title = title if len(title) <= 28 else title[:27] + "…"
|
canvas.drawRightString(
|
page_width - doc.rightMargin, page_height - 11.2 * mm, short_title
|
)
|
canvas.line(
|
doc.leftMargin,
|
13.5 * mm,
|
page_width - doc.rightMargin,
|
13.5 * mm,
|
)
|
canvas.setFont("MMRegular", 7.4)
|
canvas.drawString(doc.leftMargin, 9.2 * mm, DISCLAIMER)
|
canvas.drawRightString(
|
page_width - doc.rightMargin, 9.2 * mm, f"第 {doc.page} 页"
|
)
|
canvas.restoreState()
|
|
temporary_pdf = destination.with_name(f".{destination.stem}.building.pdf")
|
temporary_pdf.unlink(missing_ok=True)
|
document = MinutesDocument(
|
str(temporary_pdf),
|
pagesize=A4,
|
leftMargin=20 * mm,
|
rightMargin=20 * mm,
|
topMargin=20 * mm,
|
bottomMargin=19 * mm,
|
title=title,
|
author="Codex",
|
subject="会议纪要",
|
)
|
|
def quote_box(text: str) -> Any:
|
box = Table(
|
[[Paragraph(inline_markup(text), styles["quote"])]],
|
colWidths=[document.width],
|
hAlign="LEFT",
|
)
|
box.setStyle(
|
TableStyle(
|
[
|
("BACKGROUND", (0, 0), (-1, -1), pale_gold),
|
("BOX", (0, 0), (-1, -1), 0.7, colors.HexColor("#E2C78E")),
|
("LINEBEFORE", (0, 0), (0, -1), 3.2, accent),
|
("LEFTPADDING", (0, 0), (-1, -1), 10),
|
("RIGHTPADDING", (0, 0), (-1, -1), 10),
|
("TOPPADDING", (0, 0), (-1, -1), 8),
|
("BOTTOMPADDING", (0, 0), (-1, -1), 8),
|
]
|
)
|
)
|
return box
|
|
def table_flowable(table_lines: list[str]) -> Any:
|
raw_rows = [split_table_row(line) for line in table_lines if not is_table_separator(line)]
|
column_count = max(len(row) for row in raw_rows)
|
raw_rows = [row + [""] * (column_count - len(row)) for row in raw_rows]
|
data = []
|
for row_number, row in enumerate(raw_rows):
|
row_style = styles["table_header"] if row_number == 0 else styles["table_cell"]
|
data.append([Paragraph(inline_markup(cell), row_style) for cell in row])
|
table = LongTable(
|
data,
|
colWidths=table_column_widths(column_count, document.width),
|
repeatRows=1,
|
hAlign="LEFT",
|
splitByRow=1,
|
)
|
vertical_padding = 3.0 if column_count == 2 and len(data) > 12 else 4.8
|
commands = [
|
("BACKGROUND", (0, 0), (-1, 0), navy),
|
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
|
("GRID", (0, 0), (-1, -1), 0.45, grid),
|
("VALIGN", (0, 0), (-1, -1), "TOP"),
|
("LEFTPADDING", (0, 0), (-1, -1), 5),
|
("RIGHTPADDING", (0, 0), (-1, -1), 5),
|
("TOPPADDING", (0, 0), (-1, -1), vertical_padding),
|
("BOTTOMPADDING", (0, 0), (-1, -1), vertical_padding),
|
]
|
for row_number in range(1, len(data)):
|
if row_number % 2 == 0:
|
commands.append(("BACKGROUND", (0, row_number), (-1, row_number), alt_row))
|
table.setStyle(TableStyle(commands))
|
return table
|
|
source_lines = markdown.splitlines()
|
story: list[Any] = []
|
index = 0
|
while index < len(source_lines):
|
stripped = source_lines[index].strip()
|
if not stripped:
|
index += 1
|
continue
|
if stripped.startswith("# "):
|
story.extend(
|
[
|
Spacer(1, 5 * mm),
|
Paragraph(inline_markup(stripped[2:]), styles["title"]),
|
Table(
|
[[""]],
|
colWidths=[42 * mm],
|
rowHeights=[1.2 * mm],
|
style=TableStyle(
|
[("BACKGROUND", (0, 0), (-1, -1), accent)]
|
),
|
hAlign="LEFT",
|
),
|
Spacer(1, 5 * mm),
|
]
|
)
|
index += 1
|
continue
|
if stripped.startswith("## ") or stripped.startswith("### "):
|
level = 2 if stripped.startswith("## ") and not stripped.startswith("### ") else 3
|
marker_length = 3 if level == 2 else 4
|
heading = Paragraph(
|
inline_markup(stripped[marker_length:]), styles[f"h{level}"]
|
)
|
probe = index + 1
|
while probe < len(source_lines) and not source_lines[probe].strip():
|
probe += 1
|
if (
|
probe + 1 < len(source_lines)
|
and "|" in source_lines[probe]
|
and is_table_separator(source_lines[probe + 1])
|
):
|
heading.keepWithNext = 0
|
story.append(heading)
|
index += 1
|
continue
|
if stripped.startswith(">"):
|
quote_lines = []
|
while index < len(source_lines) and source_lines[index].strip().startswith(">"):
|
quote_lines.append(source_lines[index].strip()[1:].strip())
|
index += 1
|
story.extend([quote_box(" ".join(quote_lines)), Spacer(1, 3 * mm)])
|
continue
|
if (
|
"|" in stripped
|
and index + 1 < len(source_lines)
|
and is_table_separator(source_lines[index + 1])
|
):
|
table_lines = [source_lines[index], source_lines[index + 1]]
|
index += 2
|
while index < len(source_lines):
|
candidate = source_lines[index].strip()
|
if not candidate or "|" not in candidate:
|
break
|
table_lines.append(source_lines[index])
|
index += 1
|
story.extend([table_flowable(table_lines), Spacer(1, 3.4 * mm)])
|
continue
|
bullet = re.match(r"^[-*]\s+(.+)$", stripped)
|
if bullet:
|
story.append(
|
Paragraph(f"• {inline_markup(bullet.group(1))}", styles["bullet"])
|
)
|
index += 1
|
continue
|
numbered = re.match(r"^(\d+)\.\s+(.+)$", stripped)
|
if numbered:
|
story.append(
|
Paragraph(
|
f"<b>{numbered.group(1)}.</b> {inline_markup(numbered.group(2))}",
|
styles["numbered"],
|
)
|
)
|
index += 1
|
continue
|
story.append(Paragraph(inline_markup(stripped), styles["body"]))
|
index += 1
|
|
started = time.perf_counter()
|
try:
|
document.build(story)
|
os.replace(temporary_pdf, destination)
|
except BaseException:
|
temporary_pdf.unlink(missing_ok=True)
|
raise
|
return {
|
"title": title,
|
"markdown": str(source),
|
"pdf": str(destination),
|
"pdf_bytes": destination.stat().st_size,
|
"markdown_sha256": sha256_file(source),
|
"pdf_sha256": sha256_file(destination),
|
"font_regular": str(regular),
|
"font_bold": str(bold),
|
"elapsed_seconds": round(time.perf_counter() - started, 3),
|
}
|
|
|
def resolve_pdftoppm(explicit: Path | None = None) -> Path:
|
if explicit:
|
return require_file(explicit)
|
env_path = os.environ.get("PDFTOPPM")
|
if env_path:
|
return require_file(Path(env_path))
|
executable = Path(sys.executable).resolve()
|
for parent in executable.parents:
|
candidate = parent / "native" / "poppler" / "Library" / "bin" / "pdftoppm.exe"
|
if candidate.is_file():
|
return candidate.resolve()
|
from_path = shutil.which("pdftoppm")
|
if from_path:
|
return Path(from_path).resolve()
|
raise FileNotFoundError(
|
"pdftoppm not found; pass --pdftoppm or set PDFTOPPM"
|
)
|
|
|
def render_pdf_pages(
|
pdf_path: Path, render_dir: Path, pdftoppm: Path, dpi: int
|
) -> list[Path]:
|
if dpi < 72:
|
raise ValueError("render DPI must be at least 72")
|
render_dir.mkdir(parents=True, exist_ok=True)
|
for candidate in render_dir.glob("page-*.png"):
|
candidate.unlink()
|
command = [
|
str(pdftoppm),
|
"-png",
|
"-r",
|
str(dpi),
|
str(pdf_path),
|
str(render_dir / "page"),
|
]
|
completed = subprocess.run(command, capture_output=True, text=True, encoding="utf-8")
|
if completed.returncode != 0:
|
raise RuntimeError(
|
f"pdftoppm failed with exit {completed.returncode}: {completed.stderr.strip()}"
|
)
|
pages = sorted(render_dir.glob("page-*.png"))
|
if not pages:
|
raise RuntimeError("pdftoppm returned success but produced no page images")
|
return pages
|
|
|
def create_contact_sheets(page_images: Sequence[Path], render_dir: Path) -> list[Path]:
|
try:
|
from PIL import Image, ImageDraw
|
except ImportError as exc:
|
raise RuntimeError("contact sheets require Pillow") from exc
|
for candidate in render_dir.glob("contact-*.png"):
|
candidate.unlink()
|
outputs: list[Path] = []
|
thumb_width = 560
|
for group_index in range(0, len(page_images), 4):
|
batch = page_images[group_index : group_index + 4]
|
thumbnails = []
|
for image_path in batch:
|
with Image.open(image_path) as source:
|
image = source.convert("RGB")
|
height = round(image.height * thumb_width / image.width)
|
image = image.resize((thumb_width, height), Image.Resampling.LANCZOS)
|
canvas = Image.new("RGB", (thumb_width, height + 36), "white")
|
canvas.paste(image, (0, 36))
|
ImageDraw.Draw(canvas).text((12, 10), image_path.stem, fill="black")
|
thumbnails.append(canvas)
|
cell_height = max(image.height for image in thumbnails)
|
sheet = Image.new(
|
"RGB", (thumb_width * 2 + 30, cell_height * 2 + 30), (225, 230, 233)
|
)
|
for cell_index, thumbnail in enumerate(thumbnails):
|
x = (cell_index % 2) * (thumb_width + 10) + 5
|
y = (cell_index // 2) * (cell_height + 10) + 5
|
sheet.paste(thumbnail, (x, y))
|
output = render_dir / f"contact-{group_index // 4 + 1:02d}.png"
|
sheet.save(output, optimize=True)
|
outputs.append(output)
|
return outputs
|
|
|
def markdown_headings(markdown: str) -> list[str]:
|
headings = []
|
for line in markdown.splitlines():
|
match = re.match(r"^#{1,3}\s+(.+)$", line.strip())
|
if match:
|
headings.append(re.sub(r"\*\*|`", "", match.group(1)).strip())
|
return headings
|
|
|
def markdown_plain_text(markdown: str) -> str:
|
value = re.sub(r"```.*?```", "", markdown, flags=re.DOTALL)
|
value = re.sub(r"!\[[^]]*]\([^)]*\)", "", value)
|
value = re.sub(r"\[([^]]+)]\([^)]*\)", r"\1", value)
|
value = re.sub(r"[#>*_`|:-]", "", value)
|
return re.sub(r"\s+", "", value)
|
|
|
def normalized_text_key(value: str) -> str:
|
return re.sub(r"\s+", "", value)
|
|
|
def run_pdf_qa(
|
markdown_path: Path,
|
pdf_path: Path,
|
qa_dir: Path,
|
pdftoppm_override: Path | None = None,
|
dpi: int = DEFAULT_RENDER_DPI,
|
) -> dict[str, Any]:
|
try:
|
from PIL import Image
|
from pypdf import PdfReader
|
except ImportError as exc:
|
raise RuntimeError("PDF QA requires pypdf and Pillow") from exc
|
|
markdown_source = require_file(markdown_path, [".md"])
|
pdf_source = require_file(pdf_path, [".pdf"])
|
output = qa_dir.expanduser().resolve()
|
output.mkdir(parents=True, exist_ok=True)
|
markdown = markdown_source.read_text(encoding="utf-8-sig")
|
headings = markdown_headings(markdown)
|
|
started = time.perf_counter()
|
reader = PdfReader(str(pdf_source))
|
page_texts = [page.extract_text() or "" for page in reader.pages]
|
full_text = "\n".join(page_texts)
|
normalized_full_text = normalized_text_key(full_text)
|
normalized_markdown = markdown_plain_text(markdown)
|
normalized_pdf = normalized_full_text
|
missing_headings = [
|
heading
|
for heading in headings
|
if normalized_text_key(heading) not in normalized_full_text
|
]
|
empty_pages = [index + 1 for index, text in enumerate(page_texts) if len(text.strip()) < 50]
|
|
pdftoppm = resolve_pdftoppm(pdftoppm_override)
|
pages = render_pdf_pages(pdf_source, output, pdftoppm, dpi)
|
dimensions = []
|
for image_path in pages:
|
with Image.open(image_path) as image:
|
dimensions.append(image.size)
|
contact_sheets = create_contact_sheets(pages, output)
|
|
checks = {
|
"has_pages": len(page_texts) > 0,
|
"render_count_matches_page_count": len(pages) == len(page_texts),
|
"no_empty_text_pages": not empty_pages,
|
"no_replacement_character": "\ufffd" not in full_text,
|
"all_markdown_headings_present": not missing_headings,
|
"same_render_dimensions": len(set(dimensions)) == 1,
|
"disclaimer_present_when_required": (
|
DISCLAIMER not in markdown or DISCLAIMER in full_text
|
),
|
"last_heading_reaches_final_page": (
|
not headings
|
or normalized_text_key(headings[-1])
|
in normalized_text_key(page_texts[-1])
|
),
|
"text_length_ratio_reasonable": (
|
len(normalized_markdown) == 0
|
or len(normalized_pdf) / len(normalized_markdown) >= 0.90
|
),
|
}
|
automated_pass = all(checks.values())
|
report = {
|
"schema_version": "meeting-minutes-pdf-qa.v1",
|
"tool_version": TOOL_VERSION,
|
"generated_at": utc_now_iso(),
|
"markdown": {
|
"path": str(markdown_source),
|
"bytes": markdown_source.stat().st_size,
|
"sha256": sha256_file(markdown_source),
|
"plain_text_chars": len(normalized_markdown),
|
"heading_count": len(headings),
|
},
|
"pdf": {
|
"path": str(pdf_source),
|
"bytes": pdf_source.stat().st_size,
|
"sha256": sha256_file(pdf_source),
|
"page_count": len(page_texts),
|
"extracted_text_chars": len(full_text),
|
"normalized_text_ratio": (
|
round(len(normalized_pdf) / len(normalized_markdown), 3)
|
if normalized_markdown
|
else None
|
),
|
},
|
"render": {
|
"pdftoppm": str(pdftoppm),
|
"dpi": dpi,
|
"page_images": [str(path) for path in pages],
|
"contact_sheets": [str(path) for path in contact_sheets],
|
"dimensions": [list(value) for value in sorted(set(dimensions))],
|
},
|
"checks": checks,
|
"details": {
|
"empty_pages": empty_pages,
|
"missing_headings": missing_headings,
|
"page_text_char_counts": [len(text) for text in page_texts],
|
},
|
"automated_checks_passed": automated_pass,
|
"manual_visual_review_required": True,
|
"manual_visual_review_scope": [
|
"首页标题和免责声明",
|
"章节切换和孤立标题",
|
"长表格跨页与重复表头",
|
"问答章节",
|
"末页跟踪清单或待核对项",
|
"乱码、裁切、重叠、黑块和异常空白",
|
],
|
"elapsed_seconds": round(time.perf_counter() - started, 3),
|
}
|
atomic_write_json(output / "qa-report.json", report)
|
atomic_write_text(output / "qa-report.md", qa_report_markdown(report))
|
return report
|
|
|
def qa_report_markdown(report: dict[str, Any]) -> str:
|
status = "PASS" if report["automated_checks_passed"] else "FAIL"
|
lines = [
|
"# 会议纪要 PDF 自动验收报告",
|
"",
|
f"- 自动检查：**{status}**",
|
f"- Markdown：`{report['markdown']['path']}`",
|
f"- PDF：`{report['pdf']['path']}`",
|
f"- 页数：{report['pdf']['page_count']}；文字比例：{report['pdf']['normalized_text_ratio']}",
|
f"- 渲染：{len(report['render']['page_images'])} 页；联系表：{len(report['render']['contact_sheets'])} 张",
|
"- 注意：自动检查通过不等于视觉验收完成，仍需逐页或联系表人工检查。",
|
"",
|
"## 自动检查",
|
"",
|
]
|
for name, passed in report["checks"].items():
|
lines.append(f"- {'PASS' if passed else 'FAIL'}：`{name}`")
|
lines.extend(["", "## 人工视觉检查范围", ""])
|
for item in report["manual_visual_review_scope"]:
|
lines.append(f"- {item}")
|
lines.extend(["", "## 联系表", ""])
|
for path in report["render"]["contact_sheets"]:
|
lines.append(f"- `{path}`")
|
lines.append("")
|
return "\n".join(lines)
|
|
|
def add_common_pdf_arguments(parser: argparse.ArgumentParser) -> None:
|
parser.add_argument("--font-regular", type=Path)
|
parser.add_argument("--font-bold", type=Path)
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
parser = argparse.ArgumentParser(
|
description="Prepare, render and verify MB-X meeting minutes"
|
)
|
parser.add_argument("--version", action="version", version=TOOL_VERSION)
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
index_parser = subparsers.add_parser("index", help="build transcript navigation index")
|
index_parser.add_argument("transcript", type=Path)
|
index_parser.add_argument("--output-dir", type=Path, required=True)
|
index_parser.add_argument("--max-items", type=int, default=DEFAULT_MAX_ITEMS)
|
|
pdf_parser = subparsers.add_parser("pdf", help="render final Markdown to PDF")
|
pdf_parser.add_argument("markdown", type=Path)
|
pdf_parser.add_argument("--output-pdf", type=Path, required=True)
|
add_common_pdf_arguments(pdf_parser)
|
|
qa_parser = subparsers.add_parser("qa", help="run PDF extraction and render QA")
|
qa_parser.add_argument("markdown", type=Path)
|
qa_parser.add_argument("pdf", type=Path)
|
qa_parser.add_argument("--qa-dir", type=Path, required=True)
|
qa_parser.add_argument("--pdftoppm", type=Path)
|
qa_parser.add_argument("--dpi", type=int, default=DEFAULT_RENDER_DPI)
|
|
deliver_parser = subparsers.add_parser(
|
"deliver", help="render PDF and run automated QA in one command"
|
)
|
deliver_parser.add_argument("markdown", type=Path)
|
deliver_parser.add_argument("--output-pdf", type=Path, required=True)
|
deliver_parser.add_argument("--qa-dir", type=Path, required=True)
|
deliver_parser.add_argument("--pdftoppm", type=Path)
|
deliver_parser.add_argument("--dpi", type=int, default=DEFAULT_RENDER_DPI)
|
add_common_pdf_arguments(deliver_parser)
|
return parser
|
|
|
def emit_summary(payload: dict[str, Any]) -> None:
|
print(json.dumps(payload, ensure_ascii=False, separators=(",", ":")))
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
parser = build_parser()
|
args = parser.parse_args(argv)
|
try:
|
if args.command == "index":
|
result = build_transcript_index(args.transcript, args.output_dir, args.max_items)
|
emit_summary(
|
{
|
"status": "PASS",
|
"command": "index",
|
"line_count": result["source"]["line_count"],
|
"timestamp_regressions": result["sequence_check"]["timestamp_regression_count"],
|
"duplicate_groups": result["sequence_check"]["exact_normalized_duplicate_group_count"],
|
"output_dir": str(args.output_dir.expanduser().resolve()),
|
"elapsed_seconds": result["elapsed_seconds"],
|
}
|
)
|
return 0
|
if args.command == "pdf":
|
result = render_markdown_pdf(
|
args.markdown, args.output_pdf, args.font_regular, args.font_bold
|
)
|
emit_summary({"status": "PASS", "command": "pdf", **result})
|
return 0
|
if args.command == "qa":
|
result = run_pdf_qa(
|
args.markdown, args.pdf, args.qa_dir, args.pdftoppm, args.dpi
|
)
|
emit_summary(
|
{
|
"status": "PASS" if result["automated_checks_passed"] else "FAIL",
|
"command": "qa",
|
"page_count": result["pdf"]["page_count"],
|
"contact_sheets": len(result["render"]["contact_sheets"]),
|
"qa_dir": str(args.qa_dir.expanduser().resolve()),
|
"manual_visual_review_required": True,
|
"elapsed_seconds": result["elapsed_seconds"],
|
}
|
)
|
return 0 if result["automated_checks_passed"] else 2
|
if args.command == "deliver":
|
pdf_result = render_markdown_pdf(
|
args.markdown, args.output_pdf, args.font_regular, args.font_bold
|
)
|
qa_result = run_pdf_qa(
|
args.markdown,
|
args.output_pdf,
|
args.qa_dir,
|
args.pdftoppm,
|
args.dpi,
|
)
|
emit_summary(
|
{
|
"status": "PASS" if qa_result["automated_checks_passed"] else "FAIL",
|
"command": "deliver",
|
"pdf": pdf_result["pdf"],
|
"page_count": qa_result["pdf"]["page_count"],
|
"contact_sheets": len(qa_result["render"]["contact_sheets"]),
|
"qa_dir": str(args.qa_dir.expanduser().resolve()),
|
"manual_visual_review_required": True,
|
"elapsed_seconds": round(
|
pdf_result["elapsed_seconds"] + qa_result["elapsed_seconds"], 3
|
),
|
}
|
)
|
return 0 if qa_result["automated_checks_passed"] else 2
|
parser.error(f"unknown command: {args.command}")
|
except Exception as exc:
|
emit_summary(
|
{
|
"status": "ERROR",
|
"command": args.command,
|
"error_type": type(exc).__name__,
|
"error": str(exc),
|
}
|
)
|
return 1
|
return 1
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
