重构项目为LLM 驱动
This commit is contained in:
wandering
457
src/bot.py
Normal file
457
src/bot.py
Normal file
@@ -0,0 +1,457 @@
|
||||
"""
|
||||
浏览器自动化填报
|
||||
|
||||
使用 Playwright 操作财务报销系统，自动完成登录、填单、上传附件等操作。
|
||||
|
||||
对外接口:
|
||||
load_invoice_data(csv_path, config) -> list[dict] 从 CSV 加载并补全默认值
|
||||
run_bot(config, invoices) 启动浏览器并执行填报流程
|
||||
"""
|
||||
|
||||
import csv
|
||||
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 _safe_float(value: str | None, default: float = 0.0) -> float:
|
||||
"""安全转换为浮点数，空值或转换失败时返回默认值"""
|
||||
if value is None or str(value).strip() == "":
|
||||
return default
|
||||
try:
|
||||
return float(value)
|
||||
except (ValueError, TypeError):
|
||||
return default
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# CSV 数据加载
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
|
||||
def load_invoice_data(csv_path: str, config: dict[str, str | Path]) -> list[dict[str, str | float | Any | Path]]:
|
||||
"""从 CSV 加载发票数据，自动补全空白字段的默认值
|
||||
|
||||
CSV 以支付记录为主键，每行包含 _invoices_json 字段（JSON 序列化的发票列表）。
|
||||
本函数还原为发票级别的数据列表。
|
||||
"""
|
||||
import json
|
||||
|
||||
invoices = []
|
||||
with open(csv_path, encoding="utf-8") as f:
|
||||
reader = csv.DictReader(f)
|
||||
for row in reader:
|
||||
invoices_json = row.get("_invoices_json", "")
|
||||
if not invoices_json:
|
||||
continue
|
||||
try:
|
||||
inv_list = json.loads(invoices_json)
|
||||
for inv in inv_list:
|
||||
invoices.append(
|
||||
{
|
||||
"seq": row.get("序号", ""),
|
||||
"invoice_no": inv.get("发票号码", ""),
|
||||
"invoice_date": inv.get("开票日期", ""),
|
||||
"item_name": inv.get("项目名称", ""),
|
||||
"spec_model": inv.get("规格型号", ""),
|
||||
"total_amount": _safe_float(inv.get("价税合计")),
|
||||
"seller_name": inv.get("销售方名称", ""),
|
||||
"person_name": inv.get("人员姓名") or config.get("default_name", ""),
|
||||
"card_date": _format_date(row.get("刷卡日期") or ""),
|
||||
"card_no": row.get("公务卡号") or config.get("default_card_no", ""),
|
||||
"card_amount": _safe_float(row.get("刷卡金额")),
|
||||
"remark": row.get("备注") or "",
|
||||
"person_id": row.get("工号") or config.get("default_person_id", ""),
|
||||
}
|
||||
)
|
||||
except json.JSONDecodeError:
|
||||
log.warning(f"无法解析发票 JSON: {invoices_json[:50]}...")
|
||||
return invoices
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 报销机器人
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
|
||||
class ReimburseBot:
|
||||
"""财务报销自动化机器人"""
|
||||
|
||||
def __init__(self, config: dict[str, Any], headless: bool = False):
|
||||
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": 1920, "height": 1080})
|
||||
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) -> 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["reimburse_page"]
|
||||
self.page.goto(common_url, wait_until="domcontentloaded", timeout=15000)
|
||||
self._wait_for('text="单据状态:"', timeout=5000)
|
||||
|
||||
def open_reimburse_menu(self) -> None:
|
||||
"""点击「新增」创建新报销单"""
|
||||
log.info("创建新报销单...")
|
||||
self.page.wait_for_timeout(2000)
|
||||
|
||||
try:
|
||||
self.page.click('button:has-text("新增")', 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_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_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.page.wait_for_timeout(1000)
|
||||
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_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("#next3", 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 submit(self) -> None:
|
||||
"""提交报销单"""
|
||||
log.info("提交报销单...")
|
||||
try:
|
||||
self.page.click("#submit", timeout=5000)
|
||||
self.page.wait_for_timeout(1000)
|
||||
self._screenshot("submitted")
|
||||
except Exception as e:
|
||||
log.error(f"提交失败: {e}")
|
||||
self._screenshot("submit_error")
|
||||
raise
|
||||
|
||||
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 run_bot(
|
||||
config: dict[str, Any], invoices: list[dict[str, Any]], headless: bool = False, work_dir: Path | None = None
|
||||
) -> None:
|
||||
"""执行完整的浏览器填报流程"""
|
||||
if not config["username"] or not config["password"]:
|
||||
raise ValueError("缺少用户名或密码")
|
||||
|
||||
bot = ReimburseBot(config, headless=headless)
|
||||
bot.work_dir = work_dir
|
||||
try:
|
||||
bot.launch()
|
||||
bot.login_portal()
|
||||
bot.navigate_to_reimburse()
|
||||
bot.open_reimburse_menu()
|
||||
bot.fill_basic_info()
|
||||
bot.add_reimburse_items(invoices)
|
||||
bot.fill_payment(invoices)
|
||||
bot.upload_attachments(invoices)
|
||||
# bot.submit() # 确认无误后再取消注释
|
||||
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], invoices: list[dict[str, Any]], work_dir: Path) -> None:
|
||||
"""Web 模式填报 — headless，附件从指定目录读取"""
|
||||
if not config["username"] or not config["password"]:
|
||||
raise ValueError("缺少用户名或密码")
|
||||
|
||||
bot = ReimburseBot(config, headless=True)
|
||||
bot.work_dir = work_dir
|
||||
try:
|
||||
bot.launch()
|
||||
bot.login_portal()
|
||||
bot.navigate_to_reimburse()
|
||||
bot.open_reimburse_menu()
|
||||
bot.fill_basic_info()
|
||||
bot.add_reimburse_items(invoices)
|
||||
bot.fill_payment(invoices)
|
||||
bot.upload_attachments(invoices)
|
||||
except Exception as e:
|
||||
log.error(f"操作失败: {e}")
|
||||
try:
|
||||
bot._screenshot("error")
|
||||
except Exception:
|
||||
pass
|
||||
raise
|
||||
finally:
|
||||
bot.close()
|
||||
Reference in New Issue
Block a user