日常报销和差旅报销都可以走通

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"))