diff --git a/.agents/docs/good-experience/2026-06-11-CSV-BOM-列名校验失败.md b/.agents/docs/good-experience/2026-06-11-出现问题好的排查流程.md similarity index 100% rename from .agents/docs/good-experience/2026-06-11-CSV-BOM-列名校验失败.md rename to .agents/docs/good-experience/2026-06-11-出现问题好的排查流程.md diff --git a/.coverage b/.coverage index fab3f7c..76b6741 100644 Binary files a/.coverage and b/.coverage differ diff --git a/README.md b/README.md index f34b558..53b79a4 100644 --- a/README.md +++ b/README.md @@ -52,19 +52,66 @@ ```mermaid flowchart TB - PDF[PDF 发票 / 图片] --> Extract[extractor 提取] - Extract --> Classify{发票类型分类} + PDF[PDF 发票 / 图片] --> Extract[extractor 多模态提取] + Extract --> Cache[(.invoice_cache/*.json)] + + Cache --> Classify{发票类型分类} Classify -->|差旅发票| Travel[高铁票 / 酒店住宿] Classify -->|普通发票| General[普通发票] - Travel --> CSV[(invoice_summary.csv)] - General --> CSV - CSV -->|仅普通发票| Fill[fill_consumable_doc] + Classify -->|支付记录| Payment[支付截图] + Classify -->|申请单| Application[出差事前申请单] + + Travel --> Matcher[matcher 金额匹配] + Payment --> Matcher + Matcher --> MatchResult[(match_result.json)] + + Cache --> TravelLLM[LLM 差旅信息提取] + MatchResult --> TravelLLM + TravelLLM --> TravelInfo[(travel_info.json)] + + Cache --> NormalLLM[LLM 普通发票信息提取] + MatchResult --> NormalLLM + NormalLLM --> NormalInfo[(normal_info.json)] + + TravelInfo -->|差旅基本信息| Bot_T[bot/travel.py
差旅填报流程] + TravelInfo -->|报销明细| Bot_T + TravelInfo -->|支付方式| Bot_T + TravelInfo -->|补助清单| Bot_T + TravelInfo -->|附件清单| Bot_T + Bot_T --> Submit_T[差旅报销提交] + + NormalInfo -->|报销说明| Bot_G[bot/normal.py
普通填报流程] + NormalInfo -->|发票总数/金额| Bot_G + NormalInfo -->|支付方式| Bot_G + NormalInfo -->|附件清单| Bot_G + Bot_G --> Submit_G[普通报销提交] + + General --> CSV[(invoice_summary.csv)] + CSV --> Fill[fill_consumable_doc] Fill --> Doc[易耗品、出库单.doc] - CSV --> Bot[bot 浏览器自动化] - Bot -->|差旅模式| Submit_T[差旅报销填报] - Bot -->|普通模式| Submit_G[普通报销填报] ``` +### 关键中间产物 + +| 文件 | 生成阶段 | 作用 | +|------|---------|------| +| `.invoice_cache/*.json` | extractor 提取 | 单张发票/支付记录/申请单的结构化数据 | +| `match_result.json` | matcher 匹配 | 支付截图与发票的关联关系（按金额匹配） | +| `travel_info.json` | LLM 差旅信息提取 | 综合发票缓存 + 匹配结果，生成差旅报销所需的全部结构化数据 | +| `normal_info.json` | LLM 普通发票信息提取 | 综合普通发票 + 匹配结果，生成普通报销所需的全部结构化数据 | +| `invoice_summary.csv` | extractor 提取 | 普通发票汇总（用于生成易耗品出库单） | + +### bot 模块架构 + +`bot/` 包负责浏览器自动化填报，仅接收已提取的信息并执行填报操作，不承担信息提取职责： + +| 模块 | 职责 | +|------|------| +| `bot/base.py` | `BaseBot` 基类：浏览器生命周期、登录、导航、截图 | +| `bot/travel.py` | 差旅填报流程：基本信息 → 差旅明细 → 支付方式 → 补助清单 → 附件上传 | +| `bot/normal.py` | 普通填报流程：基本信息 → 总明细 → 支付方式 → 附件上传 | +| `bot/__init__.py` | 入口函数：`run_bot()` / `run_bot_web()`，负责类型判断和流程路由 | + ## 环境要求 - **Python 3.12+** diff --git a/src/README.md b/src/README.md index e64f3dc..3c59e0a 100644 --- a/src/README.md +++ b/src/README.md @@ -1,5 +1,5 @@ --- -last_reviewed: 2026-06-11 +last_reviewed: 2026-06-12 --- # src — 主源码目录 @@ -12,41 +12,42 @@ last_reviewed: 2026-06-11 |-----------|------| | `__init__.py` | 包初始化：提供 `get_logger()` 日志工厂（支持终端 + 文件双输出，按日期自动分文件） | | `config.py` | 配置加载：从 `config.json` 读取用户凭据和默认值，从环境变量读取服务端配置（SSO 地址、LLM 参数） | -| `pipeline.py` | 流程编排：串联发票提取 → 分类 → CSV 保存 → 浏览器填报，支持分步执行 | +| `pipeline.py` | 流程编排：串联发票提取 → 类型判断 → 差旅/普通信息提取 → 浏览器填报，支持分步执行 | | `main.py` | CLI 入口：支持 `--step` 分步执行、`-u/-p` 覆盖凭据、`--cache-dir` 指定缓存目录 | -| `bot.py` | 浏览器自动化：Playwright 驱动的财务系统填报机器人（含差旅/普通两种模式，支持 headless） | +| `bot/` | 浏览器自动化：Playwright 驱动的财务系统填报机器人（仅负责接收信息并填报） | | `doc/` | 文档处理模块：PDF 渲染、LLM 提取、支付匹配、发票分类、出库单生成 | | `web/` | Web 界面模块：Flask 应用、SSE 日志、可编辑表格、移动端上传、会话隔离 | ## 数据流 +```mermaid +graph TD + A[CLI/Web 入口] --> B["pipeline.py (编排)"] + B --> C["doc/extractor.py (统一提取入口)"] + C --> D["doc/pdf.py (PDF 渲染为图片)"] + C --> E["doc/llm_extractor.py (多模态 LLM 识别)"] + E --> F["发票 invoice_type=train/hotel/general"] + E --> G["支付记录 invoice_type=payment"] + E --> H["出差事前申请单 invoice_type=application"] + C --> I["doc/matcher.py (发票与支付记录按金额匹配)"] + I --> J["一对一匹配 发票数 == 刷卡数"] + I --> K["一对多匹配 贪心算法 相对容差 3%"] + C --> L["doc/invoice.py (CSV/JSON 读写)"] + L --> M["payment_records.csv (支付记录级别)"] + L --> N["invoice_summary.csv (发票级别)"] + L --> O["travel_applications.json (出差申请单)"] + B --> R{"判断报销类型"} + R -->|差旅| T["doc/llm_extractor.py (差旅信息提取)"] + R -->|普通| V["doc/llm_extractor.py (普通发票信息提取)"] + T --> W["travel_info.json (差旅信息: 交通/住宿明细、补贴、附件清单)"] + V --> X["normal_info.json (普通发票信息: 报销说明、发票总数、总金额、支付方式、附件清单)"] + W --> P["bot/ (浏览器填报 - 仅接收信息并填报)"] + X --> P + P --> Q["差旅模式: travel_info.json → 填报差旅单 → 上传差旅附件"] + P --> S["普通模式: 基本信息 → 录入明细 → 支付信息 → 上传附件"] ``` -CLI/Web 入口 → pipeline.py (编排) - ↓ - doc/extractor.py (统一提取入口) - ↓ - ┌──── doc/pdf.py (PDF 渲染为图片) - │ - ├── doc/llm_extractor.py (多模态 LLM 识别) - │ ├── 发票 (invoice_type=train/hotel/general) - │ ├── 支付记录 (invoice_type=payment) - │ └── 出差事前申请单 (invoice_type=application) - │ - ├── doc/matcher.py (发票与支付记录按金额匹配) - │ ├── 一对一匹配 (发票数 == 刷卡数) - │ └── 一对多匹配 (贪心算法，相对容差 3%) - │ - └── doc/invoice.py (CSV/JSON 读写) - ├── payment_records.csv (支付记录级别) - ├── invoice_summary.csv (发票级别) - └── travel_applications.json (出差申请单) - ↓ - doc/fill_consumable_doc.py (普通发票 → Word 出库单) - ↓ - bot.py (浏览器填报) - ├── 差旅模式: 差旅信息提取 → 填报差旅单 → 上传差旅附件 - └── 普通模式: 基本信息 → 录入明细 → 支付信息 → 上传附件 -``` + +**数据流变更（2026-06-11）：** 差旅信息提取从 `bot.py` 提升到 `pipeline.py` 编排层。在发票提取和匹配完成后立即判断报销类型，差旅发票调用 LLM 提取 `travel_info.json`，普通发票调用 LLM 提取 `normal_info.json`。Bot 仅负责接收信息并填报，不再承担信息提取职责。 ## 文档处理子模块 (`doc/`) @@ -57,6 +58,7 @@ CLI/Web 入口 → pipeline.py (编排) - **JSON 缓存**：提取结果缓存于 `.invoice_cache/`，避免重复处理 - **金额匹配**：支持一对多匹配，相对容差 3%，未匹配发票单独列为记录 - **差旅信息提取**：综合发票、支付记录和匹配结果，提取出差事由、地点、时间等 +- **普通发票信息提取**：综合普通发票、支付记录和匹配结果，提取报销说明、发票总数、总金额、支付方式、附件清单 - **出库单生成**：将 CSV 数据填入 Word 模板（pywin32 COM，仅 Windows） ## Web 界面子模块 (`web/`) diff --git a/src/__init__.py b/src/__init__.py index 8893098..1e3581a 100644 --- a/src/__init__.py +++ b/src/__init__.py @@ -22,15 +22,26 @@ def get_logger(name: str) -> logging.Logger: 输出格式: 2026-05-24 12:34:56 [INFO ] extractor: 扫描目录: ... 日志同时输出到终端和 logs/<日期>.log + + 只在最顶层的 logger 上添加标准 handler（stream + file）， + 子 logger 通过 propagate 将日志传递给父 logger 统一处理。 + 这样 _SSELogHandler 只需挂在父 logger 上即可捕获所有子日志。 """ logger = logging.getLogger(name) + logger.setLevel(logging.INFO) - # 使用专属标记判断是否已初始化标准 handler（stream + file）， - # 避免被 _SSELogHandler 等外部 handler 干扰 - if not getattr(logger, "_standard_handlers_initialized", False): - logger.setLevel(logging.INFO) + # 检查是否有父 logger 已初始化标准 handler + # 如果有，子 logger 不重复添加，靠 propagate 传递即可 + parent_name = name.rsplit(".", 1)[0] if "." in name else None + parent_has_handlers = False + if parent_name: + parent = logging.getLogger(parent_name) + parent_has_handlers = getattr(parent, "_standard_handlers_initialized", False) + + if not getattr(logger, "_standard_handlers_initialized", False) and not parent_has_handlers: formatter = logging.Formatter(_LOG_FMT, _LOG_DATE_FMT) + # 只有没有已初始化的父 logger 时，才添加标准 handler # 终端输出 utf8_stream = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") stream_handler = logging.StreamHandler(utf8_stream) diff --git a/src/bot.py b/src/bot.py deleted file mode 100644 index b151cdd..0000000 --- a/src/bot.py +++ /dev/null @@ -1,649 +0,0 @@ -""" -浏览器自动化填报 - -使用 Playwright 操作财务报销系统，自动完成登录、填单、上传附件等操作。 - -对外接口: - run_bot(config, invoices) 启动浏览器并执行填报流程 -""" - -import json -from pathlib import Path -from typing import Any - -from . import get_logger - -log = get_logger("bot") - - -# ------------------------------------------------------------------ -# 日期格式化 -# ------------------------------------------------------------------ - - -def _format_date(date_str: str) -> str: - """将 '2026/5/13' 或 '2026-5-13' 转为 '2026-05-13'""" - if not date_str: - return "" - parts = date_str.replace("-", "/").split("/") - if len(parts) == 3: - return f"{parts[0].zfill(4)}-{parts[1].zfill(2)}-{parts[2].zfill(2)}" - return date_str - - -def _classify_invoice_batch( - invoices: list[dict[str, str]], -) -> dict[str, list[dict[str, str]]]: - """按发票类型分组""" - travel: list[dict[str, str]] = [] - general: list[dict[str, str]] = [] - application: list[dict[str, str]] = [] - for inv in invoices: - inv_type = inv.get("invoice_type", "general") - if inv_type == "application": - application.append(inv) - elif inv_type in ("train", "hotel"): - travel.append(inv) - else: - general.append(inv) - return {"travel": travel, "general": general, "application": application} - - -# ------------------------------------------------------------------ -# 报销机器人 -# ------------------------------------------------------------------ - - -class ReimburseBot: - """财务报销自动化机器人""" - - def __init__(self, config: dict[str, Any], headless: bool = False, work_dir: Path | None = None): - self.config = config - self.headless = headless - self.work_dir: Path | None = None - self.browser: Any = None - self.context: Any = None - self.page: Any = None - - from playwright.sync_api import sync_playwright - - self._pw_ctx = sync_playwright() - self.pw = self._pw_ctx.__enter__() - - def launch(self) -> None: - """启动浏览器""" - self.browser = self.pw.chromium.launch(headless=self.headless) - self.context = self.browser.new_context(viewport={"width": 1360, "height": 768}) - self.page = self.context.new_page() - self.page.set_default_timeout(30000) - - def login_portal(self) -> None: - """登录信息门户""" - log.info("登录信息门户...") - - self.page.goto(self.config["sso_login_url"], wait_until="domcontentloaded") - self._wait_for('text="微信扫码登录"', timeout=5000) - - try: - self.page.fill('input[placeholder*="工号"], input[placeholder*="学号"]', self.config["username"]) - self.page.fill('input[placeholder*="密码"]', self.config["password"]) - except Exception: - log.warning("未找到登录输入框，可能已登录") - - try: - checkbox = self.page.query_selector('input[type="checkbox"]') - if checkbox and not checkbox.is_checked(): - checkbox.click() - except Exception: - pass - - for selector in ['button:has-text("登录")', 'input[value="登录"]', 'text="登录"']: - try: - self.page.click(selector, timeout=3000) - break - except Exception: - continue - - self._wait_for_portal() - - def _wait_for_portal(self) -> None: - """等待跳转到统一信息平台""" - for _ in range(30): - self.page.wait_for_timeout(1000) - url = self.page.url - if any(kw in url for kw in ("tyrz.fynu.edu.cn/zs-uip", "tyrz.fynu.edu.cn/oshall", "portal")): - self._screenshot("portal_loaded") - return - log.error("等待门户跳转超时") - self._screenshot("portal_timeout") - raise TimeoutError("登录超时，未跳转到信息门户") - - def navigate_to_reimburse(self, page_key: str = "reimburse_page") -> None: - """从统一信息平台进入报销系统""" - log.info("进入报销系统...") - self._wait_for('text="快捷入口"', timeout=5000) - - try: - self.page.click('text="财务系统"', timeout=5000) - except Exception: - log.warning("未找到财务系统入口") - - new_tab = None - for _ in range(15): - self.page.wait_for_timeout(1000) - for p in self.context.pages: - if "dddl" in p.url or "210.45.32.214" in p.url: - new_tab = p - break - if new_tab: - break - - if new_tab: - self.page = new_tab - self._wait_for('text="网络报销"', timeout=5000) - else: - log.warning(f"未找到单点登录页面，当前 URL: {self.page.url}") - - for p in self.context.pages[:-1]: - try: - p.close() - except Exception: - pass - - try: - link = self.page.query_selector('a:has(img[src*="wlbx"])') - if link: - reimburse_url = link.get_attribute("href") - self.page.goto(reimburse_url, wait_until="domcontentloaded", timeout=15000) - except Exception: - pass - - self._wait_for('text="报销录入"', timeout=5000) - common_url = self.config["reimburse_url"] + self.config[page_key] - self.page.goto(common_url, wait_until="domcontentloaded", timeout=15000) - self._wait_for('text="单据状态:"', timeout=5000) - - def create_new_form(self) -> None: - """点击「新增」创建新单据""" - log.info("创建新单据...") - self.page.wait_for_timeout(2000) - - try: - self.page.click("#insert", timeout=5000) - except Exception: - try: - self.page.click("text=新增", timeout=3000) - except Exception as err: - self._screenshot("no_add_button") - raise RuntimeError("无法点击新增按钮") from err - - self.page.wait_for_timeout(3000) - self._screenshot("after_add_click") - - def fill_travel_info(self, basic_info: dict[str, Any]) -> None: - """填写差旅报销基本信息""" - log.info("填写差旅报销信息...") - - try: - self.page.fill("#CAUSE", basic_info.get("travel_purpose", "")) - self.page.fill("#SITE", basic_info.get("travel_location", "")) - - self.page.click("#PROJECTCODE", timeout=10000) - self.page.wait_for_timeout(1000) - self.page.wait_for_selector("#promodal .fixed-table-body tbody tr", timeout=10000) - first_row = self.page.query_selector("#promodal .fixed-table-body tbody tr") - if first_row: - first_row.click() - self.page.wait_for_timeout(1000) - - self.page.fill("#THEKSRQ", _format_date(basic_info.get("start_date", ""))) - self.page.fill("#THEJSRQ", _format_date(basic_info.get("end_date", ""))) - self.page.click("#saveAndNext", timeout=5000) - self.page.wait_for_timeout(2000) - self._screenshot("travel_basic_done") - except Exception: - log.error("填写基本信息失败") - self._screenshot("travel_basic_error") - - def fill_basic_info(self, description: str = "元器件采购报销") -> None: - """填写基本信息""" - log.info("填写基本信息...") - - try: - self.page.fill("#EXPENEXPLAIN", description) - except Exception: - pass - - try: - self.page.click("#PROJECTCODE", timeout=10000) - self.page.wait_for_timeout(1000) - except Exception: - pass - - self._screenshot("step3_project_modal") - - try: - self.page.wait_for_selector("#promodal .fixed-table-body tbody tr", timeout=10000) - first_row = self.page.query_selector("#promodal .fixed-table-body tbody tr") - if first_row: - first_row.click() - self.page.wait_for_timeout(1000) - except Exception: - pass - - self._screenshot("step3_project_selected") - - try: - self.page.click("#saveAndNext", timeout=5000) - self.page.wait_for_timeout(2000) - except Exception: - pass - - self._screenshot("step3_done") - - def add_travel_items(self, travel_items: dict[str, Any]) -> None: - """录入差旅报销明细""" - vehicle_map = { - "火车": "01", - "汽车": "02", - "轮船": "03", - "自带车": "04", - "公务车": "05", - "飞机": "06", - "租车": "07", - "自驾车": "08", - } - try: - traffic_info = travel_items.get("transport_fee") or [] - for item in traffic_info: - self.page.click("#insertDetail", timeout=5000) - self._wait_for('text="增加明细"', timeout=5000) - self.page.select_option("#cost", "1") - self.page.wait_for_timeout(500) - vehicle = item.get("vehicle_type", "") - if vehicle in vehicle_map: - self.page.select_option("#jtgj", vehicle_map[vehicle]) - - self.page.fill("#ksdd", item.get("departure_place", "")) - self.page.fill("#jsdd", item.get("arrival_place", "")) - self.page.fill( - '#t1 input[name="expenPwTraveldetail.MONEY"]', - str(item.get("amount", "")), - ) - self.page.fill( - '#t1 input[name="expenPwTraveldetail.HOWBILL"]', - str(item.get("bill_count", "")), - ) - self.page.fill( - '#t1 input[name="expenPwTraveldetail.SMARK"]', - str(item.get("remark", "")), - ) - self.page.click("#detailAdd", timeout=3000) - self.page.wait_for_timeout(1000) - - hotel_info = travel_items.get("hotel_fee") or [] - for item in hotel_info: - self.page.click("#insertDetail", timeout=5000) - self._wait_for('text="增加明细"', timeout=5000) - self.page.select_option("#cost", "2") - self.page.wait_for_timeout(500) - self.page.fill("#ksrq2", _format_date(str(item.get("checkin_date", "")))) - self.page.fill("#jsrq2", _format_date(str(item.get("checkout_date", "")))) - self.page.fill("#ts2", str(item.get("days", ""))) - self.page.fill("#rs2", str(item.get("person_count", ""))) - self.page.fill( - '#t2 input[name="expenPwTraveldetail.FPMONEY"]', - str(item.get("invoice_amount", "")), - ) - self.page.fill( - '#t2 input[name="expenPwTraveldetail.MONEY"]', - str(item.get("reimburse_amount", "")), - ) - self.page.fill( - '#t2 input[name="expenPwTraveldetail.SMARK"]', - str(item.get("remark", "")), - ) - self.page.click("#detailAdd", timeout=3000) - self.page.wait_for_timeout(1000) - - conference_info = travel_items.get("conference_fee") or [] - for item in conference_info: - self.page.click("#insertDetail", timeout=5000) - self._wait_for('text="增加明细"', timeout=5000) - self.page.select_option("#cost", "3") - self.page.wait_for_timeout(500) - self.page.fill( - '#t3 input[name="expenPwTraveldetail.HOWBILL"]', - str(item.get("bill_count", "")), - ) - self.page.fill( - '#t3 input[name="expenPwTraveldetail.MONEY"]', - str(item.get("amount", "")), - ) - self.page.fill( - '#t3 input[name="expenPwTraveldetail.SMARK"]', - str(item.get("remark", "")), - ) - self.page.click("#detailAdd", timeout=3000) - self.page.wait_for_timeout(1000) - except Exception as e: - log.error(f"录入总明细失败: {e}") - self._screenshot("item_total_error") - raise - - def add_reimburse_items(self, invoices: list[dict[str, Any]]) -> None: - """录入报销明细（一条总明细）""" - card_amount = sum(inv["card_amount"] for inv in invoices) - log.info(f"录入报销明细 (合计 ¥{card_amount:.2f})...") - - try: - self.page.click("#insertDetail", timeout=5000) - self._wait_for('text="经济事项名称"', timeout=5000) - self.page.click("#economicscode2") - self.page.wait_for_timeout(1000) - - try: - self.page.wait_for_selector("#econmodal .fixed-table-body tbody tr", timeout=10000) - rows = self.page.query_selector_all("#econmodal .fixed-table-body tbody tr") - if len(rows) >= 3: - rows[2].click() - self.page.wait_for_timeout(1000) - except Exception: - pass - - self.page.fill('input[name="expenPwCommondetail.HOWBILLS"]', f"{len(invoices)}") - self.page.fill("#je_zwzcdz", f"{card_amount:.2f}") - self.page.click("#detailAdd", timeout=3000) - self.page.wait_for_timeout(1000) - - self._screenshot("item_total") - except Exception as e: - log.error(f"录入总明细失败: {e}") - self._screenshot("item_total_error") - raise - - def fill_travel_payment(self, payment_info: list[dict[str, Any]]) -> None: - """录入差旅支付信息""" - log.info("录入差旅支付信息...") - try: - self.page.click('text="下一步(支付方式)"', timeout=5000) - self._wait_for('text="下一步(补助清单)"', timeout=5000) - - for info in payment_info: - self.page.click("#insertPay", timeout=5000) - self.page.wait_for_timeout(1000) - self.page.fill("#personid2", self.config["default_name"]) - self.page.fill("#accountname2", self.config["default_person_id"]) - self.page.fill("#receiptdate2", _format_date(info["card_date"])) - self.page.fill("#localaccount2", self.config["default_card_no"]) - self.page.fill("#receiptmoney2", str(info["card_amount"])) - self.page.fill("#money2", str(info["card_amount"])) - self.page.fill("#merchant2", info.get("merchant", "")) - self.page.fill("#smark2", info.get("remark", "")) - self.page.click("#payAdd", timeout=3000) - self.page.wait_for_timeout(1000) - except Exception as e: - log.error(f"支付方式录入失败: {e}") - self._screenshot("step5_error") - raise - - self._screenshot("step5_done") - - def fill_payment(self, invoices: list[dict[str, Any]]) -> None: - """录入支付信息""" - log.info("录入支付信息...") - try: - self.page.click('text="下一步(支付方式)"', timeout=5000) - self._wait_for('text="下一步(附件清单)"', timeout=5000) - - for inv in invoices: - self.page.click("#insertPay", timeout=5000) - self.page.wait_for_timeout(1000) - self.page.fill("#personid2", inv["person_id"]) - self.page.fill("#accountname2", inv["person_name"]) - self.page.fill("#receiptdate2", inv["card_date"]) - self.page.fill("#localaccount2", inv["card_no"]) - self.page.fill("#receiptmoney2", str(inv["card_amount"])) - self.page.fill("#money2", str(inv["card_amount"])) - self.page.fill("#merchant2", inv["seller_name"]) - self.page.fill("#smark2", inv["remark"]) - self.page.click("#payAdd", timeout=3000) - self.page.wait_for_timeout(1000) - except Exception as e: - log.error(f"支付方式录入失败: {e}") - self._screenshot("step5_error") - raise - - self._screenshot("step5_done") - - def upload_attachments(self, invoices: list[dict[str, Any]]) -> None: - """上传附件""" - log.info("上传附件...") - try: - self.page.click("#next4", timeout=5000) - self._wait_for("#submit2", timeout=5000) - - attachment_files = sorted((self.work_dir or Path(__file__).parent.parent).glob("*.pdf")) - if not attachment_files: - log.warning("未找到附件 PDF，跳过附件上传") - return - - for i, inv in enumerate(invoices): - file_path = attachment_files[i] if i < len(attachment_files) else None - try: - self._wait_for("#insertAcc", timeout=20000) - self.page.click("#insertAcc", timeout=5000) - self._wait_for("#fjlx", timeout=5000) - self.page.select_option("#fjlx", "1") - explanation = f"{inv['item_name']} - {inv['invoice_no']}" - self.page.fill("#fpsmxx", explanation) - except Exception: - pass - - if file_path and file_path.exists(): - try: - self.page.set_input_files("#file", str(file_path)) - self.page.wait_for_timeout(1000) - except Exception as e: - log.error(f"文件上传失败: {e}") - try: - self.page.click("#cjtj", timeout=5000) - except Exception: - try: - self.page.press("body", "Escape") - except Exception: - pass - log.info("上传附件完成") - except Exception as e: - log.error(f"附件上传失败: {e}") - self._screenshot("step6_error") - raise - - self._screenshot("step6_done") - - def upload_travel_attachments(self, attachment_info: list[dict[str, Any]]) -> None: - """上传差旅附件""" - log.info("上传差旅附件...") - try: - self.page.click("#next4", timeout=5000) - self._wait_for("#submit2", timeout=5000) - - for info in attachment_info: - attachment_file = self.work_dir / info["filename"] - self._wait_for("#insertAcc", timeout=20000) - self.page.click("#insertAcc", timeout=5000) - self._wait_for("#fjlx", timeout=5000) - if info["attachment_type"] == "invoice": - self.page.select_option("#fjlx", "1") - else: - self.page.select_option("#fjlx", "2") - self.page.fill("#fpsmxx", info["attachment_desc"]) - if attachment_file and attachment_file.exists(): - self.page.set_input_files("#file", str(attachment_file)) - self.page.wait_for_timeout(1000) - self.page.click("#cjtj", timeout=5000) - log.info("上传差旅附件完成") - except Exception as e: - log.error(f"差旅附件上传失败: {e}") - self._screenshot("travel_attachment_error") - raise - self._screenshot("travel_attachment_done") - - def close(self) -> None: - """关闭浏览器""" - if self.context: - self.context.close() - if self.browser: - self.browser.close() - try: - self._pw_ctx.__exit__(None, None, None) - except Exception: - pass - - def _wait_for(self, selector: str, timeout: int | None = None) -> None: - self.page.wait_for_selector(selector, timeout=timeout) - - def _screenshot(self, name: str) -> None: - img_dir = Path(__file__).parent.parent / "images" - img_dir.mkdir(exist_ok=True) - self.page.screenshot(path=str(img_dir / f"debug_{name}.png")) - - def fill_travel_subsidy(self, subsidy_info: list[dict[str, Any]]) -> None: - """录入差旅补助清单""" - log.info("录入差旅补助清单...") - try: - self.page.click("#next3", timeout=5000) - self._wait_for("#next4", timeout=5000) - - for info in subsidy_info: - self.page.click("#insertSubsidy", timeout=5000) - self._wait_for('text="增加补助清单"', timeout=5000) - self.page.click("#jzg3", timeout=5000) - self.page.wait_for_timeout(500) - if info["person_name"] and info["person_name"] != "": - self.page.fill("#seacher", info["person_name"]) - elif info["person_id"] and info["person_id"] != "": - self.page.fill("#seacher", info["person_id"]) - else: - raise ValueError(f"人员编号和人员姓名不能同时为空: {info}") - self.page.click("#cx", timeout=5000) - self.page.wait_for_selector("div.fixed-table-loading", state="hidden", timeout=10000) - self.page.click("#tableEmp tbody tr", timeout=10000) - self.page.wait_for_timeout(1000) - - open_bank = self.page.input_value("#openbank1") - if not open_bank: - log.info("员工开户行未填写，默认填写中国工商银行") - self.page.fill("#openbank1", "中国工商银行") - - self.page.fill("#startdate1", _format_date(info["start_date"])) - self.page.fill("#enddate1", _format_date(info["end_date"])) - self.page.fill("#trafficdays1", str(info["days"])) - self.page.fill("#fooddays1", str(info["days"])) - self.page.fill("#trafficnorm1", str(80)) - self.page.fill("#foodnorm1", str(100)) - trafficmoney = int(info["days"]) * 80 - foodmoney = int(info["days"]) * 100 - subsidymoney = trafficmoney + foodmoney - self.page.fill("#trafficmoney1", str(trafficmoney)) - self.page.fill("#foodmoney1", str(foodmoney)) - self.page.fill("#subsidymoney1", str(subsidymoney)) - self.page.click("#add", timeout=3000) - self.page.wait_for_timeout(1000) - except Exception as e: - log.error(f"差旅补助清单录入失败: {e}") - self._screenshot("subsidy_error") - raise - - self._screenshot("subsidy_done") - - -# ------------------------------------------------------------------ -# 对外入口 -# ------------------------------------------------------------------ - - -def run_bot(config: dict[str, Any], headless: bool = False, work_dir: Path | None = None) -> None: - """执行完整的浏览器填报流程，根据发票类型自动路由""" - if not config["username"] or not config["password"]: - raise ValueError("缺少用户名或密码") - - if not work_dir: - raise ValueError("缺少工作目录") - - from .doc.llm_extractor import load_cache - - cache_map = load_cache(work_dir) - doc_values = [v for k, v in cache_map.items() if k != "travel_info"] - invoices = [data for data in doc_values if data.get("invoice_type") not in ("application", "payment")] - applications = [data for data in doc_values if data.get("invoice_type") == "application"] - - groups = _classify_invoice_batch(invoices) - travel_invoices = groups["travel"] - general_invoices = groups["general"] - - log.info( - f"发票分类: 差旅 {len(travel_invoices)} 张, 普通 {len(general_invoices)} 张, " - f"出差事前申请单 {len(applications)} 份" - ) - - if travel_invoices and general_invoices: - raise ValueError( - f"发票类型混合（差旅 {len(travel_invoices)} 张 + 普通 {len(general_invoices)} 张），不支持混报，请分开提交" - ) - - is_travel = bool(travel_invoices) - travel_info: dict[str, Any] = cache_map.get("travel_info", {}) - if is_travel and not travel_info: - from .doc.llm_extractor import CACHE_DIR_NAME, extract_travel_info - - travel_info = extract_travel_info(source_dir=work_dir) - cache_dir = work_dir / CACHE_DIR_NAME - cache_dir.mkdir(parents=True, exist_ok=True) - with open(cache_dir / "travel_info.json", "w", encoding="utf-8") as f: - json.dump(travel_info, f, ensure_ascii=False, indent=2) - log.info("差旅信息已保存到缓存") - - bot = ReimburseBot(config, headless=headless, work_dir=work_dir) - bot.work_dir = work_dir - try: - bot.launch() - bot.login_portal() - - if is_travel: - log.info("处理差旅发票...") - bot.navigate_to_reimburse(page_key="travel_page") - bot.create_new_form() - basic_info = travel_info["basic_info"] - bot.fill_travel_info(basic_info) - travel_items = travel_info["reimbursement_details"] - bot.add_travel_items(travel_items) - payment_info = travel_info["payment_methods"] - bot.fill_travel_payment(payment_info) - subsidy_info = travel_info["subsidy_list"] - bot.fill_travel_subsidy(subsidy_info) - attachment_info = travel_info["attachments"] - bot.upload_travel_attachments(attachment_info) - else: - log.info("处理普通发票...") - bot.navigate_to_reimburse(page_key="reimburse_page") - bot.create_new_form() - bot.fill_basic_info() - bot.add_reimburse_items(general_invoices) - bot.fill_payment(general_invoices) - bot.upload_attachments(general_invoices) - - except Exception as e: - log.error(f"操作失败: {e}") - try: - bot._screenshot("error") - except Exception: - pass - raise - finally: - bot.close() - - -def run_bot_web(config: dict[str, Any], work_dir: Path) -> None: - """Web 模式填报 — headless，附件从指定目录读取""" - run_bot(config, headless=True, work_dir=work_dir) diff --git a/src/bot/README.md b/src/bot/README.md new file mode 100644 index 0000000..97ea030 --- /dev/null +++ b/src/bot/README.md @@ -0,0 +1,33 @@ +--- +last_reviewed: 2026-06-12 +--- + +# bot — 浏览器自动化填报模块 + +使用 Playwright 操作财务报销系统，自动完成登录、填单、上传附件等操作。 + +## 模块清单 + +| 文件 | 说明 | +|------|------| +| `__init__.py` | 对外入口：`run_bot()` 和 `run_bot_web()`，负责类型判断和流程路由 | +| `base.py` | `BaseBot` 基类：浏览器生命周期、登录、导航、截图、日期格式化 | +| `travel.py` | 差旅报销填报流程：基本信息 → 差旅明细 → 支付方式 → 补助清单 → 附件上传 | +| `normal.py` | 普通发票报销填报流程：基本信息 → 总明细 → 支付方式 → 附件上传 | + +## 架构设计 + +``` +run_bot(config, travel_info, normal_info) + ├── 创建 BaseBot，启动浏览器，登录门户 + ├── travel_info 存在 → travel.run(bot, travel_info) + └── normal_info 存在 → normal.run(bot, normal_info) +``` + +- **`BaseBot`** 只保留公共操作（launch、login、navigate、create_new_form、close、screenshot） +- **差旅/普通流程** 作为独立函数接受 `bot: BaseBot` 参数，符合函数式编程偏好 +- **`__init__.py`** 仅做路由分发，不包含具体填报逻辑 + +## 变更历史 + +- **2026-06-12**：从 `bot.py` 单文件重构为 `bot/` 包，分离差旅和普通报销逻辑 \ No newline at end of file diff --git a/src/bot/__init__.py b/src/bot/__init__.py new file mode 100644 index 0000000..df2dcb9 --- /dev/null +++ b/src/bot/__init__.py @@ -0,0 +1,93 @@ +""" +浏览器自动化填报 + +使用 Playwright 操作财务报销系统，自动完成登录、填单、上传附件等操作。 + +对外接口: + run_bot(config, travel_info, normal_info) 启动浏览器并执行填报流程 + run_bot_web(config, work_dir) Web 模式填报（从缓存加载信息） +""" + +from pathlib import Path +from typing import Any + +from .. import get_logger +from .base import BaseBot + +log = get_logger("bot") + + +def run_bot( + config: dict[str, Any], + headless: bool = False, + work_dir: Path | None = None, + travel_info: dict[str, Any] | None = None, + normal_info: dict[str, Any] | None = None, +) -> None: + """执行完整的浏览器填报流程，根据发票类型自动路由 + + Args: + config: 配置字典。 + headless: 是否无头模式。 + work_dir: 工作目录。 + travel_info: 差旅信息（由 pipeline 层提前提取并传入，非差旅时传 None）。 + normal_info: 普通发票信息（由 pipeline 层提前提取并传入，非普通时传 None）。 + """ + if not config["username"] or not config["password"]: + raise ValueError("缺少用户名或密码") + + if not work_dir: + raise ValueError("缺少工作目录") + + bot = BaseBot(config, headless=headless) + bot.work_dir = work_dir + + try: + bot.launch() + bot.login_portal() + + if travel_info is not None: + log.info("处理差旅发票...") + bot.navigate_to_reimburse(page_key="travel_page") + bot.create_new_form() + from . import travel + + travel.run(bot, travel_info) + elif normal_info is not None: + log.info("处理普通发票...") + bot.navigate_to_reimburse(page_key="reimburse_page") + bot.create_new_form() + from . import normal + + normal.run(bot, normal_info) + else: + raise ValueError("缺少差旅信息（travel_info）和普通发票信息（normal_info），无法继续填报") + + except Exception as e: + log.error(f"操作失败: {e}") + try: + bot._screenshot("error") + except Exception: + pass + raise + finally: + bot.close() + + +def run_bot_web(config: dict[str, Any], work_dir: Path) -> None: + """Web 模式填报 — headless，附件从指定目录读取 + + Web 端的信息提取由 app.py 的管道负责，此处从缓存加载。 + """ + from ..doc.llm_extractor import load_cache + + cache_map = load_cache(work_dir) + travel_info = cache_map.get("travel_info") + normal_info = cache_map.get("normal_info") + run_bot( + config, + headless=True, + work_dir=work_dir, + travel_info=travel_info, + normal_info=normal_info, + ) diff --git a/src/bot/base.py b/src/bot/base.py new file mode 100644 index 0000000..bfd81e4 --- /dev/null +++ b/src/bot/base.py @@ -0,0 +1,204 @@ +""" +浏览器自动化填报 — 公共基类 + +提供浏览器生命周期管理、登录、导航、截图等公共操作。 +""" + +from pathlib import Path +from typing import Any + +from .. import get_logger + +log = get_logger("bot") + + +# ------------------------------------------------------------------ +# 工具函数 +# ------------------------------------------------------------------ + + +def format_date(date_str: str) -> str: + """将 '2026/5/13' 或 '2026-5-13' 转为 '2026-05-13'""" + if not date_str: + return "" + parts = date_str.replace("-", "/").split("/") + if len(parts) == 3: + return f"{parts[0].zfill(4)}-{parts[1].zfill(2)}-{parts[2].zfill(2)}" + return date_str + + +# ------------------------------------------------------------------ +# 基类 +# ------------------------------------------------------------------ + + +class BaseBot: + """浏览器自动化基类 — 管理浏览器生命周期与公共操作""" + + def __init__(self, config: dict[str, Any], headless: bool = False) -> None: + self.config = config + self.headless = headless + self.work_dir: Path | None = None + self.browser: Any = None + self.context: Any = None + self.page: Any = None + + from playwright.sync_api import sync_playwright + + self._pw_ctx = sync_playwright() + self.pw = self._pw_ctx.__enter__() + + # --------------------------------------------------------------- + # 浏览器生命周期 + # --------------------------------------------------------------- + + def launch(self) -> None: + """启动浏览器""" + self.browser = self.pw.chromium.launch(headless=self.headless) + self.context = self.browser.new_context(viewport={"width": 1360, "height": 768}) + self.page = self.context.new_page() + self.page.set_default_timeout(30000) + + def close(self) -> None: + """关闭浏览器""" + if self.context: + self.context.close() + if self.browser: + self.browser.close() + try: + self._pw_ctx.__exit__(None, None, None) + except Exception: + pass + + # --------------------------------------------------------------- + # 登录 + # --------------------------------------------------------------- + + def login_portal(self) -> None: + """登录信息门户""" + log.info("登录信息门户...") + + self.page.goto(self.config["sso_login_url"], wait_until="domcontentloaded") + self._wait_for('text="微信扫码登录"', timeout=5000) + + try: + self.page.fill( + 'input[placeholder*="工号"], input[placeholder*="学号"]', + self.config["username"], + ) + self.page.fill('input[placeholder*="密码"]', self.config["password"]) + except Exception: + log.warning("未找到登录输入框，可能已登录") + + try: + checkbox = self.page.query_selector('input[type="checkbox"]') + if checkbox and not checkbox.is_checked(): + checkbox.click() + except Exception: + pass + + for selector in ['button:has-text("登录")', 'input[value="登录"]', 'text="登录"']: + try: + self.page.click(selector, timeout=3000) + break + except Exception: + continue + + self._wait_for_portal() + + def _wait_for_portal(self) -> None: + """等待跳转到统一信息平台""" + for _ in range(30): + self.page.wait_for_timeout(1000) + url = self.page.url + if any( + kw in url + for kw in ( + "tyrz.fynu.edu.cn/zs-uip", + "tyrz.fynu.edu.cn/oshall", + "portal", + ) + ): + self._screenshot("portal_loaded") + return + log.error("等待门户跳转超时") + self._screenshot("portal_timeout") + raise TimeoutError("登录超时，未跳转到信息门户") + + # --------------------------------------------------------------- + # 导航 + # --------------------------------------------------------------- + + def navigate_to_reimburse(self, page_key: str = "reimburse_page") -> None: + """从统一信息平台进入报销系统""" + log.info("进入报销系统...") + self._wait_for('text="快捷入口"', timeout=5000) + + try: + self.page.click('text="财务系统"', timeout=5000) + except Exception: + log.warning("未找到财务系统入口") + + new_tab = None + for _ in range(15): + self.page.wait_for_timeout(1000) + for p in self.context.pages: + if "dddl" in p.url or "210.45.32.214" in p.url: + new_tab = p + break + if new_tab: + break + + if new_tab: + self.page = new_tab + self._wait_for('text="网络报销"', timeout=5000) + else: + log.warning(f"未找到单点登录页面，当前 URL: {self.page.url}") + + for p in self.context.pages[:-1]: + try: + p.close() + except Exception: + pass + + try: + link = self.page.query_selector('a:has(img[src*="wlbx"])') + if link: + reimburse_url = link.get_attribute("href") + self.page.goto(reimburse_url, wait_until="domcontentloaded", timeout=15000) + except Exception: + pass + + self._wait_for('text="报销录入"', timeout=5000) + common_url = self.config["reimburse_url"] + self.config[page_key] + self.page.goto(common_url, wait_until="domcontentloaded", timeout=15000) + self._wait_for('text="单据状态:"', timeout=5000) + + def create_new_form(self) -> None: + """点击「新增」创建新单据""" + log.info("创建新单据...") + self.page.wait_for_timeout(2000) + + try: + self.page.click("#insert", timeout=5000) + except Exception: + try: + self.page.click("text=新增", timeout=3000) + except Exception as err: + self._screenshot("no_add_button") + raise RuntimeError("无法点击新增按钮") from err + + self.page.wait_for_timeout(3000) + self._screenshot("after_add_click") + + # --------------------------------------------------------------- + # 辅助方法 + # --------------------------------------------------------------- + + def _wait_for(self, selector: str, timeout: int | None = None) -> None: + self.page.wait_for_selector(selector, timeout=timeout) + + def _screenshot(self, name: str) -> None: + img_dir = Path(__file__).parent.parent.parent / "images" + img_dir.mkdir(exist_ok=True) + self.page.screenshot(path=str(img_dir / f"debug_{name}.png")) diff --git a/src/bot/normal.py b/src/bot/normal.py new file mode 100644 index 0000000..eea2c0d --- /dev/null +++ b/src/bot/normal.py @@ -0,0 +1,176 @@ +""" +普通报销填报流程 + +负责普通发票报销的完整填报步骤： + 基本信息 → 总明细 → 支付方式 → 附件上传 +""" + +from typing import Any + +from .. import get_logger +from .base import BaseBot, format_date + +log = get_logger("bot.normal") + + +# ------------------------------------------------------------------ +# 普通报销流程 +# ------------------------------------------------------------------ + + +def run( + bot: BaseBot, + normal_info: dict[str, Any], +) -> None: + """执行普通发票报销填报流程 + + Args: + bot: 已启动并登录的 BaseBot 实例。 + normal_info: 普通发票信息字典。 + """ + log.info("开始普通发票报销填报...") + + log.info("填写基本信息...") + description = normal_info.get("basic_info", {}).get("reimbursement_description", "元器件采购报销") + fill_basic_info(bot, description) + + total_invoices = normal_info.get("reimbursement_details", {}).get("total_invoices", 0) + total_amount = normal_info.get("reimbursement_details", {}).get("total_amount", 0) + log.info(f"录入普通发票总明细 (共 {total_invoices} 张, 合计 ¥{total_amount:.2f})...") + add_normal_item(bot, total_invoices, total_amount) + + payment_info = normal_info.get("payment_methods", []) + log.info(f"录入普通发票支付信息 (共 {len(payment_info)} 笔)...") + fill_normal_payment(bot, payment_info) + + log.info("上传普通发票附件...") + attachment_info = normal_info.get("attachments", []) + upload_normal_attachments(bot, attachment_info) + + log.info("普通发票报销填报完成") + + +# ------------------------------------------------------------------ +# 基本信息 +# ------------------------------------------------------------------ + + +def fill_basic_info(bot: BaseBot, description: str = "元器件采购报销") -> None: + """填写基本信息""" + + bot.page.fill("#EXPENEXPLAIN", description) + bot.page.click("#PROJECTCODE", timeout=10000) + bot.page.wait_for_timeout(1000) + + bot.page.wait_for_selector("#promodal .fixed-table-body tbody tr", timeout=10000) + first_row = bot.page.query_selector("#promodal .fixed-table-body tbody tr") + + if first_row: + first_row.click() + bot.page.wait_for_timeout(1000) + + bot.page.click("#saveAndNext", timeout=5000) + bot.page.wait_for_timeout(2000) + + bot._screenshot("step3_done") + + +# ------------------------------------------------------------------ +# 总明细 +# ------------------------------------------------------------------ + + +def add_normal_item(bot: BaseBot, total_invoices: int, total_amount: float) -> None: + """录入普通发票总明细""" + try: + bot.page.click("#insertDetail", timeout=5000) + bot._wait_for('text="经济事项名称"', timeout=5000) + bot.page.click("#economicscode2") + bot.page.wait_for_timeout(1000) + + try: + bot.page.wait_for_selector("#econmodal .fixed-table-body tbody tr", timeout=10000) + rows = bot.page.query_selector_all("#econmodal .fixed-table-body tbody tr") + if len(rows) >= 3: + rows[2].click() + bot.page.wait_for_timeout(1000) + except Exception: + pass + + bot.page.fill('input[name="expenPwCommondetail.HOWBILLS"]', str(total_invoices)) + bot.page.fill("#je_zwzcdz", f"{total_amount:.2f}") + bot.page.click("#detailAdd", timeout=3000) + bot.page.wait_for_timeout(1000) + + bot._screenshot("normal_item_done") + except Exception as e: + log.error(f"录入总明细失败: {e}") + bot._screenshot("normal_item_error") + raise + + +# ------------------------------------------------------------------ +# 支付方式 +# ------------------------------------------------------------------ + + +def fill_normal_payment(bot: BaseBot, payment_info: list[dict[str, Any]]) -> None: + """录入普通发票支付信息""" + try: + bot.page.click('text="下一步(支付方式)"', timeout=5000) + bot._wait_for('text="下一步(附件清单)"', timeout=5000) + + for info in payment_info: + bot.page.click("#insertPay", timeout=5000) + bot.page.wait_for_timeout(1000) + bot.page.fill("#personid2", bot.config["default_name"]) + bot.page.fill("#accountname2", bot.config["default_person_id"]) + bot.page.fill("#receiptdate2", format_date(str(info.get("card_date", "")))) + bot.page.fill("#localaccount2", bot.config["default_card_no"]) + bot.page.fill("#receiptmoney2", str(info.get("card_amount", 0))) + bot.page.fill("#money2", str(info.get("card_amount", 0))) + bot.page.fill("#merchant2", str(info.get("merchant", ""))) + bot.page.fill("#smark2", str(info.get("remark", ""))) + bot.page.click("#payAdd", timeout=3000) + bot.page.wait_for_timeout(1000) + except Exception as e: + log.error(f"支付方式录入失败: {e}") + bot._screenshot("normal_payment_error") + raise + + bot._screenshot("normal_payment_done") + + +# ------------------------------------------------------------------ +# 附件上传 +# ------------------------------------------------------------------ + + +def upload_normal_attachments(bot: BaseBot, attachment_info: list[dict[str, Any]]) -> None: + """上传普通发票附件""" + log.info(f"上传普通发票附件 (共 {len(attachment_info)} 个)...") + try: + bot.page.click("#next3", timeout=5000) + bot._wait_for("#submit2", timeout=5000) + + for info in attachment_info: + attachment_file = bot.work_dir / info["filename"] + bot._wait_for("#insertAcc", timeout=20000) + bot.page.click("#insertAcc", timeout=5000) + bot._wait_for("#fjlx", timeout=5000) + if info.get("attachment_type") == "invoice": + bot.page.select_option("#fjlx", "1") + else: + bot.page.select_option("#fjlx", "2") + bot.page.fill("#fpsmxx", info.get("attachment_desc", "")) + if attachment_file and attachment_file.exists(): + bot.page.set_input_files("#file", str(attachment_file)) + bot.page.wait_for_timeout(1000) + bot.page.click("#cjtj", timeout=5000) + log.info("上传普通发票附件完成") + except Exception as e: + log.error(f"附件上传失败: {e}") + bot._screenshot("normal_attachment_error") + raise + + bot._screenshot("normal_attachment_done") diff --git a/src/bot/travel.py b/src/bot/travel.py new file mode 100644 index 0000000..e11afba --- /dev/null +++ b/src/bot/travel.py @@ -0,0 +1,295 @@ +""" +差旅报销填报流程 + +负责差旅报销的完整填报步骤： + 基本信息 → 差旅明细 → 支付方式 → 补助清单 → 附件上传 +""" + +from typing import Any + +from .. import get_logger +from .base import BaseBot, format_date + +log = get_logger("bot.travel") + + +# ------------------------------------------------------------------ +# 差旅填报流程 +# ------------------------------------------------------------------ + + +def run(bot: BaseBot, travel_info: dict[str, Any]) -> None: + """执行差旅报销填报流程 + + Args: + bot: 已启动并登录的 BaseBot 实例。 + travel_info: 差旅信息字典（包含 basic_info、reimbursement_details 等）。 + """ + log.info("开始差旅报销填报...") + + log.info("填写差旅报销信息...") + basic_info = travel_info["basic_info"] + fill_travel_info(bot, basic_info) + + log.info("填写差旅报销明细...") + travel_items = travel_info["reimbursement_details"] + add_travel_items(bot, travel_items) + + log.info("填写差旅报销支付方式...") + payment_info = travel_info["payment_methods"] + fill_travel_payment(bot, payment_info) + + log.info("填写差旅报销补助清单...") + subsidy_info = travel_info["subsidy_list"] + fill_travel_subsidy(bot, subsidy_info) + + log.info("上传差旅报销附件...") + attachment_info = travel_info["attachments"] + upload_travel_attachments(bot, attachment_info) + + log.info("差旅报销填报完成") + + +# ------------------------------------------------------------------ +# 基本信息 +# ------------------------------------------------------------------ + + +def fill_travel_info(bot: BaseBot, basic_info: dict[str, Any]) -> None: + """填写差旅报销基本信息""" + try: + bot.page.fill("#CAUSE", basic_info.get("travel_purpose", "")) + bot.page.fill("#SITE", basic_info.get("travel_location", "")) + + bot.page.click("#PROJECTCODE", timeout=10000) + bot.page.wait_for_timeout(1000) + bot.page.wait_for_selector("#promodal .fixed-table-body tbody tr", timeout=10000) + first_row = bot.page.query_selector("#promodal .fixed-table-body tbody tr") + if first_row: + first_row.click() + bot.page.wait_for_timeout(1000) + + bot.page.fill("#THEKSRQ", format_date(basic_info.get("start_date", ""))) + bot.page.fill("#THEJSRQ", format_date(basic_info.get("end_date", ""))) + bot.page.click("#saveAndNext", timeout=5000) + bot.page.wait_for_timeout(2000) + bot._screenshot("travel_basic_done") + except Exception: + log.error("填写基本信息失败") + bot._screenshot("travel_basic_error") + + +# ------------------------------------------------------------------ +# 差旅明细 +# ------------------------------------------------------------------ + + +def add_travel_items(bot: BaseBot, travel_items: dict[str, Any]) -> None: + """录入差旅报销明细""" + vehicle_map = { + "火车": "01", + "汽车": "02", + "轮船": "03", + "自带车": "04", + "公务车": "05", + "飞机": "06", + "租车": "07", + "自驾车": "08", + } + + try: + traffic_info = travel_items.get("transport_fee") or [] + for item in traffic_info: + bot.page.click("#insertDetail", timeout=5000) + bot._wait_for('text="增加明细"', timeout=5000) + bot.page.select_option("#cost", "1") + bot.page.wait_for_timeout(500) + vehicle = item.get("vehicle_type", "") + if vehicle in vehicle_map: + bot.page.select_option("#jtgj", vehicle_map[vehicle]) + + bot.page.fill("#ksdd", item.get("departure_place", "")) + bot.page.fill("#jsdd", item.get("arrival_place", "")) + bot.page.fill( + '#t1 input[name="expenPwTraveldetail.MONEY"]', + str(item.get("amount", "")), + ) + bot.page.fill( + '#t1 input[name="expenPwTraveldetail.HOWBILL"]', + str(item.get("bill_count", "")), + ) + bot.page.fill( + '#t1 input[name="expenPwTraveldetail.SMARK"]', + str(item.get("remark", "")), + ) + bot.page.click("#detailAdd", timeout=3000) + bot.page.wait_for_timeout(1000) + + hotel_info = travel_items.get("hotel_fee") or [] + for item in hotel_info: + bot.page.click("#insertDetail", timeout=5000) + bot._wait_for('text="增加明细"', timeout=5000) + bot.page.select_option("#cost", "2") + bot.page.wait_for_timeout(500) + bot.page.fill("#ksrq2", format_date(str(item.get("checkin_date", "")))) + bot.page.fill("#jsrq2", format_date(str(item.get("checkout_date", "")))) + bot.page.fill("#ts2", str(item.get("days", ""))) + bot.page.fill("#rs2", str(item.get("person_count", ""))) + bot.page.fill( + '#t2 input[name="expenPwTraveldetail.FPMONEY"]', + str(item.get("invoice_amount", "")), + ) + bot.page.fill( + '#t2 input[name="expenPwTraveldetail.MONEY"]', + str(item.get("reimburse_amount", "")), + ) + bot.page.fill( + '#t2 input[name="expenPwTraveldetail.SMARK"]', + str(item.get("remark", "")), + ) + bot.page.click("#detailAdd", timeout=3000) + bot.page.wait_for_timeout(1000) + + conference_info = travel_items.get("conference_fee") or [] + for item in conference_info: + bot.page.click("#insertDetail", timeout=5000) + bot._wait_for('text="增加明细"', timeout=5000) + bot.page.select_option("#cost", "3") + bot.page.wait_for_timeout(500) + bot.page.fill( + '#t3 input[name="expenPwTraveldetail.HOWBILL"]', + str(item.get("bill_count", "")), + ) + bot.page.fill( + '#t3 input[name="expenPwTraveldetail.MONEY"]', + str(item.get("amount", "")), + ) + bot.page.fill( + '#t3 input[name="expenPwTraveldetail.SMARK"]', + str(item.get("remark", "")), + ) + bot.page.click("#detailAdd", timeout=3000) + bot.page.wait_for_timeout(1000) + except Exception as e: + log.error(f"录入总明细失败: {e}") + bot._screenshot("item_total_error") + raise + + +# ------------------------------------------------------------------ +# 支付方式 +# ------------------------------------------------------------------ + + +def fill_travel_payment(bot: BaseBot, payment_info: list[dict[str, Any]]) -> None: + """录入差旅支付信息""" + try: + bot.page.click('text="下一步(支付方式)"', timeout=5000) + bot._wait_for('text="下一步(补助清单)"', timeout=5000) + + for info in payment_info: + bot.page.click("#insertPay", timeout=5000) + bot.page.wait_for_timeout(1000) + bot.page.fill("#personid2", bot.config["default_name"]) + bot.page.fill("#accountname2", bot.config["default_person_id"]) + bot.page.fill("#receiptdate2", format_date(info["card_date"])) + bot.page.fill("#localaccount2", bot.config["default_card_no"]) + bot.page.fill("#receiptmoney2", str(info["card_amount"])) + bot.page.fill("#money2", str(info["card_amount"])) + bot.page.fill("#merchant2", info.get("merchant", "")) + bot.page.fill("#smark2", info.get("remark", "")) + bot.page.click("#payAdd", timeout=3000) + bot.page.wait_for_timeout(1000) + except Exception as e: + log.error(f"支付方式录入失败: {e}") + bot._screenshot("step5_error") + raise + + bot._screenshot("step5_done") + + +# ------------------------------------------------------------------ +# 补助清单 +# ------------------------------------------------------------------ + + +def fill_travel_subsidy(bot: BaseBot, subsidy_info: list[dict[str, Any]]) -> None: + """录入差旅补助清单""" + try: + bot.page.click("#next3", timeout=5000) + bot._wait_for("#next4", timeout=5000) + + for info in subsidy_info: + bot.page.click("#insertSubsidy", timeout=5000) + bot._wait_for('text="增加补助清单"', timeout=5000) + bot.page.click("#jzg3", timeout=5000) + bot.page.wait_for_timeout(500) + if info["person_name"] and info["person_name"] != "": + bot.page.fill("#seacher", info["person_name"]) + elif info["person_id"] and info["person_id"] != "": + bot.page.fill("#seacher", info["person_id"]) + else: + raise ValueError(f"人员编号和人员姓名不能同时为空: {info}") + bot.page.click("#cx", timeout=5000) + bot.page.wait_for_selector("div.fixed-table-loading", state="hidden", timeout=10000) + bot.page.click("#tableEmp tbody tr", timeout=10000) + bot.page.wait_for_timeout(1000) + + open_bank = bot.page.input_value("#openbank1") + if not open_bank: + log.info("员工开户行未填写，默认填写中国工商银行") + bot.page.fill("#openbank1", "中国工商银行") + + bot.page.fill("#startdate1", format_date(info["start_date"])) + bot.page.fill("#enddate1", format_date(info["end_date"])) + bot.page.fill("#trafficdays1", str(info["days"])) + bot.page.fill("#fooddays1", str(info["days"])) + bot.page.fill("#trafficnorm1", str(80)) + bot.page.fill("#foodnorm1", str(100)) + trafficmoney = int(info["days"]) * 80 + foodmoney = int(info["days"]) * 100 + subsidymoney = trafficmoney + foodmoney + bot.page.fill("#trafficmoney1", str(trafficmoney)) + bot.page.fill("#foodmoney1", str(foodmoney)) + bot.page.fill("#subsidymoney1", str(subsidymoney)) + bot.page.click("#add", timeout=3000) + bot.page.wait_for_timeout(1000) + except Exception as e: + log.error(f"差旅补助清单录入失败: {e}") + bot._screenshot("subsidy_error") + raise + + bot._screenshot("subsidy_done") + + +# ------------------------------------------------------------------ +# 附件上传 +# ------------------------------------------------------------------ + + +def upload_travel_attachments(bot: BaseBot, attachment_info: list[dict[str, Any]]) -> None: + """上传差旅附件""" + try: + bot.page.click("#next4", timeout=5000) + bot._wait_for("#submit2", timeout=5000) + + for info in attachment_info: + attachment_file = bot.work_dir / info["filename"] + bot._wait_for("#insertAcc", timeout=20000) + bot.page.click("#insertAcc", timeout=5000) + bot._wait_for("#fjlx", timeout=5000) + if info["attachment_type"] == "invoice": + bot.page.select_option("#fjlx", "1") + else: + bot.page.select_option("#fjlx", "2") + bot.page.fill("#fpsmxx", info["attachment_desc"]) + if attachment_file and attachment_file.exists(): + bot.page.set_input_files("#file", str(attachment_file)) + bot.page.wait_for_timeout(1000) + bot.page.click("#cjtj", timeout=5000) + log.info("上传差旅附件完成") + except Exception as e: + log.error(f"差旅附件上传失败: {e}") + bot._screenshot("travel_attachment_error") + raise + bot._screenshot("travel_attachment_done") diff --git a/src/doc/README.md b/src/doc/README.md index 0c05efe..440be1a 100644 --- a/src/doc/README.md +++ b/src/doc/README.md @@ -1,10 +1,10 @@ --- -last_reviewed: 2026-06-09 +last_reviewed: 2026-06-11 --- # src/doc — 文档处理模块 -负责发票信息提取、基于 LLM 的支付截图信息识别、以及将数据填入 Word 出库单模板。 +负责发票信息提取、基于 LLM 的支付截图信息识别、差旅/普通报销信息提取、以及将数据填入 Word 出库单模板。 ## 模块清单 @@ -13,12 +13,12 @@ last_reviewed: 2026-06-09 | ------------------------ | -------------------------------------------------- | | `extractor.py` | 编排入口：串联 PDF 读取 → LLM 提取 → 支付截图匹配 → 分类 | | `pdf.py` | PDF 图片渲染（PyMuPDF） | -| `llm_extractor.py` | 基于 LLM 的信息提取（发票文本 + 支付截图多模态） | +| `llm_extractor.py` | 基于 LLM 的信息提取（发票文本 + 支付截图多模态 + 差旅/普通报销信息综合提取） | | `matcher.py` | 发票与支付截图按金额匹配，回填刷卡信息至发票记录 | | `invoice.py` | 发票类型常量、分类逻辑、CSV 读写工具 | | `fill_consumable_doc.py` | 将 CSV 数据填入易耗品出库单 Word 模板（pywin32 COM） | | `prompt.py` | LLM 提示词模板加载 | -| `prompts/` | 提示词模板文件（`invoice_system.md`、`travel_info_system.md`） | +| `prompts/` | 提示词模板文件（`invoice_system.md`、`travel_info_system.md`、`normal_info_system.md`） | ## 数据流 @@ -32,6 +32,11 @@ PDF 发票 → pdf.py → llm_extractor.py → [发票列表] invoice.py 分类 → CSV（已回填刷卡日期/卡号/金额） ↓ fill_consumable_doc → 易耗品出库单.doc + +[发票列表 + 匹配结果] → llm_extractor.py + ↓ + 差旅发票 → extract_travel_info() → travel_info.json + 普通发票 → extract_normal_info() → normal_info.json ``` ## 依赖说明 @@ -45,4 +50,11 @@ PDF 发票 → pdf.py → llm_extractor.py → [发票列表] - `fill_consumable_doc.py` 依赖 Microsoft Word + COM，仅 Windows 可用 - LLM 提取不会覆盖 CSV 中已有非空字段 - 提示词模板位于 `prompts/` 目录，由 `prompt.py` 加载 -- LLM 提取失败时直接报错，无正则回退 \ No newline at end of file +- LLM 提取失败时直接报错，无正则回退 + +## 变更说明（2026-06-11） + +- `llm_extractor.py` 新增 `extract_normal_info()`：综合普通发票、支付记录和匹配结果，提取报销说明、发票总数、总金额、支付方式、附件清单，缓存为 `normal_info.json` +- `llm_extractor.py` 的 `load_cache()` 扩展支持加载 `normal_info.json` +- `prompt.py` 新增 `build_normal_info_system_prompt()`：加载 `normal_info_system.md` +- `prompts/` 新增 `normal_info_system.md`：普通发票信息提取的系统提示词 \ No newline at end of file diff --git a/src/doc/extractor.py b/src/doc/extractor.py index 8c9b99b..44da31a 100644 --- a/src/doc/extractor.py +++ b/src/doc/extractor.py @@ -51,16 +51,17 @@ def _get_cache_dir(source_dir: Path) -> Path: def _get_json_path(file_path: Path, cache_dir: Path) -> Path: - """根据文件路径生成对应的 JSON 缓存路径""" - return cache_dir / f"{file_path.stem}.json" + """根据文件路径生成对应的 JSON 缓存路径（包含后缀名以区分同名的 PDF/图片）""" + return cache_dir / f"{file_path.stem}{file_path.suffix}.json" def _save_to_cache(file_path: Path, extracted_data: dict[str, Any], cache_dir: Path) -> Path: - """将提取结果保存到 JSON 缓存文件，并记录源文件路径""" + """将提取结果保存到 JSON 缓存文件，并记录源文件路径和后缀名""" json_path = _get_json_path(file_path, cache_dir) cache_data = { "source_file": str(file_path), "source_filename": file_path.name, + "source_extension": file_path.suffix.lower(), "extracted_data": extracted_data, } with open(json_path, "w", encoding="utf-8") as f: @@ -69,13 +70,16 @@ def _save_to_cache(file_path: Path, extracted_data: dict[str, Any], cache_dir: P return json_path -def _load_from_cache(json_path: Path) -> dict[str, Any] | None: - """从 JSON 缓存文件加载提取结果""" +def _load_from_cache(json_path: Path, expected_extension: str | None = None) -> dict[str, Any] | None: + """从 JSON 缓存文件加载提取结果，可选校验后缀名一致性""" if not json_path.exists(): return None try: with open(json_path, encoding="utf-8") as f: cache_data: dict[str, Any] = json.load(f) + # 校验后缀名是否一致，防止同名不同后缀的文件误命中缓存 + if expected_extension and cache_data.get("source_extension", "").lower() != expected_extension.lower(): + return None result: dict[str, Any] | None = cache_data.get("extracted_data") return result except Exception as e: @@ -100,7 +104,7 @@ def _extract_document(file_path: Path, cache_dir: Path) -> dict[str, str] | None 提取结果字典，失败时返回 None。 """ json_path = _get_json_path(file_path, cache_dir) - cached = _load_from_cache(json_path) + cached = _load_from_cache(json_path, expected_extension=file_path.suffix.lower()) if cached: cached["_source_file"] = file_path.name log.info(f"使用缓存: {file_path.name}") diff --git a/src/doc/llm_extractor.py b/src/doc/llm_extractor.py index 4d871c4..fd89f79 100644 --- a/src/doc/llm_extractor.py +++ b/src/doc/llm_extractor.py @@ -26,6 +26,7 @@ from typing import Any, cast from .. import get_logger from .prompt import ( build_invoice_system_prompt, + build_normal_info_system_prompt, build_travel_info_system_prompt, ) @@ -218,9 +219,9 @@ def load_cache(source_dir: Path) -> dict[str, Any]: with open(json_path, encoding="utf-8") as f: cache_data = json.load(f) - # travel_info.json 结构不同，直接存储 - if json_path.name == "travel_info.json": - cache_map["travel_info"] = cache_data + # travel_info.json / normal_info.json 结构不同，直接存储 + if json_path.name in ("travel_info.json", "normal_info.json"): + cache_map[json_path.name.replace(".json", "")] = cache_data continue extracted = cache_data.get("extracted_data", {}) @@ -324,3 +325,71 @@ def extract_travel_info( except Exception as e: log.error("LLM 差旅信息提取失败: %s", e) raise + + +# ------------------------------------------------------------------ +# 普通发票信息提取 +# ------------------------------------------------------------------ + + +def extract_normal_info( + source_dir: Path | None = None, +) -> dict[str, Any]: + """根据普通发票（非差旅），让 LLM 提取报销相关信息。 + + 仅支持从 JSON 缓存加载数据。 + + Args: + source_dir: 源文件目录（必填，包含 .invoice_cache 子目录）。 + + Returns: + 包含报销说明、发票总数、总金额、支付方式、附件清单等字段的字典。 + """ + if not source_dir: + log.warning("未提供 source_dir，无法加载缓存数据") + return {} + + system_prompt = build_normal_info_system_prompt() + + # 构建 source filename -> 缓存数据的映射 + cache_map = load_cache(source_dir) + + # 加载发票与支付记录的匹配结果 + match_result = load_match_result(source_dir) + + # 拼接纯文本消息 + parts = [ + "以下是本次报销的所有源文件及其提取出的结构化数据。" + "每个源文件的数据来自 OCR 识别和发票信息提取，已按文件名分组展示。" + ] + + # 如果有匹配结果，作为额外上下文提供 + if match_result: + parts.append( + "【发票与支付记录匹配结果】" + "以下数据已将发票信息与对应的支付记录进行关联匹配，" + "用于判断每笔支付对应的发票和商户信息。\n" + json.dumps(match_result, ensure_ascii=False, indent=2) + ) + + # 按源文件名提供结构化数据 + for filename, extracted in cache_map.items(): + parts.append( + f"【源文件: {filename}】" + "以下为从该文件提取的结构化发票/支付/申请单数据。\n" + json.dumps(extracted, ensure_ascii=False, indent=2) + ) + + parts.append("\n=== 请返回 JSON 格式结果 ===") + user_message = "\n".join(parts) + log.info(f"user_message: {user_message}") + try: + response = _llm_query_multimodal( + system_prompt=system_prompt, + text=user_message, + reasoning_effort="low", + ) + result = _parse_json_response(response) + log.info("LLM 普通发票信息提取成功") + return result + except Exception as e: + log.error("LLM 普通发票信息提取失败: %s", e) + raise diff --git a/src/doc/matcher.py b/src/doc/matcher.py index fbe1c25..084ed0d 100644 --- a/src/doc/matcher.py +++ b/src/doc/matcher.py @@ -18,12 +18,13 @@ 2. 解析发票和刷卡记录的金额，进行总额校验 - 发票总额 < 刷卡总额时发出 warning 3. 按金额降序排序 -4. 根据数量关系选择匹配策略： +4. 文件名匹配（最高优先级）：发票和刷卡记录的文件名（不含后缀）一致时直接匹配 +5. 根据数量关系选择匹配策略： - 数量相等 → 一对一匹配：按金额从大到小依次配对，相对容差内即匹配 - 发票更多 → 一对多匹配：对每张刷卡记录贪心凑金额，相对容差内结束 -5. 构建以支付记录为主键的结果列表 -6. 未匹配的发票单独作为一条记录（无刷卡信息） -7. 清理内部字段，输出支付记录列表 +6. 构建以支付记录为主键的结果列表 +7. 未匹配的发票单独作为一条记录（无刷卡信息） +8. 清理内部字段，输出支付记录列表 ## 容差计算 @@ -45,6 +46,7 @@ match_invoices_to_cards(invoices, cards, tolerance) -> list[dict] """ +from pathlib import Path from typing import Any from .. import get_logger @@ -170,10 +172,18 @@ def _match( """执行匹配，返回 {card_index: [invoice_indices]} 的映射 tolerance 为相对容差比例（如 0.03 表示 3%） + + 匹配优先级（从高到低）： + 1. 文件名匹配：发票和刷卡记录的文件名（不含后缀）一致时直接匹配 + 2. 精确匹配：金额差 <= 0.01 元 + 3. 一对一 / 一对多贪心匹配：按金额容差匹配 """ result: dict[int, list[int]] = {} assigned: set[int] = set() + # ---- 阶段 0：文件名匹配（最高优先级）---- + _match_by_filename(invoices, cards, assigned, result) + if len(invoices) == len(cards): _match_one_to_one(invoices, cards, tolerance, assigned, result) else: @@ -182,6 +192,34 @@ def _match( return result +def _match_by_filename( + invoices: list[dict[str, Any]], + cards: list[dict[str, Any]], + assigned: set[int], + result: dict[int, list[int]], +) -> None: + """文件名匹配：发票和刷卡记录的文件名（不含后缀）一致时直接匹配""" + for card_idx, card in enumerate(cards): + card_name = card.get("_source_file", "") + if not card_name: + continue + card_stem = Path(card_name).stem + + for idx, inv in enumerate(invoices): + if idx in assigned: + continue + inv_name = inv.get("_source_file", "") + if not inv_name: + continue + inv_stem = Path(inv_name).stem + + if inv_stem == card_stem: + assigned.add(idx) + result[card_idx] = [idx] + log.info(f"[文件名匹配] {inv_name} ↔ {card_name}") + break + + def _match_one_to_one( invoices: list[dict[str, Any]], cards: list[dict[str, Any]], diff --git a/src/doc/prompt.py b/src/doc/prompt.py index 21d3dc0..e34b509 100644 --- a/src/doc/prompt.py +++ b/src/doc/prompt.py @@ -24,3 +24,8 @@ def build_invoice_system_prompt() -> str: def build_travel_info_system_prompt() -> str: """构建差旅信息提取系统提示词。""" return _load_prompt("travel_info_system.md") + + +def build_normal_info_system_prompt() -> str: + """构建普通发票信息提取系统提示词。""" + return _load_prompt("normal_info_system.md") diff --git a/src/doc/prompts/invoice_system.md b/src/doc/prompts/invoice_system.md index 742161a..f36a1b1 100644 --- a/src/doc/prompts/invoice_system.md +++ b/src/doc/prompts/invoice_system.md @@ -1,4 +1,4 @@ -你是财务文档信息提取助手。你的任务是从图片中提取结构化信息，可能是支付截图、银行转账记录、微信/支付宝付款凭证等，也可能是发票文件，也可能是出差事前申请单，不管任何形式都要用统一的 JSON 格式返回信息。 +你是财务文档信息提取助手。你的任务是从图片中提取结构化信息，可能是支付截图、银行转账记录、微信/支付宝付款凭证等，也可能是发票文件，也可能是出差事前申请单，也可能是易耗品、出库单，不管任何形式都要用统一的 JSON 格式返回信息。 第一步要先判断是，支付记录、高铁票、酒店住宿，普通发票，然后不同类型输出的信息不同。 @@ -36,6 +36,11 @@ "invoice_date": "2026-06-03", "total_amount": "536.00"//必填项 } +以下是易耗品出入库单的完整示例： +```json +{ + "invoice_type": "note",//必填项且只有这一项 +} ``` ## 重要规则 - **必填项**：invoice_type、invoice_date、ride_date、departure、arrival、person_name、total_amount 字段为必填，必须填写。 @@ -85,4 +90,7 @@ 6. total_amount: 金额数字 7. seller_name: 卖方全称 +如果是易耗品出入库单，返回如下字段： +1. invoice_type: "note" + **千万注意！千万注意！**：严格只输出 JSON，不要输出任何其他文字、Markdown 标记或解释。 \ No newline at end of file diff --git a/src/doc/prompts/normal_info_system.md b/src/doc/prompts/normal_info_system.md new file mode 100644 index 0000000..c9d6398 --- /dev/null +++ b/src/doc/prompts/normal_info_system.md @@ -0,0 +1,78 @@ +# 普通报销信息提取系统提示词 + +你是财务报销信息提取助手。根据发票信息和付款记录，提取报销相关的结构化数据，并以严格符合类型要求的 JSON 格式返回。 + +> **核心原则**：类型约束为最高优先级规则，任何情况下不得违反。 + +## 强制类型约束 + +以下类型规则为最高优先级，任何情况下不得违反。 + +### 1. 根节点字段（共 4 个，类型不可变更） + +| 字段名 | 强制类型 | 空值处理 | +| --- | --- | --- | +| `basic_info` | 对象 (dict) | 必填，所有子字段必须完整存在 | +| `reimbursement_details` | 对象 (dict) | 必填，必须且仅包含下述 2 个子字段 | +| `payment_methods` | 数组 (list) | 必填，无数据时赋值为 `[]` | +| `attachments` | 数组 (list) | 必填，无数据时赋值为 `[]` | + +### 2. `reimbursement_details` 子字段（共 2 个） + +| 字段名 | 强制类型 | 空值处理 | +| --- | --- | --- | +| `total_invoices` | 数字 (int) | 必填 | +| `total_amount` | 数字 (float) | 必填 | + +### 3. 禁止行为 + +* 省略任何根节点字段或 `reimbursement_details` 的子字段 +* 将 `payment_methods` 或 `attachments` 赋值为 `null`、字符串、数字或对象 +* 在 `reimbursement_details` 中添加未定义的子字段 +* 合并不同模块的数组数据 + +**正确示例**： +```json +{ + "basic_info": {...}, + "reimbursement_details": {...}, + "payment_methods": [], + "attachments": [{"filename":"发票.pdf", ...}] +} +``` + +**错误示例**： +```json +{ + "basic_info": {...}, + "reimbursement_details": {...}, + "payment_methods": null, +} +``` + +## 输入数据说明 +你会收到以下数据： +1. **发票信息**：包含购买物品的发票信息 +2. **付款记录**：包含刷卡日期、刷卡金额、公务卡号等信息 + +需要提取的信息： +1. `basic_info`:(必填，每一项都必须填，给出合理的猜测) + 1. `reimbursement_description`：根据所有信息写一句20字以内的报销说明 +2. `reimbursement_details`:(至少有一项) + 1. `total_invoices`: 发票的总份数 + 2. `total_amount`：填写付款记录总金额 +3. `payment_methods`：（多少笔支付记录就有多少条；多项请采用上述通用 JSON 数组格式） + 1. `card_date`：根据付款记录，格式 YYYY-M-D + 2. `card_amount`：根据付款记录填写，单位为元，数字 + 3. `merchant`：根据发票信息推测商户信息 + 4. `remark`：说明该笔付款关联的发票信息 +4. `attachments`：（必填，用户已经告诉你所有文件了`【源文件: {filename}】`，"invoice_type": "payment"的不作为附件） + 1. `filename`: 严格使用用户提供的原始文件名，不得修改任何字符 + 2. `attachment_type`：从以下两个选项中选择：invoice、other + 3. `attachment_desc`：简要描述该文件的基本信息 + +## 最终输出要求 + +* 仅输出纯 JSON 字符串，不包含任何思考过程、解释文字或 Markdown 标记 +* JSON 语法必须正确，无多余逗号、引号等错误 +* 严格遵守所有强制性类型约束，违反类型要求的输出视为无效 \ No newline at end of file diff --git a/src/doc/prompts/travel_info_system.md b/src/doc/prompts/travel_info_system.md index 2fd40b0..3d52290 100644 --- a/src/doc/prompts/travel_info_system.md +++ b/src/doc/prompts/travel_info_system.md @@ -3,7 +3,7 @@ 你是财务差旅信息提取助手。你的任务是根据发票信息、付款记录，提取出差相关的结构化信息，并以严格符合以下类型要求的 JSON 格式返回。所有类型约束为最高优先级规则，任何情况下不得违反。 🔴 最高优先级：强制性类型约束（优先级高于所有其他规则） -1. 根节点必须包含且仅包含以下 6 个字段，字段类型绝对不可变更： +1. 根节点必须包含且仅包含以下 5 个字段，字段类型绝对不可变更： | 字段名 | 强制类型 | 空值处理规则 | | ------ | --------- | ------------------ | diff --git a/src/pipeline.py b/src/pipeline.py index b5269b0..900f0a8 100644 --- a/src/pipeline.py +++ b/src/pipeline.py @@ -1,16 +1,22 @@ """ 报销全流程编排 -将发票提取 → 浏览器填报串联为一条管道， +将发票提取 → 差旅/普通信息提取 → 浏览器填报串联为一条管道， 数据在内存中流转，同时生成 CSV 中间产物。 发票类型区分： - 差旅发票（train/hotel）：不生成易耗品出库单，走差旅报销流程 - 普通发票（general）：生成易耗品出库单，走普通报销流程 + +数据流变更（2026-06-11）： + 在发票提取和匹配完成后立即判断报销类型（差旅/普通）， + 差旅调用 LLM 提取 travel_info.json，普通预留 normal_info.json。 + Bot 仅负责接收信息并填报，不再承担信息提取职责。 """ +import json from pathlib import Path -from typing import Any +from typing import Any, cast from . import get_logger from .config import load_config @@ -22,6 +28,12 @@ from .doc.invoice import ( from .doc.invoice import ( save_csv as save_payment_csv, ) +from .doc.llm_extractor import ( + CACHE_DIR_NAME, + extract_normal_info, + extract_travel_info, + load_cache, +) def _classify_invoice_batch( @@ -54,6 +66,64 @@ def _classify_from_cache(cache_path: Path) -> dict[str, list[dict[str, Any]]]: return _classify_invoice_batch(invoices) +def _extract_travel_info_if_needed(groups: dict[str, list[dict[str, Any]]], cache_path: Path) -> dict[str, Any] | None: + """当存在差旅发票时，调用 LLM 提取差旅信息并缓存。 + + Returns: + 差旅信息字典，非差旅时返回 None。 + """ + if not groups.get("travel"): + return None + + from .doc.llm_extractor import CACHE_DIR_NAME, load_cache + + # 检查缓存是否已有 + cache_map = load_cache(cache_path) + if cache_map.get("travel_info"): + log.info("使用已有差旅信息缓存") + return cast(dict[str, Any] | None, cache_map["travel_info"]) + + log.info("开始提取差旅信息...") + travel_info = extract_travel_info(source_dir=cache_path) + + # 保存到缓存 + cache_dir = cache_path / CACHE_DIR_NAME + cache_dir.mkdir(parents=True, exist_ok=True) + + with open(cache_dir / "travel_info.json", "w", encoding="utf-8") as f: + json.dump(travel_info, f, ensure_ascii=False, indent=2) + log.info("差旅信息已保存到缓存") + return travel_info + + +def _extract_normal_info_if_needed(groups: dict[str, list[dict[str, Any]]], cache_path: Path) -> dict[str, Any] | None: + """当存在普通发票时，调用 LLM 提取普通报销信息并缓存。 + + Returns: + 普通报销信息字典，非普通时返回 None。 + """ + if not groups.get("general"): + return None + + # 检查缓存是否已有 + cache_map = load_cache(cache_path) + if cache_map.get("normal_info"): + log.info("使用已有普通发票信息缓存") + return cast(dict[str, Any] | None, cache_map["normal_info"]) + + log.info("开始提取普通发票信息...") + normal_info = extract_normal_info(source_dir=cache_path) + + # 保存到缓存 + cache_dir = cache_path / CACHE_DIR_NAME + cache_dir.mkdir(parents=True, exist_ok=True) + + with open(cache_dir / "normal_info.json", "w", encoding="utf-8") as f: + json.dump(normal_info, f, ensure_ascii=False, indent=2) + log.info("普通发票信息已保存到缓存") + return normal_info + + def run_pipeline( step: str = "all", username: str | None = None, @@ -78,10 +148,12 @@ def run_pipeline( cache_path = Path(cache_dir) if cache_dir else project_dir / "scripts" / "data" # -------------------------------------------------- - # Step 1: 发票提取 + # Step 1: 发票提取 + 类型判断 + 信息提取 # -------------------------------------------------- payment_records: list[dict[str, str]] | None = None groups: dict[str, list[dict[str, str]]] | None = None + travel_info: dict[str, Any] | None = None + normal_info: dict[str, Any] | None = None if step in ("all", "invoice"): log.info("=" * 60) @@ -101,6 +173,13 @@ def run_pipeline( log.info(f"发票分类: 差旅 {len(groups['travel'])} 张, 普通 {len(groups['general'])} 张") + # 发票提取完成后立即判断类型 + is_travel = bool(groups["travel"]) and not bool(groups["general"]) + if is_travel: + travel_info = _extract_travel_info_if_needed(groups, cache_path) + else: + normal_info = _extract_normal_info_if_needed(groups, cache_path) + if step == "invoice": log.info("[1/2] 发票提取 完成") return 0 @@ -118,12 +197,17 @@ def run_pipeline( if groups is None: groups = _classify_from_cache(cache_path) - if groups["travel"] and not groups["general"]: + is_travel = bool(groups["travel"]) and not bool(groups["general"]) + if is_travel: + if travel_info is None: + travel_info = _extract_travel_info_if_needed(groups, cache_path) log.info("检测到纯差旅发票，使用差旅报销模式") - run_bot(config, work_dir=cache_path) + run_bot(config, work_dir=cache_path, travel_info=travel_info) else: + if normal_info is None: + normal_info = _extract_normal_info_if_needed(groups, cache_path) log.info("检测到普通发票，使用普通报销模式") - run_bot(config, work_dir=cache_path) + run_bot(config, work_dir=cache_path, normal_info=normal_info) if step == "submit": log.info("[2/2] 报销提交 完成") diff --git a/src/web/README.md b/src/web/README.md index afbf7bf..5c9913d 100644 --- a/src/web/README.md +++ b/src/web/README.md @@ -1,5 +1,5 @@ --- -last_reviewed: 2026-06-09 +last_reviewed: 2026-06-11 --- # src/web 模块设计说明 @@ -29,18 +29,32 @@ src/web/ ## 数据流 -``` -用户上传文件 → 创建 session → 文件写入 uploads// - ↓ -后台线程执行管道: extract_invoices() → enrich_with_llm() → save_csv() - ↓ -结果写入 session 目录: invoice_summary.csv / result.json / session.log - ↓ -前端 SSE 轮询 result.json 变化 → 显示完成状态 - ↓ -前端加载 CSV 数据 → 可编辑表格展示 → 用户修改后保存 - ↓ -用户点击提交 → run_financial_submit() → bot 自动填报财务系统 +```mermaid +graph TD + A[用户上传文件] --> B[创建 session] + B --> C[文件写入 uploads//] + C --> D[后台线程执行管道] + D --> E["extract_invoices()"] + E --> F["enrich_with_llm()"] + F --> G["save_csv()"] + G --> H["结果写入 session 目录"] + H --> I["invoice_summary.csv"] + H --> J["invoice_groups.json"] + H --> K["result.json"] + H --> L["session.log"] + E --> M{"判断报销类型"} + M -->|差旅| N["extract_travel_info()"] + M -->|普通| O["extract_normal_info()"] + N --> P["travel_info.json"] + O --> Q["normal_info.json"] + K --> R["前端 SSE 轮询 → 显示完成状态"] + R --> S["前端加载 CSV → 可编辑表格"] + S --> T["用户修改后保存"] + T --> U["用户点击提交"] + U --> V["run_financial_submit()"] + V --> W["bot 自动填报财务系统"] + W --> X["差旅模式: travel_info.json → 填报差旅单 → 上传差旅附件"] + W --> Y["普通模式: normal_info.json → 基本信息 → 录入明细 → 支付信息 → 上传附件"] ``` ## API 路由 @@ -70,7 +84,16 @@ src/web/ - **差旅发票**（高铁票/酒店住宿）：不生成易耗品出库单，走差旅报销流程 - **普通发票**：生成易耗品出库单（Word 文档），走普通报销流程 -`classify_invoice_batch()` 根据发票内容自动分类，`_try_fill_consumable_doc()` 仅对普通发票生成出库单。 +`extract_invoices()` 在提取阶段完成分类，结果保存为 `invoice_groups.json`（包含 `travel_count`、`general_count`、`application_count`）。后续步骤（出库单生成、财务填报）统一从该文件读取分类结果，避免重复解析 CSV 和 JSON 字段。 + +### LLM 信息提取（2026-06-11） + +发票提取和匹配完成后，根据类型分流调用 LLM 提取结构化报销信息： + +- **差旅发票**：调用 `extract_travel_info()`，提取出差事由、地点、时间、交通/住宿明细、补助清单、支付方式、附件清单，缓存为 `travel_info.json`。 +- **普通发票**：调用 `extract_normal_info()`，提取报销说明、发票总数、总金额、支付方式、附件清单，缓存为 `normal_info.json`。 + +Bot 填报时优先使用 LLM 提取的信息（`travel_info`/`normal_info`），降级时从缓存加载原始发票数据。 ### 移动端同步 diff --git a/src/web/app.py b/src/web/app.py index 803d941..6d2c35b 100644 --- a/src/web/app.py +++ b/src/web/app.py @@ -19,7 +19,7 @@ import threading import time import uuid from pathlib import Path -from typing import Any +from typing import Any, cast from urllib.parse import quote from flask import Flask, Response, jsonify, render_template, request, stream_with_context @@ -46,24 +46,6 @@ from src.doc.invoice import ( # noqa: E402, I001 ) -def _classify_invoice_batch( - invoices: list[dict[str, str]], -) -> dict[str, list[dict[str, str]]]: - """按发票类型分组""" - travel: list[dict[str, str]] = [] - general: list[dict[str, str]] = [] - application: list[dict[str, str]] = [] - for inv in invoices: - inv_type = inv.get("invoice_type", "general") - if inv_type == "application": - application.append(inv) - elif inv_type in ("train", "hotel"): - travel.append(inv) - else: - general.append(inv) - return {"travel": travel, "general": general, "application": application} - - fill_log = get_logger("fill_consumable_doc") CONSUMABLE_TEMPLATE = PROJECT_ROOT / CONSUMABLE_DOC_FILENAME @@ -72,11 +54,32 @@ app = Flask(__name__, template_folder="templates") UPLOAD_BASE = PROJECT_ROOT / "src" / "web" / "uploads" SESSION_LOG_FILE = "session.log" SESSION_RESULT_FILE = "result.json" +INVOICE_GROUPS_FILE = "invoice_groups.json" -# ================================================================ -# 日志收集器 — 捕获管道日志到文件，SSE 端点通过 tail -f 读取 -# ================================================================ +def _save_invoice_groups(session_dir: Path, groups: dict[str, list[dict[str, str]]]) -> None: + """保存发票分类结果到 session 目录的 JSON 文件""" + groups_path = session_dir / INVOICE_GROUPS_FILE + # 只保存各组的数量统计，避免重复存储完整发票数据 + data = { + "travel_count": len(groups.get("travel", [])), + "general_count": len(groups.get("general", [])), + "application_count": len(groups.get("application", [])), + } + with open(groups_path, "w", encoding="utf-8") as f: + json.dump(data, f, ensure_ascii=False, indent=2) + + +def _load_invoice_groups(session_dir: Path) -> dict[str, int] | None: + """从 session 目录加载发票分类统计""" + groups_path = session_dir / INVOICE_GROUPS_FILE + if not groups_path.exists(): + return None + try: + with open(groups_path, encoding="utf-8") as f: + return cast(dict[str, int] | None, json.load(f)) + except Exception: + return None class _SSELogHandler(logging.Handler): @@ -164,69 +167,33 @@ def _resolve_invoice_csv(session_dir: Path) -> Path | None: return None -def _has_general_invoices(rows: list[dict[str, str]]) -> bool: - """检查发票列表中是否包含普通发票（需要生成易耗品出库单）""" - invoices = [] - for row in rows: - # 支付记录格式：从 _invoices_json 还原 - invoices_json = row.get("_invoices_json", "") - if invoices_json: - try: - invoices.extend(json.loads(invoices_json)) - except json.JSONDecodeError: - pass - # 发票级别格式：直接使用 - elif "发票类型" in row: - invoices.append(row) - if not invoices: - return False - groups = _classify_invoice_batch(invoices) - return len(groups["general"]) > 0 - - -def _get_invoice_groups(rows: list[dict[str, str]]) -> dict[str, int]: - """统计发票类型分布""" - invoices = [] - for row in rows: - # 支付记录格式：从 _invoices_json 还原 - invoices_json = row.get("_invoices_json", "") - if invoices_json: - try: - invoices.extend(json.loads(invoices_json)) - except json.JSONDecodeError: - pass - # 发票级别格式：直接使用 - elif "发票类型" in row: - invoices.append(row) - groups = _classify_invoice_batch(invoices) - return { - "travel_count": len(groups["travel"]), - "general_count": len(groups["general"]), - } +# ================================================================ +# 出库单填写 +# ================================================================ def _try_fill_consumable_doc(session_dir: Path, config: dict[str, Any]) -> dict[str, Any]: """根据 CSV 填写易耗品出库单，供会话目录下载。 - 仅当存在普通发票时才生成出库单。纯差旅发票跳过。 + 从 invoice_groups.json 读取分类结果，仅当存在普通发票时才生成出库单。 """ if not CONSUMABLE_TEMPLATE.exists(): fill_log.warning("出库单模板不存在: %s", CONSUMABLE_TEMPLATE) return {"ok": False, "error": "出库单模板不存在，请将模板放在项目根目录"} + # 从统一的分类结果读取，避免重复解析 CSV/JSON + groups = _load_invoice_groups(session_dir) + if groups is None: + return {"ok": False, "error": "未找到发票分类数据，请先处理"} + + if not groups.get("general_count", 0): + fill_log.info("纯差旅发票，跳过易耗品出库单生成") + return {"ok": False, "skipped": True, "error": "差旅发票无需生成易耗品出库单"} + csv_path = _resolve_payment_csv(session_dir) if csv_path is None: return {"ok": False, "error": "未找到发票 CSV"} - # 检查是否有普通发票 - rows = load_csv(csv_path) - if rows is None: - return {"ok": False, "error": "CSV 读取失败"} - - if not _has_general_invoices(rows): - fill_log.info("纯差旅发票，跳过易耗品出库单生成") - return {"ok": False, "skipped": True, "error": "差旅发票无需生成易耗品出库单"} - out_doc = session_dir / CONSUMABLE_DOC_FILENAME try: fill_log.info("开始填写出库单: %s", out_doc.name) @@ -265,6 +232,10 @@ def run_pipeline_web(session_dir: Path, config: dict[str, Any]) -> dict[str, Any """在 Web 会话目录中执行发票提取，结果写入 session 目录下的文件 注意：不再自动提交财务系统。提交通由 /api/submit-financial/ 触发。 + + 在发票提取和匹配完成后立即判断报销类型： + - 差旅发票：调用 LLM 提取差旅信息并缓存到 travel_info.json + - 普通发票：无需额外提取（normal_info.json 待实现） """ start = time.time() @@ -281,6 +252,38 @@ def run_pipeline_web(session_dir: Path, config: dict[str, Any]) -> dict[str, Any if applications: save_application_json(applications, session_dir / "travel_applications.json") + # 保存分类结果（供后续步骤统一读取） + _save_invoice_groups(session_dir, groups) + + # ---- Step 2: 差旅/普通信息提取 ---- + is_travel = bool(groups.get("travel")) and not bool(groups.get("general")) + if is_travel: + from src.doc.llm_extractor import CACHE_DIR_NAME, extract_travel_info, load_cache + + # 检查缓存是否已有 + cache_map = load_cache(session_dir) + if not cache_map.get("travel_info"): + fill_log.info("开始提取差旅信息...") + travel_info = extract_travel_info(source_dir=session_dir) + cache_dir = session_dir / CACHE_DIR_NAME + cache_dir.mkdir(parents=True, exist_ok=True) + with open(cache_dir / "travel_info.json", "w", encoding="utf-8") as f: + json.dump(travel_info, f, ensure_ascii=False, indent=2) + fill_log.info("差旅信息已保存到缓存") + else: + from src.doc.llm_extractor import CACHE_DIR_NAME, extract_normal_info, load_cache + + # 检查缓存是否已有 + cache_map = load_cache(session_dir) + if not cache_map.get("normal_info"): + fill_log.info("开始提取普通发票信息...") + normal_info = extract_normal_info(source_dir=session_dir) + cache_dir = session_dir / CACHE_DIR_NAME + cache_dir.mkdir(parents=True, exist_ok=True) + with open(cache_dir / "normal_info.json", "w", encoding="utf-8") as f: + json.dump(normal_info, f, ensure_ascii=False, indent=2) + fill_log.info("普通发票信息已保存到缓存") + # 统计发票总数 invoice_count = sum(len(inv.get("_matched_invoices", [])) for inv in invoices) @@ -302,7 +305,7 @@ def run_pipeline_web(session_dir: Path, config: dict[str, Any]) -> dict[str, Any def run_financial_submit(session_dir: Path, config: dict[str, Any]) -> dict[str, Any]: """执行财务系统填报（从前端确认后调用） - 根据发票类型选择填报模式： + 从 invoice_groups.json 读取分类结果，根据发票类型选择填报模式： - 纯差旅发票：差旅报销模式（TODO） - 含普通发票：普通报销模式 """ @@ -312,14 +315,10 @@ def run_financial_submit(session_dir: Path, config: dict[str, Any]) -> dict[str, from src.bot import run_bot_web - # TODO: 改为从缓存读取或直接请求 LLM，与差旅报销保持一致 - # bot_invoices = load_invoice_data(str(csv_path), config) - - # 判断发票类型 - rows = load_csv(csv_path) - if rows: - invoice_groups = _get_invoice_groups(rows) - if invoice_groups["travel_count"] and not invoice_groups["general_count"]: + # 从统一的分类结果读取，避免重复解析 + groups = _load_invoice_groups(session_dir) + if groups: + if groups.get("travel_count", 0) and not groups.get("general_count", 0): fill_log.info("检测到纯差旅发票，使用差旅报销模式") # TODO: 差旅报销填报流程 else: