夜雨聆风在保险公司做财务，少不了和各种报表系统打交道。每到月末，要登录报表平台，手工选维度、点查询、等结果、下载——重复劳动不说，还容易被系统超时折磨。
这个报表分析系统，大家是否都熟悉，从上个东家到现在的公司的时候一看到这个系统还以为回到了原来，后来发现有很多公司都在用这个系统，这个系统原来不好的一点就是任何报表定时定制跑都要找 IT，那么个性的报表就只能手工，每次提取 10 几张报表那酸痛,有了小龙虾那这些机械的事情就可以交给它了
这篇文章用一个真实案例（赔付率报表的多维度提取），说明如何用 WorkBuddy 的 Skill 机制，把"人点报表"变成"AI 自动跑"。代码完整，但敏感信息（URL、账号密码）已脱敏，你完全可以迁移到自己的场景。
一、整体思路：报表自动化的三层架构
┌──────────────────────────────────────────────┐│ WorkBuddy Skill (SKILL.md) │ ← 自然语言触发层│ "提取报表" → 调用执行脚本 │└──────────────────┬───────────────────────────┘ │ invoke┌──────────────────▼───────────────────────────┐│ Python + Playwright (report.py) │ ← 核心自动化逻辑│ 登录 → 填参数 → 提交离线任务 → 等待 → 下载 │└──────────────────┬───────────────────────────┘ │┌──────────────────▼───────────────────────────┐│ 报表平台（Browser） │ ← 真实系统（被自动化操作）└──────────────────────────────────────────────┘核心依赖只有两个：
• playwright：控制真实浏览器完成所有交互• SKILL.md：告诉 WorkBuddy 何时、怎么调用这个脚本
二、核心代码结构（report.py 全解析）
2.1 整体框架
脚本分四步走：
① login() → 打开登录页，填账号密码，处理"用户没注销"弹窗② submit_report_task() → 进入菜单 → 选报表 → 填日期 → 勾维度 → 提交离线任务③ wait_and_download() → 轮询任务中心，直到"已完成" → 触发 JS 下载④ execute_report_tasks() → 编排前3步，批量跑多个报表2.2 登录函数
from playwright.sync_api import sync_playwright, TimeoutErrorBASE_URL = "https://rpt.example-company.com" # 脱敏：换成你自己的报表平台地址LOGIN_URL = f"{BASE_URL}/login.jsp"TASK_CENTER_URL = f"{BASE_URL}/schedule/tasklist.jsp?pageop=SCH"# 伪装 IE7，避免样式兼容问题UA_IE7 = ( "Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 10.0; " "WOW64; Trident/7.0; .NET4.0C; .NET4.0E; Tablet PC 2.0)")def login(page, user: str, pwd: str): """登录报表平台，处理会话冲突重试""" page.goto(LOGIN_URL) for attempt in range(1, 4): page.fill("#user", user) page.fill("#password", pwd) with page.expect_navigation(): page.click("img[src*='go.gif']") # 处理"用户没注销"会话冲突 try: page.wait_for_selector("text=用户没注销", timeout=1000) print(f"⚠️ 第 {attempt} 次：发现未注销会话，清理后重试…") page.locator("input[name='check'][value='clear']").check() page.fill("#user", user) page.fill("#password", pwd) with page.expect_navigation(): page.click("img[src*='go.gif']") except TimeoutError: print(f"✅ 登录成功（第 {attempt} 次尝试）") return True raise RuntimeError("连续 3 次仍遇会话冲突，登录终止")💡 关键经验：报表平台通常有会话并发限制（同一账号不能多处登录）。处理方式是先"强制清会话再登录"，而非报错退出。
2.3 提交离线任务
这是最核心的函数，把"人在页面上点的每一步"翻译成 Playwright 操作：
def submit_report_task(params: dict) -> None: """ 进入报表参数页 → 填日期 → 勾维度 → 提交离线任务 params 格式： { 'page': Playwright Page对象, 'menu_text': "左侧菜单文本", 'report_row_id': "报表行名称", 'date_from': ("年", "月", "日"), 'date_to': ("年", "月", "日"), 'date_bis': ("年", "月", "日"), 'dim_ids': ["维度ID1", "维度ID2", ...], 'hidden_vars': {"表单隐藏字段名": "值", ...}, 'click_menu': True/False # 是否需要重新点击菜单 } """ page = params['page'] menu_text = params['menu_text'] report_row_id = params['report_row_id'] date_from = params['date_from'] date_to = params['date_to'] date_bis = params['date_bis'] dim_ids = params['dim_ids'] hidden_vars = params['hidden_vars'] click_menu = params.get('click_menu', True) # ── Step 1: 左侧菜单导航 ── contents = page.frame(name="contents") if click_menu: # 点击顶级菜单（如"财务报表"） contents.get_by_role("cell", name=menu_text, exact=True).click(timeout=60000) # 点击具体报表行 contents.get_by_role("cell", name=report_row_id, exact=True).click(timeout=60000) # ── Step 2: 填日期字段 ── frm = page.frame(name="main") (y1, m1, d1), (y2, m2, d2), (yb, mb, db) = date_from, date_to, date_bis frm.fill('input[name="year11"]', y1); frm.fill('input[name="month11"]', m1) frm.fill('input[name="day11"]', d1) frm.fill('input[name="year12"]', y2); frm.fill('input[name="month12"]', m2) frm.fill('input[name="day12"]', d2) frm.fill('input[name="year21"]', yb); frm.fill('input[name="month21"]', mb) frm.fill('input[name="day21"]', db) # ── Step 3: 勾选维度（险种、机构、渠道等）── topic_ifr = (page.frame_locator("frame[name='main']") .frame_locator("iframe[name='ifrTopic']")) for tid in dim_ids: topic_ifr.locator(f"#vis{tid}").check() time.sleep(10) # 等服务端算维度列表 # ── Step 4: 设置隐藏字段（公司代码等）── for k, v in hidden_vars.items(): frm.evaluate(f'document.getElementById("{k}").value = {v!r}') # ── Step 5: 点"下一步"，进入参数确认页 ── frm.get_by_role("img", name="下一步").click() # ── Step 6: 勾"离线任务"，提交 ── frm.locator("input[name='isSchedule']").check() frm.wait_for_selector("#img6", state="visible", timeout=5000) frm.locator("#img6").click() page.wait_for_load_state("networkidle") print("📨 离线任务已提交")💡 Playwright 技巧：报表系统大量使用 iframe 和 frame 嵌套。定位方式：
• page.frame(name="xxx")：直接按 name 取 iframe• page.frame_locator().frame_locator()：链式调用处理多层嵌套• 维度列表在 iframe[name='ifrTopic']里，要用frame_locator才能进入
2.4 等待任务完成并下载
报表平台通常是"离线任务"模式：提交后服务器异步生成，生成完毕后才允许下载：
import redef wait_and_download_last_row( page: Page, out_dir: str | Path = "reports", row_index: int = 0, max_wait: int = 900, # 最多等15分钟 poll_interval: int = 30 # 每30秒轮询一次) -> str: """轮询任务中心，任务完成后自动下载""" out_dir = Path(out_dir) out_dir.mkdir(parents=True, exist_ok=True) page.goto(TASK_CENTER_URL, wait_until="networkidle") waited = 0 while waited < max_wait: # 定位任务行（用 id 属性筛选，避开表头） task_rows = page.locator("table:nth-of-type(2) tr[id]").all() if row_index >= len(task_rows): raise IndexError(f"行索引 {row_index} 超出范围（共 {len(task_rows)} 行）") status = task_rows[row_index].locator("td").nth(5).text_content().strip() print(f"📊 任务 {row_index + 1} 状态: {status}") if status == "已完成": print(f"✅ 任务已完成，开始下载") break elif status in ("执行中", "等待"): print(f"⏳ {status}，剩余 {max_wait - waited} 秒…") else: raise RuntimeError(f"任务异常状态: {status}") time.sleep(poll_interval) waited += poll_interval # 每2分钟强制刷新（任务中心有时不自动更新） if waited % 120 == 0: page.reload(wait_until="networkidle") if waited >= max_wait: raise TimeoutError(f"等待超时（{max_wait}秒）") # ── 从下载按钮的 onclick 提取报表ID ── target_row = task_rows[row_index] task_id = target_row.get_attribute("id") onclick = (target_row.locator("td").nth(8).locator("a") .get_attribute("onclick")) match = re.search(r"doReportDown\((\d+),\s*(\d+)\)", onclick) rpt_id = match.group(2) if match else None # ── 触发 JS 函数下载（报表平台通常不支持直接 HTTP 下载）── print("📥 开始下载…") with page.expect_download(timeout=1200000) as dl_info: page.evaluate(f"doReportDown({task_id}, {rpt_id})") dl = dl_info.value return _save_download(dl, out_dir)def _save_download(download, out_dir: Path) -> Path: """保存下载文件，自动处理重名""" fname = download.suggested_filename outfile = out_dir / fname counter = 1 while outfile.exists(): name, ext = fname.rsplit('.', 1) outfile = out_dir / f"{name}_{counter}.{ext}" counter += 1 download.save_as(outfile) # 自动解压 ZIP if outfile.suffix.lower() == ".zip": dest = out_dir / outfile.stem zipfile.ZipFile(outfile).extractall(dest) print(f"📂 已解压到 {dest}") return outfile💡 关键经验：报表平台的下载链接通常不能直接 HTTP GET（需要浏览器 Cookie/Session），所以用
page.evaluate()调用页面 JS 函数，配合expect_download()捕获下载事件。
2.5 批量编排：跑多个报表
一个脚本可以同时提交多个离线任务，等全部完成后再逐个下载：
from datetime import datetime, timedeltadef calculate_monthly_dates() -> tuple: """自动计算上月报表的日期区间""" now = datetime.now() last_month_end = now.replace(day=1) - timedelta(days=1) year = str(last_month_end.year) month = str(last_month_end.month).zfill(2) day = str(last_month_end.day).zfill(2) # 开始日期 = 一年前的下月1号 from_y, from_m = int(year) - 1, int(month) + 1 if from_m > 12: from_m = 1; from_y += 1 date_from = (str(from_y), str(from_m).zfill(2), "01") date_to = (year, month, day) return date_from, date_to, (year, month, day), f"{year}-{month}"def execute_report_tasks(username, password, task_list, file_suffix=None, headless=True): """ 批量执行报表任务（三阶段流水线） Args: username: 报表平台账号 password: 报表平台密码 task_list: 任务配置列表，每个任务是一个 dict file_suffix: 输出目录后缀（如 "2026-03"） """ if file_suffix is None: file_suffix = datetime.now().strftime("%Y-%m") main_out = Path("reports") / file_suffix main_out.mkdir(parents=True, exist_ok=True) with sync_playwright() as pw: browser = pw.chromium.launch( headless=headless, # Windows 下用本机 Chrome，macOS/Linux 用 Playwright 内置 executable_path=r"C:\Program Files\Google\Chrome\Application\chrome.exe" ) ctx = browser.new_context( user_agent=UA_IE7, accept_downloads=True ) page = ctx.new_page() try: # ── 第一阶段：批量提交所有离线任务 ── login(page, username, password) task_dirs = [] for i, task in enumerate(task_list): task_name = task.get('task_name', f"任务{i+1}") task_out_dir = main_out / task_name task_out_dir.mkdir(parents=True, exist_ok=True) task_dirs.append(task_out_dir) # 维度名称 → ID（需要映射表） dim_ids = get_dimension_ids(task.get('dimensions', [])) # 日期缺省则自动计算 df, dt, db, _ = calculate_monthly_dates() date_from = task.get('date_from') or df date_to = task.get('date_to') or dt date_bis = task.get('date_bis') or db submit_report_task({ 'page': page, 'menu_text': task['menu_text'], 'report_row_id': task['report_row_id'], 'date_from': date_from, 'date_to': date_to, 'date_bis': date_bis, 'dim_ids': tuple(dim_ids), 'hidden_vars': task.get('hidden_vars', {}), 'click_menu': task.get('click_menu', True), }) print(f" ✅ 任务 {i+1} 已提交") # ── 第二阶段：轮询等待全部完成 ── waited = 0 while waited < 900: page.goto(TASK_CENTER_URL, wait_until="networkidle") rows = page.locator("table:nth-of-type(2) tr[id]").all() statuses = [r.locator("td").nth(5).text_content().strip() for r in rows] print(f"📊 当前状态: {statuses}") if all(s == "已完成" for s in statuses): print("✅ 全部完成，开始下载") break time.sleep(30) waited += 30 if waited % 120 == 0: page.reload(wait_until="networkidle") # ── 第三阶段：逆序下载（提交顺序和下载顺序相反）── for i, task_out_dir in enumerate(task_dirs): rev_idx = len(task_dirs) - 1 - i wait_and_download_last_row( page=page, row_index=rev_idx, download_dir=str(task_out_dir) ) print(f"\n🎉 全部完成！文件在: {main_out}") finally: ctx.close() browser.close()2.6 维度映射（dimension_mapping.py）
维度名称和 ID 的对应关系需要一张映射表，存成独立模块：
# dimension_mapping.pyDIMENSION_MAP = { "三级机构": "89", "四级机构": "90", "出单机构": "91", "险种大类": "93", "险种产品": "94", "渠道类型": "101", "渠道细分": "102", "共保情况": "110", "代理人": "115", "销售团队": "120",}def get_dimension_ids(names: list[str]) -> list[str]: """将维度名称列表转换为 ID 列表""" missing = [n for n in names if n not in DIMENSION_MAP] if missing: raise ValueError(f"未知维度名称: {missing}") return [DIMENSION_MAP[n] for n in names]三、WorkBuddy Skill 的写法
脚本写好后，只需要一个 SKILL.md，WorkBuddy 就能理解"用户说提取报表，我该怎么调用"：
---name: extract-提取报表-reportdescription: 当用户提到「提取报表」时使用此skill。---# 提取报表## 使用场景用户说以下任意一句时触发：- 「提取报表」- 「帮我提取报表」- 「有新的报表吗」## 执行流程1. **设置编码**：运行前设置 `$env:PYTHONIOENCODING="utf-8"` > ⚠️ 必须设置，否则脚本因编码错误失败2. **运行脚本**：执行 report.py3. **等待完成**：报表下载约需 3~10 分钟4. **汇报结果**：告知用户文件存放路径## 脚本信息- **脚本路径**：`D:\OneDrive\pythonProject2\Report\report.py`- **执行命令**： ```powershell $env:PYTHONIOENCODING="utf-8" python "D:\OneDrive\pythonProject2\Report\report.py"• 输出位置： reports\{年月}\目录下，按任务名分子目录
注意事项
• 报表下载可能需要几分钟，请耐心等待 • 如遇报错，详细记录错误信息反馈给用户
把这文件放进 `~/.workbuddy/skills/extract-report/SKILL.md`，WorkBuddy 就会在听到相关指令时自动执行。---## 四、完整调用示例（main）```pythonif __name__ == "__main__": # ── 方式：任务列表（推荐）── task_list = [ { 'task_name': "基础维度报表", 'menu_text': "06-财务报表", 'report_row_id': "再保后历年制赔付率报表", # 日期留空 → 自动计算为上月 'date_from': None, 'date_to': None, 'date_bis': None, 'dimensions': ["三级机构", "四级机构", "出单机构", "险种大类", "险种产品"], 'hidden_vars': {"dptID": "9%% 公司名称", "dpt_text": "9 公司名"}, 'click_menu': True, }, { 'task_name': "共保维度报表", 'menu_text': "06-财务报表", 'report_row_id': "再保后历年制赔付率报表（不含超赔）", 'date_from': None, 'date_to': None, 'date_bis': None, 'dimensions': ["三级机构", "四级机构", "出单机构", "险种大类", "险种产品", "共保情况"], 'hidden_vars': {"dptID": "9%% 公司名称", "dpt_text": "9 公司名"}, 'click_menu': False, # 已在同一菜单下，无需再点 }, ] execute_report_tasks( username="你的账号", password="你的密码", task_list=task_list, headless=True # 开发时设为 False 看浏览器 )五、把技能变成定时自动化
每月末都要跑？加一个 WorkBuddy Automation 即可：
automation_update( mode="suggested create", name="月末报表提取", prompt=( "执行报表提取任务。运行命令：" "$env:PYTHONIOENCODING='utf-8'; " "python D:\\OneDrive\\pythonProject2\\Report\\report.py" "完成后告知用户报表已保存到 reports 目录。" ), scheduleType="recurring", rrule="FREQ=MONTHLY;BYMONTHDAY=1;BYHOUR=6;BYMINUTE=0", # 每月1日凌晨6点自动跑 status="PAUSED", # 用户确认后再激活)六、踩过的坑与解决思路
time.sleep(10) 等服务端返回 | ||
page.evaluate() 调 JS，配合 expect_download() | ||
page.reload() | ||
$env:PYTHONIOENCODING="utf-8" | ||
max_wait |
七、总结：这套方案能复用到哪些场景？
• 任何有 Web 界面的报表系统：保费报表、核保清单、理赔台账，都适用 • 财务对账：自动登录财务系统，下载科目余额表、银行流水 • EKP 待办处理：自动审批、批量处理待办（Playwright + iframe 操作） • 微信公众号：草稿发布、素材上传（已有现成 Skill）
核心逻辑只有一条：用 Playwright 模拟人操作浏览器，用 Skill 把它封装成 WorkBuddy 能理解的任务。写好一次，后边就是"说一句，AI 替你跑"。
如果对你有帮助，欢迎点在看、转发。
