diff --git a/src/infra/documents/__init__.py b/src/infra/documents/__init__.py index 6932619..7ad8bca 100644 --- a/src/infra/documents/__init__.py +++ b/src/infra/documents/__init__.py @@ -18,15 +18,19 @@ from .invoice import ( from .requisition import ( DEFAULT_PROJECT, REQUISITION_FILENAME, + TEMPLATE_ENV_VAR, fill_requisition_excel, generate_requisition, load_project_desc, load_requisition_items, + resolve_requisition_template, ) __all__ = [ "CACHE_DIR_NAME", "DEFAULT_PROJECT", + "TEMPLATE_ENV_VAR", + "resolve_requisition_template", "INVOICE_LEVEL_COLUMNS", "PAYMENT_RECORD_COLUMNS", "REQUISITION_FILENAME", diff --git a/src/infra/documents/requisition.py b/src/infra/documents/requisition.py index 9a47651..d8d207d 100644 --- a/src/infra/documents/requisition.py +++ b/src/infra/documents/requisition.py @@ -13,14 +13,15 @@ （旧 Web 会话目录回退 agent_state.json） 用法: - uv run python -m src.infra.documents.requisition --session src/web/uploads/35c63986686b - uv run python -m src.infra.documents.requisition --session src/web/uploads/0c9b7e7ef1ac --exclude 千斤顶 + uv run python -m src.infra.documents.requisition --session cases/20261019-6月毕业采集 + uv run python -m src.infra.documents.requisition --session cases/xxx --exclude 千斤顶 """ from __future__ import annotations import argparse import json +import os import re import shutil import time @@ -51,6 +52,11 @@ COL_DATE = 8 DEFAULT_PROJECT = "电子元器件及工具耗材采购" +# 模板查找：环境变量 REQUISITION_TEMPLATE -> templates/ -> 项目根（兼容旧位置）。 +# 模板应是空白件：只含标题行与表头行。 +TEMPLATE_ENV_VAR = "REQUISITION_TEMPLATE" +TEMPLATE_SUBDIR = "templates" + # 泛化规格名称：这类值对申购单无意义，统一替换为"多品种" GENERIC_SPECS = {"集成电路", "电阻", "连接器", "电子元件", "二极管", "芯片"} @@ -161,6 +167,27 @@ def load_project_desc(session_dir: str | Path, fallback: str = DEFAULT_PROJECT) return _read_desc(root_dir / AGENT_STATE_FILENAME, nested=True) or fallback +def resolve_requisition_template(project_root: str | Path | None = None) -> Path | None: + """定位申购单模板；找不到返回 None，由调用方决定怎么报错。 + + 优先级：环境变量 REQUISITION_TEMPLATE 指定的文件 → 项目内 templates/ 目录 → 项目根（兼容旧位置）。 + 同一目录内取名字最大的一个，方便并存多版本模板。 + """ + root = Path(project_root) if project_root is not None else Path(__file__).resolve().parents[3] + env = os.environ.get(TEMPLATE_ENV_VAR, "").strip() + if env: + candidate = Path(env) + if candidate.is_file(): + return candidate + log.warning("环境变量 %s 指向的文件不存在: %s", TEMPLATE_ENV_VAR, env) + stem = REQUISITION_FILENAME.replace(".xls", "") + for directory in (root / TEMPLATE_SUBDIR, root): + candidates = sorted(directory.glob(f"{stem}*.xls")) + if candidates: + return candidates[-1] + return None + + def fill_requisition_excel( template_path: str | Path, output_path: str | Path, @@ -169,8 +196,9 @@ def fill_requisition_excel( ) -> Path: """将申购明细填入 Excel 模板并另存为新文件。 - 模板须含标题行与表头行（表头: 序号|物品名称|规格|数量|单价(含税)|金额|对应实验实训项目|数据采集时间）， - 数据从第 3 行开始填写，末尾追加合计行。 + 模板须含标题行与表头行（表头: 序号|物品名称|规格|数量|单价(含税)|金额|对应实验实训项目|数据采集时间）。 + 写入前会先删除第 3 行起的既有内容，所以误拿已填写的文件当模板也不会残留旧明细， + 但模板本身应当是空白件（templates/低值易耗品申购.xls）。数据从第 3 行起填写，末尾追加合计行。 """ template_path = Path(template_path) output_path = Path(output_path) @@ -233,22 +261,25 @@ def generate_requisition( ) -> Path: """一站式生成申购单：读取缓存 -> 去重 -> 填表。 - 默认以项目根目录下已存在的 `低值易耗品申购*.xls` 为模板， - 输出为同目录下带时间戳的新文件。 + 模板由 resolve_requisition_template() 决定；输出默认落在会话目录的 artifacts/ 下， + 不再写到模板旁边——否则上一次的产物会被下一次当成模板。 """ session_dir = Path(session_dir) cache_dir = session_dir / CACHE_DIR_NAME template = Path(template_path) if template_path is not None else None if template is None: - root = Path(__file__).resolve().parents[3] - candidates = sorted(root.glob(f"{REQUISITION_FILENAME.replace('.xls', '')}*.xls")) - if not candidates: - raise FileNotFoundError(f"未找到申购单模板: {root / REQUISITION_FILENAME}") - template = candidates[-1] + template = resolve_requisition_template() + if template is None: + raise FileNotFoundError( + f"未找到申购单模板：放一份空白件到 templates/{REQUISITION_FILENAME}，" + f"或用环境变量 {TEMPLATE_ENV_VAR} 指定路径" + ) if output_path is None: ts = int(time.time() * 1000) - output_path = template.parent / f"低值易耗品申购{ts}.xls" + out_dir = session_dir / "artifacts" + out_dir.mkdir(parents=True, exist_ok=True) + output_path = out_dir / f"低值易耗品申购{ts}.xls" items = load_requisition_items(cache_dir, exclude_keywords=exclude_keywords) desc = project_desc or load_project_desc(session_dir) @@ -257,9 +288,13 @@ def generate_requisition( def main() -> None: parser = argparse.ArgumentParser(description="生成低值易耗品申购单 Excel") - parser.add_argument("--session", default=".", help="会话目录（含 .invoice_cache/）") - parser.add_argument("--template", default=None, help="申购单模板 xls 路径（默认取项目根目录现有模板）") - parser.add_argument("--output", default=None, help="输出 xls 路径（默认同目录带时间戳新文件）") + parser.add_argument("--session", default=".", help="报销会话目录（含 .invoice_cache/）") + parser.add_argument( + "--template", + default=None, + help="申购单模板 xls 路径（默认：环境变量 REQUISITION_TEMPLATE -> templates/ -> 项目根）", + ) + parser.add_argument("--output", default=None, help="输出 xls 路径（默认写在会话目录 artifacts/ 下，带时间戳）") parser.add_argument("--project", default=None, help="对应实验实训项目（默认读 normal_info.json）") parser.add_argument("--exclude", default="", help="排除物品名称含这些关键词的发票，逗号分隔（如: 千斤顶,台钻）") args = parser.parse_args() diff --git a/src/mcp/tools/documents.py b/src/mcp/tools/documents.py index bdbd8ac..f71ca79 100644 --- a/src/mcp/tools/documents.py +++ b/src/mcp/tools/documents.py @@ -11,7 +11,12 @@ from typing import Any from ...cases import ARTIFACTS_DIR_NAME, CACHE_DIR_NAME, PROJECT_ROOT, CaseStore from ...infra.com_thread import run_com -from ...infra.documents import REQUISITION_FILENAME, generate_requisition, load_requisition_items +from ...infra.documents import ( + REQUISITION_FILENAME, + generate_requisition, + load_requisition_items, + resolve_requisition_template, +) from ..envelope import fail, ok _store = CaseStore() @@ -32,7 +37,10 @@ def doc_requisition(case_id: str, exclude: list[str] | None = None, project: str return fail("没有可申购的普通发票", next_action="确认已抽取普通发票，或本会话是纯差旅无需申购单") ts_template = _find_template() if ts_template is None: - return fail(f"未找到申购单模板 {REQUISITION_FILENAME}*.xls", next_action="把模板放到项目根目录") + return fail( + f"未找到申购单空白模板 {REQUISITION_FILENAME}", + next_action="放一份到 templates/ 目录，或设置环境变量 REQUISITION_TEMPLATE 指向模板文件", + ) out = d / ARTIFACTS_DIR_NAME / f"低值易耗品申购_{d.name}.xls" out.parent.mkdir(parents=True, exist_ok=True) try: @@ -65,6 +73,5 @@ def _groups(d: Path) -> dict[str, Any] | 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 + """模板位置统一由 requisition.resolve_requisition_template 决定，不在这里另写一套。""" + return resolve_requisition_template(PROJECT_ROOT) diff --git a/templates/低值易耗品申购.xls b/templates/低值易耗品申购.xls new file mode 100644 index 0000000..67d753d Binary files /dev/null and b/templates/低值易耗品申购.xls differ diff --git a/tests/test_requisition.py b/tests/test_requisition.py index c5c3a3b..7e417e5 100644 --- a/tests/test_requisition.py +++ b/tests/test_requisition.py @@ -12,6 +12,7 @@ from src.infra.documents.requisition import ( guess_spec, load_project_desc, load_requisition_items, + resolve_requisition_template, ) @@ -175,3 +176,38 @@ class TestLoadProjectDesc: def test_falls_back_on_corrupt_file(self, tmp_path: Path) -> None: (tmp_path / "agent_state.json").write_text("not json", encoding="utf-8") assert load_project_desc(tmp_path) == "电子元器件及工具耗材采购" + + +class TestResolveRequisitionTemplate: + """模板解析不涉及 COM，可以纯单测覆盖优先级。""" + + def test_env_var_wins(self, tmp_path: Path, monkeypatch) -> None: + custom = tmp_path / "自定义模板.xls" + custom.write_bytes(b"x") + monkeypatch.setenv("REQUISITION_TEMPLATE", str(custom)) + assert resolve_requisition_template(tmp_path / "proj") == custom + + def test_templates_dir_beats_root(self, tmp_path: Path, monkeypatch) -> None: + monkeypatch.delenv("REQUISITION_TEMPLATE", raising=False) + (tmp_path / "templates").mkdir() + blank = tmp_path / "templates" / "低值易耗品申购.xls" + blank.write_bytes(b"a") + (tmp_path / "低值易耗品申购999.xls").write_bytes(b"b") + assert resolve_requisition_template(tmp_path) == blank + + def test_root_fallback_takes_latest_name(self, tmp_path: Path, monkeypatch) -> None: + monkeypatch.delenv("REQUISITION_TEMPLATE", raising=False) + (tmp_path / "低值易耗品申购111.xls").write_bytes(b"a") + newest = tmp_path / "低值易耗品申购222.xls" + newest.write_bytes(b"b") + assert resolve_requisition_template(tmp_path) == newest + + def test_env_pointing_nowhere_falls_back(self, tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("REQUISITION_TEMPLATE", str(tmp_path / "不存在.xls")) + legacy = tmp_path / "低值易耗品申购.xls" + legacy.write_bytes(b"b") + assert resolve_requisition_template(tmp_path) == legacy + + def test_missing_everywhere_returns_none(self, tmp_path: Path, monkeypatch) -> None: + monkeypatch.delenv("REQUISITION_TEMPLATE", raising=False) + assert resolve_requisition_template(tmp_path) is None