220 lines
8.2 KiB
Python
220 lines
8.2 KiB
Python
#!/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:]))
|
||
|