From b2d271c5edbffaeead1814ac626bebcf8adae795 Mon Sep 17 00:00:00 2001 From: wandering Date: Mon, 19 Oct 2026 20:07:38 +0800 Subject: [PATCH] =?UTF-8?q?feat(mcp):=20=E7=94=B3=E8=B4=AD=E5=8D=95/?= =?UTF-8?q?=E5=87=BA=E5=BA=93=E5=8D=95=E7=94=9F=E6=88=90=E4=B8=8E=E6=8A=BD?= =?UTF-8?q?=E5=8F=96=E7=BB=93=E6=9E=9C=E5=B1=80=E9=83=A8=E4=BF=AE=E6=AD=A3?= =?UTF-8?q?=EF=BC=882a=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - src/infra/com_thread.py：Office COM 调用的单线程串行 actor。Word/Excel 并发驱动会互相抢 ActiveDocument 与模态框，因此所有 COM 操作排队在同一线程执行（与 P3 的 Playwright actor 分开， COM 套间与 greenlet 约束不同） - mcp__arb__doc_requisition：Excel 申购单，产物写 cases//artifacts/，支持 exclude 关键词与 project 覆盖。实机验证：2 张普通发票 + 1 笔支付 → 排除「万用表」后生成 19456 字节 .xls，5.0s - mcp__arb__doc_consumable：Word 出库单；纯差旅返回 skipped 而非报错。当前被缺模板挡住（见下） - mcp__arb__case_patch_extraction：只改传入的键，回显每个字段的 from/to，未知键告警但仍写入， 与现值相同时不写（避免无意义重抽），金额/日期仍走 normalize - server.py：工具数 15 → 18，playbook 增加第 6 步（文档产物）与核对修正说明 发现（未处理，需产品决定）：Word 模板「易耗品、出库单.doc」在 4fbb4c3(2026-08-31，同一提交新增 requisition.py)被删除，至今不在仓库里。因此 consumable.py 的出库单路径自那以后一直不可用； Web 端 _try_fill_consumable_doc 捕获异常后只记日志，所以普通发票提交一直在静默跳过出库单生成。 已确认历史版本是空白校用表单（无姓名、无卡号、无长数字串），是否恢复或改为配置模板路径待定。 门禁：ruff / format --check / mypy / deptry 全绿，pytest 132 passed。 --- src/infra/com_thread.py | 44 +++++++++++++ src/mcp/server.py | 10 ++- src/mcp/tools/documents.py | 125 ++++++++++++++++++++++++++++++++++++ src/mcp/tools/extraction.py | 42 ++++++++++++ 4 files changed, 219 insertions(+), 2 deletions(-) create mode 100644 src/infra/com_thread.py create mode 100644 src/mcp/tools/documents.py diff --git a/src/infra/com_thread.py b/src/infra/com_thread.py new file mode 100644 index 0000000..d508fbd --- /dev/null +++ b/src/infra/com_thread.py @@ -0,0 +1,44 @@ +"""Office COM 调用的单线程串行化 + +Word/Excel 自动化只能在同一线程里稳定工作，而且两个 MCP 工具并发驱动 +同一个 Office 应用会互相干扰（模态框、剪贴板、ActiveDocument 抢占）。 +因此所有 COM 调用统一提交到一个专属工作线程串行执行。 + +与 Playwright 的关系：P3 的浏览器用同样的模式，但那是另一个 actor， +不要复用同一个线程——COM 套间与 greenlet 的约束不同。 +""" + +from __future__ import annotations + +import threading +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor +from typing import Any + +_executor: ThreadPoolExecutor | None = None +_lock = threading.Lock() + + +def _pool() -> ThreadPoolExecutor: + global _executor + if _executor is None: + _executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="com") + return _executor + + +def run_com[T](fn: Callable[..., T], *args: Any, **kwargs: Any) -> T: + """在专用 COM 线程上串行执行 fn，并等待结果。 + + 异常原样抛出（包装在 Future.result 里会保留 traceback）。 + """ + with _lock: # 同一时刻只允许一个 Office 操作 + future = _pool().submit(fn, *args, **kwargs) + return future.result() + + +def shutdown() -> None: + """释放线程（进程退出或测试收尾用）。""" + global _executor + if _executor is not None: + _executor.shutdown(wait=False) + _executor = None diff --git a/src/mcp/server.py b/src/mcp/server.py index 7e6736e..ecd44be 100644 --- a/src/mcp/server.py +++ b/src/mcp/server.py @@ -13,7 +13,7 @@ from typing import Any from mcp.server import MCPServer -from .tools import analysis, extraction, materials +from .tools import analysis, documents, extraction, materials SERVER_NAME = "auto-reimbursement" @@ -33,7 +33,10 @@ INSTRUCTIONS = """财务报销自动化工具集（发票抽取交接 → 匹配 或用 mcp__arb__validate_info(draft=...) 不落盘反复试。不完整就向用户要补充材料，禁止自己宣布可提交。 拼 JSON 前先 mcp__arb__info_skeleton(info_type) 取骨架：required 为 true 的键必须存在（含值为空串的备注类 字段），骨架与 validation_rules.json 同源，别凭记忆造结构。 -6. 每步之后用 mcp__arb__case_status 复查态势；阶段由产物推导，没有内部状态机。 +6. 普通发票用 mcp__arb__doc_consumable 生成 Word 出库单、mcp__arb__doc_requisition 生成 Excel 申购单 + （COM 自动化，需本机装有 Word/Excel；纯差旅会返回 skipped，不是错误）。 + 人工核对后要改单张票据的某个字段，用 mcp__arb__case_patch_extraction（只传要改的键，会回显 from/to）。 +7. 每步之后用 mcp__arb__case_status 复查态势；阶段由产物推导，没有内部状态机。 mcp__arb__case_reset(scope) 可清派生产物重跑；mcp__arb__case_list 看全部会话。 尚未实现（**现在不要调用，也不要在回答里承诺**）：Word 出库单与 Excel 申购单、浏览器填报、 @@ -53,11 +56,14 @@ TOOLS: list[Callable[..., dict[str, Any]]] = [ extraction.case_read_text, extraction.case_save_extraction, extraction.extraction_spec, + extraction.case_patch_extraction, analysis.match_payments, analysis.save_info, analysis.validate_info, analysis.rules_get, analysis.info_skeleton, + documents.doc_consumable, + documents.doc_requisition, ] diff --git a/src/mcp/tools/documents.py b/src/mcp/tools/documents.py new file mode 100644 index 0000000..f69b4ed --- /dev/null +++ b/src/mcp/tools/documents.py @@ -0,0 +1,125 @@ +"""Word 出库单与 Excel 申购单生成工具 + +都是 COM 自动化，依赖 Windows + Microsoft Office；统一走 src.infra.com_thread +的单线程 actor，避免两个工具调用同时驱动 Word/Excel。 +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from ...cases import ARTIFACTS_DIR_NAME, CACHE_DIR_NAME, PROJECT_ROOT, CaseStore +from ...config import SESSION_CONFIG_KEYS, load_config +from ...infra.com_thread import run_com +from ...infra.documents import ( + CONSUMABLE_DOC_FILENAME, + REQUISITION_FILENAME, + fill_consumable_from_template, + generate_requisition, + load_requisition_items, +) +from ..envelope import fail, ok + +_store = CaseStore() + + +def _case_config(d: Path) -> dict[str, Any]: + """项目配置 + 会话目录里 config.json 的用户字段覆盖。""" + config = load_config() + cfg_path = d / "config.json" + if cfg_path.exists(): + import json + + try: + raw: dict[str, Any] = json.loads(cfg_path.read_text(encoding="utf-8")) + except Exception: + raw = {} + for key in SESSION_CONFIG_KEYS: + if key in raw: + config[key] = raw[key] + return config + + +def doc_consumable(case_id: str, output_name: str | None = None) -> dict[str, Any]: + """生成易耗品出库单（Word，仅普通发票需要）。 + + 取会话里的 payment_records.csv 填入项目根目录模板的副本，输出到 + cases//artifacts/。纯差旅发票会返回 skipped，不是错误。 + 需要 Windows + 已安装 Microsoft Word。 + """ + d = _store.resolve(case_id) + if d is None: + return fail(f"会话不存在: {case_id}") + csv_path = d / "payment_records.csv" + if not csv_path.exists(): + return fail("缺少 payment_records.csv", next_action="先 mcp__arb__match_payments 生成汇总") + groups = _groups(d) + if groups is not None and not groups.get("general_count", 0): + return ok(skipped=True, reason="本会话没有普通发票（纯差旅），不需要易耗品出库单") + template = PROJECT_ROOT / CONSUMABLE_DOC_FILENAME + if not template.exists(): + return fail(f"出库单模板不存在: {template}", next_action="把模板放回项目根目录") + out = d / ARTIFACTS_DIR_NAME / (output_name or CONSUMABLE_DOC_FILENAME) + out.parent.mkdir(parents=True, exist_ok=True) + config = _case_config(d) + try: + run_com(fill_consumable_from_template, csv_path, template, out, config=config) + except ImportError: + return fail("缺少 pywin32，无法驱动 Word", next_action="uv sync 安装项目依赖", needs="pywin32") + except Exception as e: + return fail(f"生成出库单失败: {e}", next_action="检查 Word 是否被其他实例占用或有弹窗未关闭", doc=str(out)) + return ok(doc=str(out), filename=out.name, bytes=out.stat().st_size, skipped=False) + + +def doc_requisition(case_id: str, exclude: list[str] | None = None, project: str | None = None) -> dict[str, Any]: + """生成低值易耗品申购单（Excel，仅普通发票）。 + + exclude 传物品名关键词可剔除不参与申购的发票；project 覆盖「对应实验实训项目」， + 默认取 normal_info 的报销事由。需要 Windows + 已安装 Microsoft Excel。 + """ + d = _store.resolve(case_id) + if d is None: + return fail(f"会话不存在: {case_id}") + keywords = {k.strip() for k in (exclude or []) if k.strip()} + items = load_requisition_items(d / CACHE_DIR_NAME, exclude_keywords=keywords or None) + if not items: + return fail("没有可申购的普通发票", next_action="确认已抽取普通发票，或本会话是纯差旅无需申购单") + ts_template = _find_template() + if ts_template is None: + return fail(f"未找到申购单模板 {REQUISITION_FILENAME}*.xls", next_action="把模板放到项目根目录") + out = d / ARTIFACTS_DIR_NAME / f"低值易耗品申购_{d.name}.xls" + out.parent.mkdir(parents=True, exist_ok=True) + try: + produced = run_com( + generate_requisition, + d, + ts_template, + out, + project, + keywords or None, + ) + except ImportError: + return fail("缺少 pywin32，无法驱动 Excel", next_action="uv sync 安装项目依赖", needs="pywin32") + except Exception as e: + return fail(f"生成申购单失败: {e}", next_action="检查 Excel 是否被其他实例占用或有弹窗未关闭", doc=str(out)) + return ok(doc=str(produced), filename=Path(produced).name, items=len(items), bytes=Path(produced).stat().st_size) + + +def _groups(d: Path) -> dict[str, Any] | None: + import json + + p = d / "invoice_groups.json" + if not p.exists(): + return None + try: + data: dict[str, Any] = json.loads(p.read_text(encoding="utf-8")) + return data + except Exception: + return None + + +def _find_template() -> Path | None: + stem = REQUISITION_FILENAME.replace(".xls", "") + candidates = sorted(PROJECT_ROOT.glob(f"{stem}*.xls")) + return candidates[-1] if candidates else None diff --git a/src/mcp/tools/extraction.py b/src/mcp/tools/extraction.py index 263b45f..9d0047b 100644 --- a/src/mcp/tools/extraction.py +++ b/src/mcp/tools/extraction.py @@ -90,6 +90,48 @@ def case_save_extraction(case_id: str, source_file: str, data: dict[str, Any]) - return ok(cached=str(path), invoice_type=normalized.get("invoice_type"), warnings=warnings) +def case_patch_extraction(case_id: str, source_file: str, patch: dict[str, Any]) -> dict[str, Any]: + """局部修正已落盘的抽取结果（人工核对后改字段用）。 + + 只合并传入的键，返回每个被改字段的 from/to，避免静默覆盖； + 未在 schema 中出现的键会告警但仍写入（有些字段是给下游自定义用的）。 + """ + d = _store.resolve(case_id) + if d is None: + return fail(f"会话不存在: {case_id}") + if not isinstance(patch, dict) or not patch: + return fail("patch 必须是非空对象") + from ...core.extraction.cache import load_cache as _load + from ...core.extraction.cache import save_extraction + + current = _load(d).get(source_file) + if current is None: + return fail(f"该材料还没有抽取结果: {source_file}", next_action="先 case_save_extraction 落一份完整结果") + from ..schemas import DOCUMENT_SCHEMAS, INVOICE_TYPES + + doc_type = str(current.get("invoice_type", "")) + schema = DOCUMENT_SCHEMAS.get(doc_type, {}) + known = set(schema.get("required", [])) | set(schema.get("optional", [])) | {"invoice_type", "_source_file"} + warnings: list[str] = [] + changed: dict[str, dict[str, Any]] = {} + merged = dict(current) + for key, value in patch.items(): + if key.startswith("_"): + warnings.append(f"忽略内部字段 {key} 的修改") + continue + if key not in changed and merged.get(key) != value: + changed[key] = {"from": merged.get(key), "to": value} + if doc_type in INVOICE_TYPES and known and key not in known: + warnings.append(f"{key} 不在 {doc_type} 的字段定义里，确认拼写无误") + merged[key] = value + if not changed: + return ok(source_file=source_file, changed={}, warnings=warnings, note="patch 与现值相同，未做写入") + normalized, normalize_warnings = normalize_extraction(merged) + normalized["_source_file"] = source_file + path = save_extraction(d, source_file, normalized) + return ok(source_file=source_file, changed=changed, warnings=warnings + normalize_warnings, cached=str(path)) + + def extraction_spec(kind: str | None = None) -> dict[str, Any]: """返回抽取字段定义。kind 省略给全部文档类型；'info' 给 travel_info/normal_info 骨架。""" return ok(**schema_doc(kind))