From e15ae99db8dc69a1985d060b5da66c7011f5ef69 Mon Sep 17 00:00:00 2001 From: houhuan Date: Sat, 18 Apr 2026 17:21:38 +0800 Subject: [PATCH] Final version: Fixed date selection logic and verified full flow --- main.py | 73 +++++++++++----- secsion_export.py | 219 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 268 insertions(+), 24 deletions(-) create mode 100644 secsion_export.py diff --git a/main.py b/main.py index 0eb994c..004a453 100644 --- a/main.py +++ b/main.py @@ -110,38 +110,62 @@ class ReportAutomation: logger.info(f"访问统计页面: {self.secsion_stats_url}") await page.goto(self.secsion_stats_url) - # 等待页面加载完成，以“导出报表”按钮为准 + # 等待页面加载完成 logger.info("等待统计页面加载...") + await page.wait_for_load_state("networkidle") export_btn = page.get_by_role("button", name="导出报表") await export_btn.wait_for(state="visible", timeout=20000) logger.info(f"设置查询日期范围: {self.start_date} 至 {self.end_date}") + # 严格按照参考文件 secsion_export.py 的逻辑 try: - # 找到所有的日期输入框 - inputs = page.locator("input") - count = await inputs.count() - - date_input_indices = [] - for i in range(count): - val = await inputs.nth(i).input_value() - if "-" in val and len(val) >= 10: - date_input_indices.append(i) - - if len(date_input_indices) >= 2: - # 填入开始日期 - await inputs.nth(date_input_indices[0]).click() - await inputs.nth(date_input_indices[0]).fill(self.start_date) - await inputs.nth(date_input_indices[0]).press("Enter") + # 两个日期输入框都叫“请选择日期”，用 nth 区分 + start_input = page.get_by_role("textbox", name="请选择日期").nth(0) + end_input = page.get_by_role("textbox", name="请选择日期").nth(1) + + async def set_date(input_box, date_str: str): + logger.info(f"尝试设置日期: {date_str}") + # 1. 点击输入框 + await input_box.click() - # 填入结束日期 - await inputs.nth(date_input_indices[1]).click() - await inputs.nth(date_input_indices[1]).fill(self.end_date) - await inputs.nth(date_input_indices[1]).press("Enter") - else: - logger.warning("未找到足够的日期输入框，尝试使用默认日期导出") + # 2. 先尝试直接填值 + Enter (参考文件逻辑) + try: + # 由于是 readonly，fill 可能会超时，这里设置较短超时 + await input_box.fill(date_str, timeout=2000) + await input_box.press("Enter") + await page.wait_for_timeout(200) + except Exception: + logger.info("直接填值失败或超时，将尝试点击日历单元格") + + # 3. 若没有生效，则打开日历点击“日”单元格 (参考文件逻辑) + val = await input_box.input_value() + if val != date_str: + logger.info(f"值未同步({val} != {date_str})，执行日历单元格点击") + # 确保日历已弹出 + await input_box.click() + day = str(int(date_str.split("-")[2])) + # 参考文件使用 cell 角色 + # page.get_by_role("cell", name=day).click() + # 考虑到可能有多个月份显示，取最后一个弹出的 + await page.get_by_role("cell", name=day).last.click() + await page.wait_for_timeout(500) + + await set_date(start_input, self.start_date) + await set_date(end_input, self.end_date) + + # 等待数据请求完成 + logger.info("等待数据请求完成...") + await asyncio.sleep(2) + + # 截图确认日期设置后的状态 + logger.info("保存日期设置确认截图: date_setting_check.png") + await page.screenshot(path="date_setting_check.png") + except Exception as e: - logger.warning(f"填充日期时遇到问题: {str(e)}") + logger.error(f"日期选择逻辑执行失败: {str(e)}") + await page.screenshot(path="date_error.png") + raise e # 点击导出报表并捕获下载 logger.info("点击导出报表...") @@ -149,7 +173,8 @@ class ReportAutomation: await export_btn.click() download = await download_info.value - filename = f"commodity_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" + # 使用服务器建议的文件名 + filename = download.suggested_filename save_path = os.path.join(self.download_dir, filename) await download.save_as(save_path) diff --git a/secsion_export.py b/secsion_export.py new file mode 100644 index 0000000..697eb17 --- /dev/null +++ b/secsion_export.py @@ -0,0 +1,219 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +secsion.com 报表导出脚本（给 OpenClaw/自动化调用） + +功能： +1) 登录 https://secsion.com:8000/ +2) 进入 /commodityStatistics +3) 选择开始/结束日期 +4) 点击“导出报表”，保存 xlsx 到本地目录 + +推荐使用 Playwright（更稳定地处理下载与证书忽略）。 +""" + +from __future__ import annotations + +import argparse +import os +import sys +import time +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Optional +from urllib.parse import urlparse + +import requests +from playwright.sync_api import TimeoutError as PWTimeoutError +from playwright.sync_api import sync_playwright + + +@dataclass(frozen=True) +class Config: + base_url: str = "https://secsion.com:8000" + login_url: str = "https://secsion.com:8000/login?redirect=%252Fhomepage" + report_url: str = "https://secsion.com:8000/commodityStatistics" + role: str = "店铺" + username: str = "" + password: str = "" + + +def _parse_date(s: str) -> str: + # 统一校验并格式化为 YYYY-MM-DD + dt = datetime.strptime(s, "%Y-%m-%d") + return dt.strftime("%Y-%m-%d") + + +def _ensure_dir(p: Path) -> None: + p.mkdir(parents=True, exist_ok=True) + + +def _download_via_requests(url: str, out_dir: Path, filename: Optional[str] = None) -> Path: + # 导出链接常见为 https://secsion.com:8082/xxx.xlsx（可能证书不受信任） + # verify=False 用于忽略证书（等价于浏览器“继续前往”） + resp = requests.get(url, stream=True, timeout=120, verify=False) + resp.raise_for_status() + + if not filename: + path = urlparse(url).path + filename = Path(path).name or f"commodity_export_{int(time.time())}.xlsx" + if not filename.lower().endswith(".xlsx"): + filename += ".xlsx" + + save_path = out_dir / filename + with open(save_path, "wb") as f: + for chunk in resp.iter_content(chunk_size=1024 * 128): + if chunk: + f.write(chunk) + return save_path + + +def export_report( + *, + cfg: Config, + start_date: str, + end_date: str, + out_dir: Path, + headless: bool = True, + timeout_ms: int = 30_000, +) -> Path: + _ensure_dir(out_dir) + + with sync_playwright() as p: + browser = p.chromium.launch(headless=headless) + context = browser.new_context(ignore_https_errors=True, accept_downloads=True) + page = context.new_page() + + # 1) 打开登录页 + page.goto(cfg.login_url, timeout=timeout_ms, wait_until="domcontentloaded") + + # 2) 选择角色（默认“店铺”） + # 页面是 radio，按可见文本定位 + page.get_by_role("radio", name=cfg.role).check() + + # 3) 输入账号密码 + page.get_by_role("textbox", name="请输入用户名").fill(cfg.username) + page.get_by_role("textbox", name="请输入密码").fill(cfg.password) + + # 4) 登录 + page.get_by_role("button", name="登 录").click() + # 等待跳转（不强依赖具体 URL，避免页面改版） + page.wait_for_load_state("networkidle", timeout=timeout_ms) + + # 5) 进入报表页 + page.goto(cfg.report_url, timeout=timeout_ms, wait_until="domcontentloaded") + page.wait_for_load_state("networkidle", timeout=timeout_ms) + + # 6) 设置日期范围 + # 两个日期输入框都叫“请选择日期”，用 nth 区分 + start_input = page.get_by_role("textbox", name="请选择日期").nth(0) + end_input = page.get_by_role("textbox", name="请选择日期").nth(1) + + def set_date(input_box, date_str: str) -> None: + # 先尝试直接填值 + Enter（部分组件会生效） + input_box.click() + input_box.fill(date_str) + input_box.press("Enter") + page.wait_for_timeout(200) + + # 若没有生效，则打开日历点击“日”单元格（更稳） + val = input_box.input_value() + if val != date_str: + input_box.click() + day = str(int(date_str.split("-")[2])) + # 月份不一致时这里可能需要先切换月份；多数情况下导出同月数据足够。 + page.get_by_role("cell", name=day).click() + page.wait_for_timeout(200) + + set_date(start_input, start_date) + set_date(end_input, end_date) + + # 7) 导出 + export_btn = page.get_by_role("button", name="导出报表") + + # 尝试走 Playwright 的下载通道（最优） + try: + with page.expect_download(timeout=20_000) as d: + export_btn.click() + download = d.value + suggested = download.suggested_filename or f"commodity_export_{start_date}_{end_date}.xlsx" + if not suggested.lower().endswith(".xlsx"): + suggested += ".xlsx" + save_path = out_dir / suggested + download.save_as(str(save_path)) + browser.close() + return save_path + except PWTimeoutError: + # 回退：有些情况下会打开一个新地址（.xlsx 链接）但不触发下载事件 + export_btn.click() + page.wait_for_timeout(1500) + + xlsx_url = None + for pg in context.pages: + if pg.url.lower().endswith(".xlsx"): + xlsx_url = pg.url + break + if not xlsx_url and page.url.lower().endswith(".xlsx"): + xlsx_url = page.url + + if not xlsx_url: + browser.close() + raise RuntimeError("未捕获到导出的 xlsx 链接；请把页面导出后的实际跳转 URL 发我，我再适配。") + + # 直链下载（忽略证书） + filename = f"commodity_export_{start_date}_{end_date}.xlsx" + save_path = _download_via_requests(xlsx_url, out_dir=out_dir, filename=filename) + browser.close() + return save_path + + +def main(argv: list[str]) -> int: + parser = argparse.ArgumentParser( + description="secsion.com 报表导出（支持某一天/时间段）。", + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + parser.add_argument("--start", required=False, help="开始日期 YYYY-MM-DD（若仅导出某一天，可只填 start）") + parser.add_argument("--end", required=False, help="结束日期 YYYY-MM-DD（缺省时与 start 相同）") + parser.add_argument("--out", default="./downloads", help="导出文件保存目录") + parser.add_argument("--headless", action="store_true", help="无头模式运行（CI/机器人推荐）") + parser.add_argument("--show", action="store_true", help="显示浏览器窗口（调试用，优先级高于 --headless）") + + parser.add_argument("--role", default="店铺", help="登录角色：店铺编码/管理员/店铺/业务员") + parser.add_argument("--username", default=os.getenv("SECSION_USERNAME", ""), help="账号（也可用环境变量 SECSION_USERNAME）") + parser.add_argument("--password", default=os.getenv("SECSION_PASSWORD", ""), help="密码（也可用环境变量 SECSION_PASSWORD）") + + args = parser.parse_args(argv) + + # 交互式兜底（给 OpenClaw 也可以直接传参，不走交互） + start = args.start or input("请输入开始日期(YYYY-MM-DD): ").strip() + end = args.end or input("请输入结束日期(YYYY-MM-DD，回车=同开始日期): ").strip() or start + + try: + start = _parse_date(start) + end = _parse_date(end) + except ValueError: + print("日期格式错误，请使用 YYYY-MM-DD，例如 2026-04-15", file=sys.stderr) + return 2 + + if not args.username or not args.password: + print("缺少账号或密码：请传 --username/--password 或设置环境变量 SECSION_USERNAME/SECSION_PASSWORD", file=sys.stderr) + return 2 + + cfg = Config(role=args.role, username=args.username, password=args.password) + out_dir = Path(args.out).expanduser().resolve() + headless = False if args.show else bool(args.headless) + + try: + saved = export_report(cfg=cfg, start_date=start, end_date=end, out_dir=out_dir, headless=headless) + except Exception as e: + print(f"导出失败：{e}", file=sys.stderr) + return 1 + + print(str(saved)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) +