""" 浏览器自动化填报 使用 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)