205 lines
6.8 KiB
Python
205 lines
6.8 KiB
Python
"""
|
|
浏览器自动化填报 — 公共基类
|
|
|
|
提供浏览器生命周期管理、登录、导航、截图等公共操作。
|
|
"""
|
|
|
|
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"))
|