# 每天花两小时导数据，这活其实可以全自动（附 Python 源码）
> 不是你不会写脚本，是没人告诉你那些琐碎问题（时区、编码、表头、excel 打开被占用）怎么处理。
## 一、两小时是怎么花掉的
先别急着看代码，先算一笔时间账。一个典型的电商运营，每天早上要做的事：
| 动作 | 实际耗时 | 其中真正有价值的部分 |
| ------------------- | ----- | ---------- |
| 打开三个后台，逐个导出订单 | 25 分钟 | 0 |
| 把三个 Excel 合成一个 | 20 分钟 | 5 分钟（看数） |
| 找出昨天没发货的单 | 15 分钟 | 10 分钟（决策） |
| 算各店销售额、对比昨天 | 20 分钟 | 15 分钟（决策） |
| 截图发群里，汇报 | 10 分钟 | 5 分钟 |
| 中途被叫走、Excel 卡死、路径找错 | 30 分钟 | 0 |
**两小时里，真正需要人判断的只有 30 分钟。** 剩下 90 分钟全是搬运。
搬运这件事的三个特点：重复、无判断、出错率高。手动合并三个 Excel，少粘一行、多粘一行，根本看不出来——但月底对账时就会出问题。
所以这不是「要不要自动化」的问题，是「为什么还在手动」的问题。
## 二、为什么大家还在手动
我问过不少做这行的，答案集中在四类：
**「我不会写代码。」** —— 这篇的代码复制就能跑，改两行配置就行。
**「试过写脚本，跑两天就废了。」** —— 这是最多的。脚本本身没错，是没处理那些「脏活」：时区、编码、日期格式、Excel 被占用、字段顺序变了。
**「每天的数据不一样，脚本写死不灵活。」** —— 配置化就能解，把「拉什么、怎么算」抽成配置。
**「出错了我不知道。」** —— 这是最致命的。脚本静默失败，你以为数据是对的，实际上是一周前的。所以要加监控和校验。
下面这份方案，重点不在「怎么写代码」，而在**怎么让脚本能长期跑下去不废**。
## 三、整体思路：配置驱动 + 定时 + 校验
```
配置(config) → 拉数(fetch) → 清洗(clean) → 汇总(aggregate) → 输出(excel) → 通知
↑ ↓
└────────────────────── 失败告警 · 数据校验 ──────────────────────────────────┘
```
三件事决定了这个方案能不能长期用：
1. **配置驱动**：加一个店、改一个指标，改 YAML 不改代码
2. **幂等 + 断点**：重跑不会产生重复数据，中途挂了能续
3. **校验 + 告警**：数据明显不对（比如订单数为 0）要主动报出来，不能静默
## 四、Python 源码
依赖：
```bash
pip install httpx pandas openpyxl pyyaml
```
### 4.1 配置文件
先把「每天要干什么」写清楚。这是让脚本活下来的第一步。
```yaml
# config.yaml
shops:
- name: 淘宝旗舰店
platform: taobao
app_key: your_appkey
app_secret: your_secret
shop_id: "12345678"
- name: 1688 批发
platform: alibaba
app_key: your_appkey
app_secret: your_secret
shop_id: "87654321"
- name: 京东自营
platform: jd
app_key: your_appkey
app_secret: your_secret
shop_id: "11223344"
report:
output_dir: ./reports
filename: "经营日报_{date}.xlsx"
timezone: Asia/Shanghai
# 导出哪些字段，顺序即列顺序
columns:
- order_id
- shop_name
- buyer_nick
- created_at
- pay_amount
- status
- waybill
rules:
# 未发货判定
pending_ship_status: ["WAIT_SELLER_SEND_GOODS", "PAID"]
# 数据校验阈值，超出即告警
alert:
min_orders: 5 # 单店订单数低于这个值，很可能拉取失败
max_amount_change: 3.0 # 单日销售额变化超过 3 倍，需要人工确认
notify:
webhook: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx"
```
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
### 4.2 统一拉数：三个平台的差异全在这里吃掉
```python
# fetch.py
import asyncio
import hashlib
import hmac
import time
from datetime import datetime, timedelta
from decimal import Decimal
from zoneinfo import ZoneInfo
import httpx
TZ = ZoneInfo("Asia/Shanghai")
def _taobao_sign(params: dict, secret: str) -> str:
keys = sorted(params)
raw = secret + "".join(f"{k}{params[k]}" for k in keys) + secret
return hashlib.md5(raw.encode()).hexdigest().upper()
def _alibaba_sign(params: dict, secret: str) -> str:
keys = sorted(params)
raw = "".join(f"{k}{params[k]}" for k in keys)
return hmac.new(secret.encode(), raw.encode(), hashlib.sha1).hexdigest().upper()
async def fetch_taobao(client, shop, start, end) -> list[dict]:
params = {
"method": "taobao.trades.sold.get",
"app_key": shop["app_key"],
"timestamp": str(int(time.time())),
"format": "json", "v": "2.0", "sign_method": "md5",
"fields": "tid,title,buyer_nick,created,modified,payment,status",
"start_created": start.strftime("%Y-%m-%d %H:%M:%S"),
"end_created": end.strftime("%Y-%m-%d %H:%M:%S"),
"page_no": 1, "page_size": 100,
}
params["sign"] = _taobao_sign(params, shop["app_secret"])
all_rows = []
page = 1
while True:
params["page_no"] = page
r = await client.post("https://eco.taobao.com/router/rest", data=params)
resp = r.json().get("trades_sold_get_response", {})
rows = (resp.get("trades") or {}).get("trade") or []
for t in rows:
all_rows.append({
"order_id": str(t["tid"]),
"shop_name": shop["name"],
"buyer_nick": t.get("buyer_nick", ""),
"created_at": t.get("created"),
"pay_amount": Decimal(str(t.get("payment", "0"))),
"status": t.get("status", ""),
"waybill": "",
})
if len(rows) < 100:
break
page += 1
return all_rows
async def fetch_alibaba(client, shop, start, end) -> list[dict]:
api = "/openapi/param2/1/com.alibaba.trade/alibaba.trade.get.soldList"
params = {
"access_token": shop["app_key"],
"createStartTime": start.strftime("%Y-%m-%d %H:%M:%S"),
"createEndTime": end.strftime("%Y-%m-%d %H:%M:%S"),
"pageNo": 1, "pageSize": 50,
}
all_rows = []
page = 1
while True:
params["pageNo"] = page
signed = dict(params)
signed["_aop_signature"] = _alibaba_sign(signed, shop["app_secret"])
r = await client.post(
f"https://gw.open.1688.com{api}/{shop['app_key']}", data=signed)
result = r.json().get("result") or {}
rows = result.get("orderList") or []
for o in rows:
all_rows.append({
"order_id": str(o.get("orderId")),
"shop_name": shop["name"],
"buyer_nick": (o.get("toLoginId") or ""),
"created_at": o.get("createTime"),
"pay_amount": Decimal(str(o.get("sumPayment") or "0")),
"status": o.get("orderStatus", ""),
"waybill": "",
})
if len(rows) < 50:
break
page += 1
return all_rows
async def fetch_jd(client, shop, start, end) -> list[dict]:
params = {
"method": "jingdong.pop.order.search",
"app_key": shop["app_key"],
"timestamp": start.strftime("%Y-%m-%d %H:%M:%S"),
"format": "json", "v": "2.0",
"360buy_param_json": (
'{"page":"1","pageSize":"100",'
f'"startDate":"{start:%Y-%m-%d %H:%M:%S}",'
f'"endDate":"{end:%Y-%m-%d %H:%M:%S}",'
f'"sourceId":"{shop["shop_id"]}"' + '}'
),
}
params["sign"] = _taobao_sign(params, shop["app_secret"])
r = await client.post("https://api.jd.com/routerjson", data=params)
resp = r.json().get("jingdong_pop_order_search_responce", {})
rows = (resp.get("searchorderinfo_result") or {}).get("orderInfoList") or []
return [{
"order_id": str(o.get("orderId")),
"shop_name": shop["name"],
"buyer_nick": o.get("pin", ""),
"created_at": o.get("orderStartTime"),
"pay_amount": Decimal(str(o.get("orderPayment") or "0")),
"status": str(o.get("orderState", "")),
"waybill": "",
} for o in rows]
FETCHERS = {"taobao": fetch_taobao, "alibaba": fetch_alibaba, "jd": fetch_jd}
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def fetch_all(shop, start, end) -> list[dict]:
async with httpx.AsyncClient(timeout=30) as client:
fn = FETCHERS[shop["platform"]]
try:
return await fn(client, shop, start, end)
except Exception as e:
print(f"[WARN] {shop['name']} 拉取失败: {e}")
return [] # 单店失败不影响其他店，但要记下来
```
注意最后那个 `try/except` 返回空列表。**单店失败不能拖垮整份报表**，但也不能装作没事——后面校验环节会发现订单数为 0 并告警。
### 4.3 清洗：所有「脚本跑两天就废」的坑都在这
```python
# clean.py
import pandas as pd
from datetime import datetime
from decimal import Decimal
from zoneinfo import ZoneInfo
TZ = ZoneInfo("Asia/Shanghai")
def to_frame(rows: list[dict]) -> pd.DataFrame:
if not rows:
return pd.DataFrame()
df = pd.DataFrame(rows)
# 1. 时间统一：有些平台返回字符串，有些返回时间戳
df["created_at"] = df["created_at"].apply(_norm_time)
# 2. 金额统一：Decimal 转 float 只在展示层做，计算用 Decimal
df["pay_amount"] = df["pay_amount"].apply(
lambda x: float(x) if isinstance(x, Decimal) else float(x or 0))
# 3. 编码：买家昵称里可能有表情和特殊字符
df["buyer_nick"] = df["buyer_nick"].astype(str).str.strip()
# 4. 去空白、统一订单号格式
df["order_id"] = df["order_id"].astype(str).str.strip()
return df
def _norm_time(v):
"""三平台时间格式归一：字符串 / 秒级时间戳 / 毫秒级时间戳"""
if v is None or v == "":
return pd.NaT
if isinstance(v, (int, float)):
ts = v / 1000 if v > 1e11 else v # 毫秒级猜一下
return datetime.fromtimestamp(ts, TZ).replace(tzinfo=None)
s = str(v).strip()
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S",
"%Y-%m-%d", "%Y/%m/%d %H:%M:%S"):
try:
return datetime.strptime(s, fmt)
except ValueError:
continue
return pd.to_datetime(s, errors="coerce")
def dedup(df: pd.DataFrame) -> pd.DataFrame:
"""跨平台去重：同一订单可能被重复拉到（重跑 / 重叠加窗）"""
if df.empty:
return df
return (df.sort_values("created_at")
.drop_duplicates(subset=["shop_name", "order_id"], keep="last"))
def validate(df: pd.DataFrame, config: dict) -> list[str]:
"""返回问题列表，空列表表示通过"""
issues = []
if df.empty:
return ["全部店铺拉取结果为空，请检查接口或授权"]
alert = config["rules"]["alert"]
by_shop = df.groupby("shop_name").agg(
cnt=("order_id", "count"), amt=("pay_amount", "sum"))
for shop_name, row in by_shop.iterrows():
if row["cnt"] < alert["min_orders"]:
issues.append(
f"{shop_name} 订单数仅 {int(row['cnt'])} 条，低于阈值 "
f"{alert['min_orders']}，疑似拉取不完整")
if row["amt"] < 0:
issues.append(f"{shop_name} 销售额为负（{row['amt']:.2f}），需核对退款")
return issues
```
`_norm_time` 这个函数看着不起眼，但它是「脚本跑两天就废」的头号原因。三个平台的时间格式分别是字符串、秒级时间戳、毫秒级时间戳，还有一种带 `T` 的 ISO 格式。不统一，后面所有按日期筛选、按天汇总全错。
### 4.4 输出 Excel：能直接发给人看的那种
```python
# report.py
from datetime import datetime
from pathlib import Path
import pandas as pd
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter
def build_report(df: pd.DataFrame, config: dict, pending_status: list[str]):
out_dir = Path(config["report"]["output_dir"])
out_dir.mkdir(parents=True, exist_ok=True)
date_str = datetime.now().strftime("%Y%m%d")
path = out_dir / config["report"]["filename"].format(date=date_str)
cols = config["report"]["columns"]
main = df[[c for c in cols if c in df.columns]].copy()
main = main.sort_values("created_at", ascending=False)
# 汇总页
summary = (df.groupby("shop_name")
.agg(订单数=("order_id", "count"),
销售额=("pay_amount", "sum"))
.reset_index())
summary["客单价"] = (summary["销售额"] / summary["订单数"]).round(2)
# 待发货页
pending = df[df["status"].isin(pending_status)][
[c for c in cols if c in df.columns]]
with pd.ExcelWriter(path, engine="openpyxl") as writer:
summary.to_excel(writer, sheet_name="汇总", index=False)
main.to_excel(writer, sheet_name="订单明细", index=False)
if not pending.empty:
pending.to_excel(writer, sheet_name="待发货", index=False)
_beautify(writer.book["汇总"])
_beautify(writer.book["订单明细"])
return path, summary
def _beautify(ws):
"""表头加粗+底色，列宽自适应，冻结首行"""
header_fill = PatternFill("solid", fgColor="E6F1FB")
for cell in ws[1]:
cell.font = Font(bold=True)
cell.fill = header_fill
cell.alignment = Alignment(horizontal="center", vertical="center")
for idx, col in enumerate(ws.columns, start=1):
width = max((len(str(c.value or "")) for c in col), default=8)
ws.column_dimensions[get_column_letter(idx)].width = min(40, width + 4)
ws.freeze_panes = "A2"
```
Excel 里有两个常见问题值得单独说：
**表头没冻结、列宽不自适应**，人拿到手的第一反应是「这什么玩意儿」。`_beautify` 十几行就解决了，成本极低，但决定了别人愿不愿意看你的报表。
**文件被占用导致写入失败**。人正在打开昨天的报表，脚本今天写同名文件就会 `PermissionError`。解法是文件名带日期（上面已经做了），再加一个重试兜底：
```python
def safe_write(build_fn, path: Path, retries: int = 3):
"""Excel 被占用时重试，实在不行换个名字存"""
import time
for i in range(retries):
try:
return build_fn(path)
except PermissionError:
if i == retries - 1:
alt = path.with_name(f"{path.stem}_v{i}{path.suffix}")
print(f"[WARN] 文件被占用，改存为 {alt.name}")
return build_fn(alt)
time.sleep(2)
```
### 4.5 通知 + 定时
```python
# notify.py
import httpx
import pandas as pd
from datetime import datetime
async def send_wecom(webhook: str, summary: pd.DataFrame, path, issues: list[str]):
lines = [f"**经营日报** {datetime.now():%Y-%m-%d}", ""]
for _, r in summary.iterrows():
lines.append(f"{r['shop_name']}：{int(r['订单数'])} 单 / ¥{r['销售额']:,.2f}")
lines.append(f"\n合计：{int(summary['订单数'].sum())} 单 / "
f"¥{summary['销售额'].sum():,.2f}")
if issues:
lines.append("\n**数据校验未通过**")
lines.extend(f"- {i}" for i in issues)
lines.append(f"\n报表：{path.name}")
async with httpx.AsyncClient(timeout=10) as client:
await client.post(webhook, json={
"msgtype": "markdown",
"markdown": {"content": "\n".join(lines)},
})
```
定时部分，Windows 用任务计划、Linux 用 cron：
```bash
# 每天早上 8:00 跑
0 8 * * * cd /opt/report && /usr/bin/python3 main.py >> logs/$(date +\%F).log 2>&1
```
```python
# main.py
import asyncio
import yaml
from datetime import datetime, timedelta
from fetch import fetch_all
from clean import to_frame, dedup, validate
from report import build_report
from notify import send_wecom
async def run():
config = yaml.safe_load(open("config.yaml", encoding="utf-8"))
# 拉昨天 00:00 到今天 00:00
today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
start, end = today - timedelta(days=1), today
all_rows = []
for shop in config["shops"]:
rows = await fetch_all(shop, start, end)
print(f"[{shop['name']}] {len(rows)} 条")
all_rows.extend(rows)
df = dedup(to_frame(all_rows))
issues = validate(df, config)
path, summary = build_report(
df, config, config["rules"]["pending_ship_status"])
print(f"报表已生成: {path}")
await send_wecom(config["notify"]["webhook"], summary, path, issues)
if issues:
raise SystemExit(1) # 让任务计划标记为失败，便于外部监控
if __name__ == "__main__":
asyncio.run(run())
```
最后那个 `raise SystemExit(1)` 值得说一下。有问题的运行必须以非零退出码结束，这样任务计划或监控系统才能发现。很多人脚本里 `print` 一句「数据异常」就算完事，结果谁也没看见。
## 五、收益
一个 3 个店、日均 300 单的团队：
| 指标 | 手动 | 自动 |
| ----------------- | -------------- | ------------- |
| 每天早上耗时 | 约 2 小时 | 0（8:00 自动出结果） |
| 数据出错 | 每月 2~3 次（粘贴错位） | 0 |
| 发现问题的时间 | 月底对账才发现 | 当天 8:00 告警 |
| 临时要数据（比如「上周三的明细」） | 重新翻后台，20 分钟 | 改个日期重跑，30 秒 |
第二行和第三行才是真正的价值。**省时间只是附带，减少的出错和提前发现才是关键。**
## 六、让脚本活过三个月的六个细节
这段比代码重要，都是踩出来的：
1. **时区写死，不要用服务器默认。** 服务器如果不是东八区，跨零点的订单会错到前一天。
2. **金额用 Decimal 到最后一步才转 float。** 汇总时用 float 累加，几百行能差出几分钱，财务一眼就看出来。
3. **文件名带日期。** 避免文件被占用，也方便回溯。别用「经营日报.xlsx」这种固定名。
4. **单店失败不中断。** 三个店有一个授权过期，不能整份报表都不出。
5. **一定要有数据校验。** 订单数为 0、销售额突变、订单号重复，都是典型的失败信号。没有校验的自动化，等于把错误也自动化了。
6. **退出码要正确。** 出问题就 `exit 1`，别默默跑完。监控系统只认退出码。
## 七、最后
这套东西的价值不在代码有多少行，在于**它把两小时压缩成了三十秒，而且不会出错**。
代码在上面，改 `config.yaml` 里的 appkey 就能跑。三个平台的签名、分页、时间格式差异都处理好了。
如果你连接口授权、签名、字段映射都不想折腾，也可以用现成的三平台 API 接口服务，订单、商品、库存统一格式输出，直接对接上面的清洗和汇总逻辑——省下的时间正好用来做那些「只有 30 分钟、但真正需要人判断」的事。
有在做的可以聊聊，你们的日报是怎么发的，还在手动贴 Excel 吗？
---
**标签**：`Python` `电商` `自动化` `Excel` `数据分析`
