nodriver 报表采集实战：自动下载、检查文件与合并数据

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install nodriver==0.50.99

import asyncio
import base64
import csv
import io
import time
from datetime import date
from decimal import Decimal
from pathlib import Path
import nodriver as uc
 
REPORTS = [
    ("2026-10-27", "date,order_id,amount,currency\n2026-10-27,o1,12.50,CNY\n2026-10-27,o2,20.00,CNY\n"),
    ("2026-10-28", "date,order_id,amount,currency\n2026-10-28,o3,9.90,CNY\n"),
]
 
async def wait_for_file(path, timeout=30):
    deadline = time.monotonic() + timeout
    previous, stable = None, 0
    while time.monotonic() < deadline:
        if path.exists():
            stat = path.stat()
            current = (stat.st_size, stat.st_mtime_ns)
            stable = stable + 1 if current == previous and stat.st_size > 0 else 0
            previous = current
            if stable >= 3:
                return path
        await asyncio.sleep(0.3)
    raise TimeoutError(f"文件未稳定：{path.name}")
 
def parse_report(path, expected_date):
    raw = path.read_bytes()
    if raw.lstrip().startswith(b"<"):
        raise ValueError("下载内容可能是 HTML，不接受为报表")
    reader = csv.DictReader(io.StringIO(raw.decode("utf-8-sig")))
    if reader.fieldnames != ["date", "order_id", "amount", "currency"]:
        raise ValueError("CSV 列名或顺序变化")
    rows = list(reader)
    if not rows:
        raise ValueError("报表没有记录，请检查是否为明确空日报")
    for row in rows:
        if date.fromisoformat(row["date"]).isoformat() != expected_date:
            raise ValueError("报表日期与请求不一致")
        value = Decimal(row["amount"])
        if not value.is_finite() or value < 0 or row["currency"] != "CNY" or not row["order_id"]:
            raise ValueError("金额、币种或订单编号不合法")
    return rows
 
async def main():
    output = Path("report-downloads")
    output.mkdir()
    browser = await uc.start(headless=True)
    records, seen = [], set()
    try:
        tab = await browser.get("about:blank")
        await tab.set_download_path(str(output.resolve()))
        for report_date, text in REPORTS:
            filename = f"report-{report_date}.csv"
            target = output / filename
            encoded = base64.b64encode(text.encode("utf-8")).decode("ascii")
            url = "data:text/csv;base64," + encoded
            await tab.download_file(url, filename)
            await wait_for_file(target)
            for row in parse_report(target, report_date):
                key = (row["date"], row["order_id"], row["currency"])
                if key in seen:
                    raise ValueError("跨文件出现重复日报记录")
                seen.add(key)
                records.append({**row, "source_file": filename})
    finally:
        browser.stop()
 
    with (output / "merged.csv").open("w", encoding="utf-8-sig", newline="") as stream:
        writer = csv.DictWriter(stream, fieldnames=[
            "date", "order_id", "amount", "currency", "source_file"
        ])
        writer.writeheader()
        writer.writerows(records)
    total = sum((Decimal(row["amount"]) for row in records), Decimal(0))
    print({"rows": len(records), "total_cny": str(total)})
 
uc.loop().run_until_complete(main())

python download_reports.py