用 Python 批量生成销售日报，告别每天复制粘贴

from dataclasses import dataclass
from datetime import date
from typing import Optional
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential


@dataclass
class YiniuProxy:
    """动态转发代理配置，集中管理鉴权与入口。"""
    order_id: str
    secret: str
    host: str = "dynamic.ynyun-proxy.com"
    port: int = 8000
    region: str = "cn"

    @property
    def tunnel(self) -> str:
        # 用户名里带订单号与地区是代理的鉴权约定，按后台接入串填
        user = f"{self.order_id}-region-{self.region}"
        return f"http://{user}:{self.secret}@{self.host}:{self.port}"


PROXY = YiniuProxy(order_id="YOUR_ORDER", secret="YOUR_SECRET").tunnel


@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10))
def fetch_channel_orders(channel: str, day: date) -> list[dict]:
    """拉取单个渠道某天的订单原始数据。

    Args:
        channel: 渠道标识，例如 "taobao" 或 "douyin"
        day: 统计日期

    Returns:
        平台返回的原始订单字典列表

    Raises:
        httpx.HTTPError: 三次重试后仍失败
    """
    resp = httpx.get(
        f"https://api.example.com/{channel}/orders",
        params={"date": day.isoformat()},
        headers={"User-Agent": "Mozilla/5.0 (compatible; daily-report/1.0)"},
        proxies={"http://": PROXY, "https://": PROXY},
        timeout=15,
    )
    resp.raise_for_status()
    return resp.json()["data"]

def fetch_all(channels: list[str], day: date) -> list[SalesRecord]:
    records: list[SalesRecord] = []
    for ch in channels:
        try:
            raw = fetch_channel_orders(ch, day)
            records.extend(normalize(raw, ch, day))
        except Exception as exc:  # 单渠道失败不应拖垮整份日报
            logging.error("渠道 %s 拉取失败: %s", ch, exc)
    return records

@dataclass
class SalesRecord:
    channel: str
    order_date: date
    order_count: int
    total_amount: float
    customer_count: int

    @property
    def avg_order_value(self) -> float:
        if self.customer_count == 0:
            return 0.0
        return self.total_amount / self.customer_count

def normalize(raw: list[dict], channel: str, day: date) -> list[SalesRecord]:
    """把平台原始订单转成统一的 SalesRecord。"""
    records: list[SalesRecord] = []
    for item in raw:
        records.append(
            SalesRecord(
                channel=channel,
                order_date=day,
                order_count=int(item["order_cnt"]),
                total_amount=float(item["pay_amount"]),
                customer_count=int(item["buyer_cnt"]),
            )
        )
    return records

from datetime import date
import pytest
from openpyxl import load_workbook
def test_aggregate_mom():
 today = [SalesRecord("self", date(2026, 8, 1), 10, 1000.0, 8)]
 prev = [SalesRecord("self", date(2026, 7, 31), 10, 500.0, 8)]
 result = aggregate(today, prev)
 assert result["mom"] == 100.0
def test_normalize_missing_field():
 raw = [{"order_cnt": "3", "pay_amount": "300", "buyer_cnt": "2"}]
 assert len(normalize(raw, "self", date(2026, 8, 1))) == 1
def test_write_report_readable(tmp_path):
 recs = [SalesRecord("self", date(2026, 8, 1), 3, 300.0, 2)]
 p = tmp_path / "r.xlsx"
 write_report(str(p), recs, aggregate(recs))
 assert load_workbook(p)["销售日报"]["A1"].value == "渠道"

import logging
logging.basicConfig(filename="report.log", level=logging.INFO,
 format="%(asctime)s %(levelname)s %(message)s")
def main() -> None:
 day = date.today()
 all_records: list[SalesRecord] = fetch_all(["taobao", "douyin", "self"], day)
 if not all_records:
 logging.error("全部渠道拉取失败，终止生成")
 return
 summary = aggregate(all_records)
 write_report(f"sales_{day.isoformat()}.xlsx", all_records, summary)
 logging.info("日报已生成：%s", summary)

import schedule
import time
schedule.every().day.at("07:00").do(main)
while True:
 schedule.run_pending()
 time.sleep(30)