Предлагаем вашему вниманию интересную сп...
注册账号免费测试1688API接口1688 的 item_review 接口(官方标准方法名:alibaba.item.review.get)是阿里巴巴开放平台提供的核心接口之一,用于获取指定商品的买家评论数据,包括文本评价、评分、晒图、采购属性(如采购数量、...
还有 85 人发表了评论 加入10773人围观注册账号免费测试1688API接口1688 的 item_review 接口(官方标准方法名:alibaba.item.review.get)是阿里巴巴开放平台提供的核心接口之一,用于获取指定商品的买家评论数据,包括文本评价、评分、晒图、采购属性(如采购数量、...
还有 85 人发表了评论 加入10773人围观# 做 ERP 的看过来:三平台的商品订单接口,不用再一家家啃文档(附 Python 源码)
> 标题里「不用再一家家啃文档」要诚实地拆开说:**文档躲不掉,但可以从 300 页降到每平台 30 页**。
>
> 降法不是划重点,是换读法:ERP 只该面对一套「动作」,三个平台怎么实现这些动作,是适配器的事。你真正要啃的只有**差异**——而差异是可以列出来的,这篇就把它列全。
>
> 先划清和之前那篇 ERP 文(第 15 篇)的分工:那篇讲数据进了 ERP 之后怎么消化(金额拆分、账期、状态映射、超时窗口),全是账务语义;这篇只讲接口层——怎么把三个平台的东西拉进来、推出去,以及 ERP 场景特有的那道门槛:**多店铺**。
## 一、ERP 对接的真实工作量:不是学三个 SDK,是管理差异
先破一个误区。多数 ERP 团队评估对接工期时算的是「学三个平台的开发文档」,这不是主要成本。主要成本是**同一个业务动作在三个平台的实现差异**,而且这些差异文档里不会并排放着,你要自己发现。
按 ERP 最关心的五个域,把差异摊开:
| 域 | 方向 | 增量参数差异 | 分页差异 | 最贵的坑 |
| -- | ------ | --------------------------------------------- | ---------------------- | --------------------------------------- |
| 商品 | 下行为主 | 淘宝按 `modified`,1688 按更新时间,京东部分接口**只支持按天** | 三平台游标/页码不一,京东有**分页漂移** | 只同步 SPU 不同步 SKU(第 1 篇的老坑,ERP 里表现为库存对不上) |
| 订单 | 下行为主 | 淘宝按 `modified`,京东**按天且建议双时间字段**,1688 窗口重叠要自己做 | 页码式,深翻页丢数据 | 用创建时间不支付时间做增量 → 跨月订单归错账期(第 15 篇) |
| 库存 | **双向** | 无增量可言,靠推送 + 定时校准 | — | 上行回写是**写操作**:覆盖式同步会把 ERP 里的预占抹掉(第 3 篇) |
| 物流 | 上行为主 | — | — | 发货回写错单号 = **真发货事故**,这是 ERP 所有接口里责任最重的一个 |
| 售后 | 下行为主 | 三平台退款状态机完全不同(淘宝 7 态、京东 6 态、1688 退款分层级) | 页码式 | 仅退款 / 部分退款 / 退货入库成本重算,接口能拉到但语义要自己拼 |
方向那一列值得多看一眼:**ERP 不是「拉数据」,是双向同步**。下行占接口数量的 80%,但上行(库存回写、发货回写)占责任的 80%——下行拉错数据顶多报表难看,上行推错是真实的发货事故和库存事故。
## 二、统一动作层:ERP 只面对 12 个动作
解法就是第 8 篇那套适配器架构,但 ERP 版要按 ERP 的动作习惯重新切。我们把三平台的所有接口收敛成 12 个动作:
```
下行(拉取) 上行(推送)
item.pull 商品资料 stock.push 库存回写
item_sku.pull SKU 明细 ship.push 发货回写
buyer.pull 收货人信息 invoice.push 开票回写
order.pull 订单 refund.push 售后审核回写
stock.pull 库存快照 memo.push 卖家备注回写
refund.pull 售后单
logistics.pull 物流轨迹
```
两个纪律:
**① 动作是 ERP 定义的,不是平台定义的。** 平台的接口按它自己的业务切(淘宝一个「订单」接口里混着交易和履约信息),ERP 按自己的单据流切。适配器的活就是把平台切法翻译成 ERP 切法——**翻译层一旦出现 `if 订单金额 > 100` 这种业务判断,说明业务逻辑漏进来了**(第 8 篇的原话,ERP 场景下更常见,因为 ERP 的诱惑就是「顺手在适配层里算个账」)。
**② 能力声明必须显式。** 三个平台不是每个动作都支持。1688 没有按 SKU 的库存概念(按起订量阶梯),京东开票回写的权限包单独申请。**假装都支持是最贵的错误**:调用方传 sku_id,适配器默默返回了商品总库存,不报错,三个月后客户对不上账才被发现。
启动自检(第 8 篇的 `check_capabilities`,ERP 版):
```python
REQUIRED = {
"taobao": ["item.pull", "order.pull", "stock.push", "ship.push", "refund.pull"],
"alibaba": ["item.pull", "order.pull", "stock.pull", "ship.push"], # 没有 stock.push
"jd": ["item.pull", "order.pull", "stock.push", "ship.push", "refund.pull"],
}
def check_capabilities(shops: list["Shop"]) -> None:
for shop in shops:
missing = [a for a in REQUIRED[shop.platform]
if a not in CAPABILITIES[shop.platform]]
if missing:
# 缺能力是配置错误,启动就该死,别等客户对账时发现
raise SystemExit(f"{shop.name}({shop.platform}) 缺动作: {missing}")
```
注意 1688 那行的写法:`REQUIRED` 里**不包含** `stock.push`,`CAPABILITIES` 里也没有——它不是缺失,是**不存在**。这两份清单的区别就是「没申请权限」和「平台没这个概念」,混着写会误导后来的人去申请一个不存在的权限包。
## 三、代码:动作路由 + 三平台订单增量 + 发货幂等
`pip install httpx` 即可。代码按 ERP 最疼的两个动作给全:`order.pull`(下行最复杂)和 `ship.push`(上行责任最重)。
### 3.1 统一模型与动作路由
```python
# erp/models.py
from __future__ import annotations
from dataclasses import dataclass, field
from decimal import Decimal
@dataclass
class OrderLine:
sku_id: str
title: str
qty: int
price: Decimal # 单价,永远 Decimal(第 4 篇:float 累加对不上账)
pay_amount: Decimal # 实付,分摊后的行金额(分摊算法见第 15 篇)
@dataclass
class Order:
platform: str
order_id: str # 平台单号
shop_id: str # 多店铺:这单属于哪家店(见第五节)
status: str # 归一后的 ERP 状态,不是平台原始状态
paid_at: str | None # 支付时间。ERP 按这个记账期,不是 created_at
lines: list[OrderLine] = field(default_factory=list)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class NotSupported(Exception):
"""显式报错 > 静默返回错数据。允许调用方 catch 后降级,不允许适配器假装成功"""
```
```python
# erp/base.py
from abc import ABC, abstractmethod
from erp.models import Order, NotSupported
CAPABILITIES = {
"taobao": {"item.pull", "item_sku.pull", "buyer.pull", "order.pull",
"stock.pull", "stock.push", "ship.push", "memo.push", "refund.pull"},
"alibaba": {"item.pull", "item_sku.pull", "order.pull",
"stock.pull", "ship.push"},
"jd": {"item.pull", "item_sku.pull", "buyer.pull", "order.pull",
"stock.pull", "stock.push", "ship.push",
"invoice.push", "refund.push", "refund.pull"},
}
class Adapter(ABC):
platform: str
def __init__(self, shop):
self.shop = shop
def can(self, action: str) -> bool:
return action in CAPABILITIES[self.platform]
@abstractmethod
async def order_pull(self, since: int, cursor: dict) -> tuple[list[Order], dict]:
"""返回 (订单列表, 游标)。游标是 opaque 的:调用方不解析,原样存取"""
async def query_order(self, order_id: str) -> Order:
"""ship.push 前的状态复核用,各平台自行实现"""
raise NotImplementedError
async def do_ship_push(self, order_id: str, waybill: str,
company: str, idem_key: str) -> str:
"""实际回写调用。idem_key 作为平台幂等键传入"""
raise NotImplementedError
async def ship_push(self, order_id: str, waybill: str, company: str) -> str:
raise NotSupported(f"{self.platform} 未开通 ship.push")
```
游标设计成 opaque 的 dict 是刻意的:**调用方不许解析游标内容**。三个平台的游标内部结构完全不同(淘宝存 `modified` 时间戳,京东要存日期 + 页码,1688 存 `last_id`),如果调用方解析了游标,适配器内部就再也不能改增量策略了。
### 3.2 三平台 `order.pull`:差异全在这三个类里
```python
# erp/adapters.py
import asyncio
import time
import httpx
from erp.base import Adapter
from erp.models import Order, OrderLine, NotSupported
from decimal import Decimal
STATUS_MAP = {
# 平台原始状态 → ERP 归一状态。映射表放适配器里(是翻译,不是业务)
"taobao": {"TRADE_FINISHED": "DONE", "WAIT_SELLER_SEND_GOODS": "TO_SHIP"},
"jd": {"FINISHED": "DONE", "WAIT_DELIVERY": "TO_SHIP"},
"alibaba": {"success": "DONE", "waitbuyerpay": "UNPAID"},
}
class TaobaoAdapter(Adapter):
platform = "taobao"
async def order_pull(self, since: int, cursor: dict) -> tuple[list[Order], dict]:
# 游标:modified 时间戳 + last_id 双字段(秒级精度一秒多单会撞,第 3 篇)
modified = cursor.get("modified", since)
last_id = cursor.get("last_id", "")
orders: list[Order] = []
last_row = None
async with httpx.AsyncClient(timeout=10) as client:
while True:
r = await client.get("/v1/taobao/trades", params={
"modified_start": modified,
"page_size": 100,
# 重叠加窗 60 秒:宁可重复不可遗漏(重复靠幂等吸收)
})
r.raise_for_status()
body = r.json()
rows = body.get("trades", [])
for row in rows:
key = f"{row['modified']}:{row['tid']}"
if key <= last_id:
continue # 双键排序,跳过已处理的
orders.append(self._to_order(row))
last_row = row
if not body.get("has_next"):
break
# 游标只推进到最后一条已处理的:结构对调用方 opaque
if last_row:
cursor = {"modified": last_row["modified"],
"last_id": f"{last_row['modified']}:{last_row['tid']}"}
return orders, cursor
def _to_order(self, row: dict) -> Order:
# 淘宝 HTTP 永远 200,错误藏在 error_response(第 11 篇)——网关已剥过一层
return Order(
platform=self.platform,
order_id=str(row["tid"]),
shop_id=self.shop.id,
status=STATUS_MAP[self.platform].get(row["status"], "UNKNOWN"),
paid_at=row.get("pay_time"),
lines=[OrderLine(
sku_id=str(s["sku_id"]),
title=s["title"],
qty=int(s["num"]),
price=Decimal(str(s["price"])),
pay_amount=Decimal(str(s["payment"])),
) for s in row.get("orders", [])],
)
class JdAdapter(Adapter):
platform = "jd"
async def order_pull(self, since: int, cursor: dict) -> tuple[list[Order], dict]:
# 京东部分订单接口只支持按天(第 3 篇的硬约束):
# 增量单位从「小时」降级为「天」,用 last_modified + 去重吸收重复
day = cursor.get("day") or time.strftime("%Y-%m-%d", time.localtime(since - 86400))
orders: list[Order] = []
async with httpx.AsyncClient(timeout=10) as client:
r = await client.post("/v1/jd/order_search", content=_jd_param_json({
"startDate": day, "endDate": day, "page": cursor.get("page", 1),
}))
r.raise_for_status()
body = jd_unwrap(r.json()) # 按接口名剥壳,key 写法还不统一(第 11 篇)
for row in body.get("orderList", []):
orders.append(self._to_order(row))
# 按天拉:游标只推进「日期」,重复行由调用方按 order_id 幂等吸收
return orders, {"day": day, "page": cursor.get("page", 1) + 1}
def _jd_param_json(payload: dict) -> str:
# 京东的坑:360buy_param_json 里的中文不能被 \uXXXX 转义,
# 签名按原文算(ensure_ascii=False,第 11 篇那个卡一天半的坑)
import json
return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
def jd_unwrap(body: dict) -> dict:
for k, v in body.items():
if isinstance(v, dict) and "result" in v:
return v["result"]
return body
class AlibabaAdapter(Adapter):
platform = "alibaba"
async def order_pull(self, since: int, cursor: dict) -> tuple[list[Order], dict]:
last_id = cursor.get("last_id", 0)
orders: list[Order] = []
async with httpx.AsyncClient(timeout=10) as client:
r = await client.get("/v1/alibaba/trades", params={
"last_id": last_id, "size": 50})
r.raise_for_status()
for row in r.json().get("result", []):
orders.append(self._to_order(row))
return orders, {"last_id": max((o.order_id for o in orders), default=last_id)}
def _to_order(self, row):
return Order(
platform=self.platform, order_id=str(row["id"]), shop_id=self.shop.id,
status=STATUS_MAP[self.platform].get(row["status"], "UNKNOWN"),
paid_at=row.get("payTime"),
lines=[OrderLine(sku_id=str(s["productID"]), title=s["name"],
qty=int(s["itemCount"]), price=Decimal(str(s["price"])),
pay_amount=Decimal(str(s["itemTotal"])))
for s in row.get("orderEntrys", [])],
)
```
这段代码里每个类的写法都不一样,**这是特性不是缺陷**:淘宝是「时间游标 + 双键排序」,京东是「按天 + 幂等吸收」,1688 是「ID 游标」。差异被关在三个类里,调用方永远只看到 `order_pull(since, cursor) -> (orders, cursor)`。
### 3.3 `ship.push`:责任最重的一个动作
```python
# erp/ship.py
import hashlib
from erp.base import Adapter
from erp.models import NotSupported
# 模块级连接,进程启动时初始化(结构见 store 的 SCHEMA 思路,第 20 篇)
db = None # type: sqlite3.Connection
SCHEMA_SHIP = """
create table if not exists ship_push(
idem_key text primary key, platform text, order_id text,
waybill text, result text, pushed_at integer);
"""
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def _already_pushed(idem: str) -> bool:
return db.execute("select 1 from ship_push where idem_key=?",
(idem,)).fetchone() is not None
async def _mark_pushed(idem: str, adapter: Adapter, order_id: str,
waybill: str, result: str):
import time
db.execute(
"insert or ignore into ship_push values(?,?,?,?,?,?)",
(idem, adapter.platform, order_id, waybill, result, int(time.time())))
db.commit()
async def ship_push(adapter: Adapter, order_id: str, waybill: str, company: str) -> str:
"""发货回写。三重防呆,缺一个都可能造成真发货事故:
1. 本地幂等键:同一单号+运单号只发一次(重试安全)
2. 先查后推:本地未确认发货的才允许推
3. 平台侧幂等:用 order_id 做平台幂等键,重复推送返回首次结果
"""
idem = hashlib.sha1(f"{adapter.platform}:{order_id}:{waybill}".encode()).hexdigest()
if await _already_pushed(idem):
return "DUPLICATE"
if not adapter.can("ship.push"):
raise NotSupported(f"{adapter.platform} 未开通 ship.push")
# 先查后推:ERP 状态推进和平台状态之间有时间差,推之前再确认一次
remote = await adapter.query_order(order_id)
if remote.status not in ("TO_SHIP",):
raise RuntimeError(f"平台状态 {remote.status} 不可发货,拒绝回写")
result = await adapter.do_ship_push(order_id, waybill, company, idem_key=idem)
await _mark_pushed(idem, adapter, order_id, waybill, result)
return result
```
为什么发货回写要三重防呆,库存回写只要两重?因为**发货回写触发的是物理世界**:平台收到回写就真的安排仓库发货了。重试导致的重复发货、错状态导致的提前发货,都是客户仓库里的实际损失——这类接口的容错原则和下行完全相反:**下行情愿重复,上行情愿少做**。
## 四、多店铺:ERP 的第一道门槛,别到最后才处理
这是 ERP 和自用脚本最本质的区别:**ERP 一个实例服务几十家客户店**。前十九篇里的「店铺」是单数,ERP 里必须是复数。这一节把多店铺的地基讲了,深度展开(存储加密、几十家店并发刷新的工程细节)留给下一篇。
三件事必须从第一天做对:
### 4.1 token 按店铺隔离,谁的 token 只能查谁的数据
第 14 篇讲过的最隐蔽的死法在多店铺下是**必死**:混淆 app 级与用户级凭证,接口全绿,拉到的是别人店铺的数据。ERP 里这不叫 bug,叫**事故**——客户 A 的订单进了客户 B 的库。
```python
# erp/shops.py
import asyncio
import time
import sqlite3
import httpx
class Shop:
def __init__(self, shop_id: str, platform: str, name: str):
self.id = shop_id
self.platform = platform
self.name = name
self.user_key = "" # 店铺级授权凭证,绝不入库表名混用
self._tokens: dict[str, tuple[str, float]] = {} # platform -> (token, expire)
self._refresh_lock = asyncio.Lock() # 单飞锁,per shop
self.quarantined_until = 0.0 # 过期隔离(见 4.3)
async def token(self, app_key: str, secret: str) -> str:
tok = self._tokens.get(self.platform)
if tok and tok[1] > time.time() + 300:
return tok[0]
async with self._refresh_lock: # 双检:拿到锁再查一次
tok = self._tokens.get(self.platform)
if tok and tok[1] > time.time() + 300:
return tok[0]
new_tok, expires_in = await self._do_refresh(app_key, secret)
self._tokens[self.platform] = (new_tok, time.time() + expires_in)
return new_tok
```
`_refresh_lock` 是 **per shop** 的,这是多店铺最容易犯的错:用一把全局锁,店铺 A 刷新时店铺 B 的正常调用也跟着排队;或者反过来,没有锁,同一家店两个协程同时刷新——京东和 1688 都是「后刷作废先刷」,两个协程会把自己刷死自己。
### 4.2 授权链路:ERP 走「他用型(ISV)」应用 + 商家授权
第 17 篇讲过应用类型决定权限天花板,ERP 场景再补三层:
- **ERP 是 ISV 应用**,自用型 appkey 根本申请不到多店铺需要的权限包。类型选错要重建应用重审,**这个决定要在写第一行代码之前做**。
- 每家客户店要**店主自己扫码授权**(生成各自的 session/token),ERP 拿的是「店铺级」凭证,不是平台级。客户解绑通常无通知,token 会悄悄失效——靠每日摘要(第 20 篇那套)或每店心跳发现。
- 给客户的说辞要按第 17 篇的场景描述写法来:「系统为该商家管理其自有店铺的订单与库存」,别出现「聚合」「采集」。
### 4.3 单店过期隔离:一家店挂了,不能炸全部
```python
# erp/runner.py
import time
async def run_round(shops: list, adapters: dict, db):
"""一轮同步。核心纪律:单店失败隔离,绝不 raise 到顶层"""
for shop in shops:
if time.time() < shop.quarantined_until:
continue # 隔离中,跳过,等摘要邮件喊人
adapter = adapters[shop.platform](shop)
try:
cursor = load_cursor(db, shop.id, "order.pull")
orders, cursor = await adapter.order_pull(since=last_round(db), cursor=cursor)
save_orders(db, orders) # 按 order_id 幂等 upsert
save_cursor(db, shop.id, "order.pull", cursor)
except Exception as e:
# 连续失败才隔离:单次失败可能只是网络抖动
shop.fail_count = getattr(shop, "fail_count", 0) + 1
if shop.fail_count >= 3:
# 401/403 大概率是客户解绑或 token 过期 → 隔离 1 小时
shop.quarantined_until = time.time() + 3600
record_failure(db, shop.id, repr(e)) # 进每日摘要
# 循环正常走完:8 家店,1 家授权过期,其余 7 家照常出数据
```
隔离的判断要区分**错误类型**:401/403(授权失效)隔离一小时等客户重新授权;429(限流)降速不隔离;500 连续三次也隔离。**把「哪家店失败了、为什么、隔离到几点」写进每日摘要**,实施顾问早上看一眼就知道今天该找哪家客户重新扫码。
## 五、给 ERP 厂商的实施清单
最后是一份可以贴在项目文档里的清单,按实施顺序:
**售前(决定做不做)**
- [ ] 应用类型确认为 ISV/他用型(选错要重建重审)
- [ ] 五个域 × 三平台的能力矩阵过一遍,缺失动作(如 1688 无 stock.push)已与产品确认替代方案
- [ ] 客户的行业类目在三个平台都有对应权限包(特殊类目权限收紧,第 17 篇)
**开发(决定做多久)**
- [ ] 只啃差异:鉴权一节 + 增量参数一节 + 五个域的接口页,每平台约 30 页
- [ ] 游标 opaque,调用方不解析
- [ ] 上行三重防呆(幂等键 / 先查后推 / 平台幂等)
- [ ] 状态映射表放适配器,业务规则不进翻译层
**运维(决定活多久)**
- [ ] token per shop 隔离 + 单飞锁
- [ ] 单店失败隔离 + 每日摘要含「失败店铺清单」
- [ ] `solo.db`/token 库 `chmod 600`,绝不进 git
---
顺带说一句给 ERP 厂商的账:这套适配层写完是一次性资产,但**它的维护成本是按客户数摊的**——一家客户分摊到的维护费可能撑不起一个全职人力,这就是为什么不少 ERP 团队最后把「平台对接」这一层整体外置(买接口服务,自己只做动作层以上的业务)。算不算得过账,取决于你的客户数,公式和第 12 篇那套 TCO 一样,把「人力」换成「实施人力」就行。
下一篇写多店铺授权的工程细节:token 落库加密、几十家店并发刷新的撞车处理、批量重新授权的引导流程。那是 ERP 接入的最后一坑。# 写给独立开发者:一个人怎么把三平台数据接口接完(附 Python 源码)
> 先把边界划清:这篇和之前那篇「3 天接通三平台」不是一回事。
>
> 那篇默认你有一个小团队、有预算、有人分工。这篇的第一约束只有一条:**没有第二个人**。你病了没人值班,你忙别的项目时没人看日志,三个月后你自己都会忘记这个服务还活着。
>
> 所以单人版的做法和团队版大量相反。团队版里的最佳实践——事件驱动、统一抽象、熔断降级——对一个人来说,多数是负债:**它们都要花维护意愿,而你的维护意愿是最稀缺的资源**。
>
> 下面按「留什么、砍什么、作弊什么、什么时候放弃」四步讲,代码是一个完整可跑的单人版骨架。
## 一、先算一笔账:一个人接,到底要多少小时
团队版的账(第 6 篇算过)是 13 天。单人版不能按天算,要按小时算——因为你的时间是从晚上和周末里抠的。
诚实的拆法:
| 模块 | 小时 | 说明 |
| -------------------- | ------ | --------------------------------- |
| 签名 + token 刷新 | 6 | 三平台三套签名,1688 的 urlPath 参与签名最容易卡半天 |
| 限流 + HTTP 客户端 | 4 | 进程内令牌桶,见下文,不用 Redis |
| 增量拉取 + 指纹比对 | 8 | 时间窗 + 指纹,SQLite 存 |
| 失败重放 + 每日摘要 | 6 | 单人版的命根子,后面细讲 |
| 第二个平台 | 8 | 第一个平台摸熟后,主要是字段差异 |
| 第三个平台 | 6 | 同上 |
| 自愈测试(拔网线、改错 key、杀进程) | 2 | 这 2 小时不能省 |
| **合计** | **40** | ≈ 5 个周末,或 2~3 周的晚上 |
注意这是**开发**。维护另算:每月 1~~2 小时(平台改字段、授权过期、配额规则变化),一年 12~~24 小时。这笔账第五节要用。
## 二、单人版架构:留什么、砍什么
对照团队版逐项过。原则只有一条:**单人无法持续维护的复杂度,一律砍掉**。
| 团队版做法 | 单人版 | 为什么 |
| -------------------- | --------------------------- | ----------------------------------------------------- |
| TMC 消息推送(事件驱动) | **轮询 5~10 分钟 + 时间窗** | 推送要维护长连接、断线重连、消息确认。轮询简单粗暴,对独立产品够用。规模上去再换,反着来一次的成本是可控的 |
| 三平台统一适配层(第 8 篇) | **先直连一个平台,第二个平台出现重复代码时再抽象** | 统一层是给维护团队的投资;单人先抽象等于为还不存在的需求写代码。规矩是老的:事不过三 |
| Redis 分布式令牌桶(第 10 篇) | **进程内令牌桶** | 你只有一个进程。少一个 Redis,少一类故障,少一笔服务器钱 |
| 舱壁隔离 + 熔断降级(第 9 篇) | **超时 + 失败进重放表** | 熔断的价值是保护其他业务线;你只有一条业务线,它挂了你反正要修 |
| 监控告警系统 | **每天一封摘要邮件** | 告警系统要配、要调阈值、要有人响应。摘要邮件零维护,早上看一眼 |
| 微服务 / 消息队列 | **单进程 + cron** | 一个进程,一个库,一个定时器。部署 = 一个 systemd 单元 |
| MySQL / PG + 运维 | **SQLite(WAL 模式)** | 零安装、零运维、备份就是拷一个文件。单人规模的量它完全扛得住 |
三条单人的铁律,比上面这张表更重要:
1. **一切设计围绕「失联自愈」**。你一定会有一周忘了它的存在。挂掉的轮次要能自己补回来(重放表),而不是等你发现。
2. **宁可重复,不可抽象**。两个平台的相似代码各写各的,等第三个出现再合并。单人项目里,错误的抽象比重复代码贵十倍——因为改错抽象的人只有你。
3. **归一化模型只做你需要的字段**。第 2 篇那套统一 Item/Sku 模型是给多消费者系统用的。你的小工具只需要展示 6 个字段,那就只映射 6 个。
## 三、代码:一个进程、一个库、五个文件
`pip install httpx` 之外零依赖。目录结构:
```
solo/
├── token_box.py # token 刷新,双检
├── gate.py # 进程内令牌桶
├── store.py # SQLite:指纹表 + 重放表
├── main.py # 每小时一轮,cron 调
└── digest.py # 每日摘要邮件
```
### 3.1 token 刷新:双检 + 提前量
第 14 篇的 `TokenManager` 用的是分布式锁——那是多实例场景。单进程用线程锁 + 双检就够了,逻辑一个类讲完:
```python
# solo/token_box.py
import time
import threading
import sqlite3
import httpx
REFRESH_AHEAD = 300 # 过期前 5 分钟就刷,别卡着点
class TokenBox:
"""单进程版 token 管理。SQLite 存 + 线程锁双检。
注意:多实例部署时这套不够,要用分布式锁(见第 14 篇)。"""
def __init__(self, db: sqlite3.Connection, platform: str,
app_key: str, secret: str, refresh_url: str):
self.db = db
self.platform = platform
self.app_key = app_key
self.secret = secret
self.refresh_url = refresh_url
self._lock = threading.Lock()
db.execute(
"create table if not exists tokens("
"platform text primary key, access_token text, expire_at real)")
def get(self) -> str:
row = self.db.execute(
"select access_token, expire_at from tokens where platform=?",
(self.platform,)).fetchone()
if row and row[1] > time.time() + REFRESH_AHEAD:
return row[0]
with self._lock: # 拿到锁后再查一次,防止两个协程轮流刷
row = self.db.execute(
"select access_token, expire_at from tokens where platform=?",
(self.platform,)).fetchone()
if row and row[1] > time.time() + REFRESH_AHEAD:
return row[0]
token, expires_in = self._refresh()
self.db.execute(
"insert into tokens(platform, access_token, expire_at) "
"values(?,?,?) on conflict(platform) do update set "
"access_token=excluded.access_token, expire_at=excluded.expire_at",
(self.platform, token, time.time() + expires_in))
self.db.commit()
return token
def _refresh(self) -> tuple[str, int]:
r = httpx.post(self.refresh_url, data={
"app_key": self.app_key, "secret": self.secret})
r.raise_for_status()
body = r.json()
# 京东和 1688 都有「后刷作废先刷」的行为,
# 刷新失败时别把旧 token 删了——还能用就先用着
return body["access_token"], int(body.get("expires_in", 86400))
```
最后那段注释是单人版特有的容忍:**刷新失败时保留旧 token 继续用**。团队版会立刻告警,你没空响应,那就让旧的多活一会儿,大部分情况它真的还能用。
### 3.2 限流:进程内令牌桶
第 10 篇的 Redis 版是给多实例共享额度用的。单进程,一个字典就够——**少一个依赖,少一类故障**:
```python
# solo/gate.py
import time
import threading
class Gate:
"""进程内令牌桶,按 '平台:接口' 分桶(第 10 篇讲过为什么必须分桶:
平台的限流维度是 接口 × 应用,不是全局)"""
def __init__(self):
self._buckets: dict[str, tuple[float, float]] = {} # key -> (tokens, last)
self._lock = threading.Lock()
def acquire(self, key: str, rate: float, burst: int | None = None):
burst = burst or max(2, int(rate * 2))
while True:
with self._lock:
now = time.monotonic()
tokens, last = self._buckets.get(key, (float(burst), now))
tokens = min(float(burst), tokens + (now - last) * rate)
if tokens >= 1:
self._buckets[key] = (tokens - 1, now)
return
wait = (1 - tokens) / rate
time.sleep(wait)
```
默认 `burst = rate × 2` 是保守起点——第 10 篇里 `burst=500` 撞 429 的教训,单人版更要远离。你没有人手盯限流告警。
### 3.3 存储:SQLite + WAL + 重放表
整个单人版的灵魂在 `replay` 表。**它替代的是团队版的值班制度**:
```python
# solo/store.py
import hashlib
import json
import time
import sqlite3
SCHEMA = """
pragma journal_mode = wal; -- 读写不互斥,单人版最关键的一行配置
create table if not exists items(
platform text not null,
item_id text not null,
fingerprint text not null,
payload text not null,
updated_at integer not null,
primary key(platform, item_id)
);
create table if not exists replay(
id integer primary key autoincrement,
platform text not null,
kind text not null,
params text not null, -- json
next_try_at integer not null,
tries integer not null default 0,
last_error text
);
"""
def open_db(path: str) -> sqlite3.Connection:
db = sqlite3.connect(path)
db.executescript(SCHEMA)
return db
def fingerprint(payload: dict) -> str:
"""指纹比对(第 1 篇的思路):变了才写库,省掉 90% 的无效写"""
keys = ("price", "stock", "title", "status") # 只盯你关心的字段
raw = json.dumps({k: payload.get(k) for k in keys},
ensure_ascii=False, sort_keys=True)
return hashlib.sha1(raw.encode()).hexdigest()
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def upsert_if_changed(db, platform, item_id, payload) -> bool:
fp = fingerprint(payload)
row = db.execute(
"select fingerprint from items where platform=? and item_id=?",
(platform, str(item_id))).fetchone()
if row and row[0] == fp:
return False
db.execute(
"insert into items(platform,item_id,fingerprint,payload,updated_at) "
"values(?,?,?,?,?) on conflict(platform,item_id) do update set "
"fingerprint=excluded.fingerprint, payload=excluded.payload, "
"updated_at=excluded.updated_at",
(platform, str(item_id), fp, json.dumps(payload, ensure_ascii=False),
int(time.time())))
return True
def add_replay(db, platform, kind, params, last_error: str):
"""失败的拉取进重放表,指数退避:1h → 2h → 4h → 封顶 12h。
已有同任务行则原地更新,别插入新行——重试路径会反复走到这里"""
p = json.dumps(params, sort_keys=True)
row = db.execute(
"select tries from replay where platform=? and kind=? and params=?",
(platform, kind, p)).fetchone()
tries = (row[0] if row else 0) + 1
delay = min(3600 * 2 ** (tries - 1), 3600 * 12)
if row:
db.execute(
"update replay set next_try_at=?, tries=?, last_error=? "
"where platform=? and kind=? and params=?",
(int(time.time()) + delay, tries, last_error[:300],
platform, kind, p))
else:
db.execute(
"insert into replay(platform,kind,params,next_try_at,tries,last_error) "
"values(?,?,?,?,?,?)",
(platform, kind, p, int(time.time()) + delay, 1, last_error[:300]))
def due_replays(db, limit=50):
return db.execute(
"select id, platform, kind, params, tries from replay "
"where next_try_at <= ? order by next_try_at limit ?",
(int(time.time()), limit)).fetchall()
def done_replay(db, replay_id: int):
db.execute("delete from replay where id=?", (replay_id,))
```
为什么要重放表,而不是「失败就重试三次」?因为**你要防的不是网络抖动,是你自己**。某天你改服务器配置把服务弄挂了,两天后才想起来——这两天拉的增量全丢了,重放表会把它们补回来,不需要你做任何事。团队版靠值班兜底,单人版靠表结构兜底。
另外,`updated_at` 加时间戳还有个隐藏用途:摘要邮件要报「最老的数据是多久之前的」,这个数字比「跑没跑成功」更能说明系统是不是悄悄停了。
### 3.4 主循环:每小时一轮
```python
# solo/main.py
"""cron: 0 * * * * cd /opt/solo && /usr/bin/python3 main.py >> run.log 2>&1"""
import asyncio
import json
import sys
import time
import httpx
from gate import Gate
from store import open_db, upsert_if_changed, add_replay, due_replays, done_replay
WINDOW_MIN = 10 # 每轮往前多看 10 分钟,重叠窗口防边界漏单(第 3 篇的思路)
RATE = {
"taobao": {"item_list": 5.0},
"alibaba": {"item_list": 3.0},
"jd": {"item_list": 2.0},
}
async def fetch_page(client, platform, updated_start) -> list[dict]:
"""各平台的增量列表接口。示例以淘宝为例,其余平台换参数即可"""
r = await client.get(f"/v1/{platform}/item_list", params={
"updated_start": updated_start,
"page_size": 100,
})
r.raise_for_status()
return r.json().get("items", [])
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def sync_platform(platform: str, since: int, gate: Gate, db) -> tuple[int, int]:
ok, fail = 0, 0
updated_start = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(since))
async with httpx.AsyncClient(
base_url="https://api.example.com",
headers={"X-App-Key": platform}, # 按你的网关实际方式传
timeout=10) as client:
try:
items = await fetch_page(client, platform, updated_start)
for it in items:
gate.acquire(f"{platform}:item_list", RATE[platform]["item_list"])
if upsert_if_changed(db, platform, it["item_id"], it):
ok += 1
db.commit()
except Exception as e:
fail += 1
add_replay(db, platform, "item_list",
{"updated_start": updated_start}, repr(e))
db.commit()
return ok, fail
async def replay_due(db):
"""启动先还账:上次没跑完的先补"""
for rid, platform, kind, params, tries in due_replays(db):
p = json.loads(params)
try:
async with httpx.AsyncClient(
base_url="https://api.example.com",
headers={"X-App-Key": platform}, timeout=10) as client:
items = await fetch_page(client, platform, p["updated_start"])
for it in items:
upsert_if_changed(db, platform, it["item_id"], it)
done_replay(db, rid)
db.commit()
except Exception as e:
# tries 会 +1,退避继续;退到 12h 还不行的,摘要邮件会报「卡住的重放」
add_replay(db, platform, kind, p, repr(e))
db.commit()
async def main():
db = open_db("solo.db")
gate = Gate()
await replay_due(db)
since = int(time.time()) - 3600 - WINDOW_MIN * 60
stats = {}
for platform in ("taobao", "alibaba", "jd"):
ok, fail = await sync_platform(platform, since, gate, db)
stats[platform] = (ok, fail)
# 结果写文件,digest.py 读它(比塞邮件逻辑进主循环干净)
json.dump({"ts": int(time.time()), "stats": stats},
open("last_run.json", "w"))
if __name__ == "__main__":
try:
asyncio.run(main())
except Exception:
# cron 环境里 print 没人看,异常落日志,明天摘要邮件会报「今天没跑」
import traceback
traceback.print_exc(file=open("run.log", "a"))
sys.exit(1)
```
几个刻意的简化,别当成偷懒:
- **不做并发抓取**(第 7 篇那套 20 路并发)。单平台每小时一次增量,量级撑得起串行。等你真的慢了再优化,那时你已经知道瓶颈在哪了。
- **不做配置文件**。三个平台的参数直接写在常量里。独立开发者的项目,YAML 配置层多数是自我安慰。
- **cron 而不是常驻进程**。进程挂了 cron 下个小时自己会拉起来——这是免费的兜底,比 systemd 的 `Restart=always` 心智负担更小(两个都上也行)。
### 3.5 每日摘要:单人版的「监控」
```python
# solo/digest.py
"""cron: 0 21 * * * python3 digest.py
每天一封邮件。内容只有四个数字,30 秒看完。"""
import json
import sqlite3
import smtplib
import time
from email.mime.text import MIMEText
from datetime import datetime, timedelta, timezone
CN = timezone(timedelta(hours=8))
def build(db) -> str:
run = json.load(open("last_run.json"))
age = datetime.now(CN) - datetime.fromtimestamp(run["ts"], CN)
oldest = db.execute("select min(updated_at) from items").fetchone()[0]
oldest_days = (time.time() - oldest) / 86400 if oldest else -1
stuck = db.execute(
"select count(*), max(tries) from replay where tries >= 4").fetchone()
lines = [
f"昨日变更: " + ", ".join(
f"{p} {ok}条" for p, (ok, f) in run["stats"].items()),
f"最后成功时间: {age.total_seconds()/3600:.1f} 小时前"
+ (" ← 超过 2 小时,检查服务" if age.total_seconds() > 7200 else ""),
f"最老数据: {oldest_days:.1f} 天前",
f"卡住的重放: {stuck[0]} 条(最多重试 {stuck[1] or 0} 次)",
]
return "\n".join(lines)
if __name__ == "__main__":
db = sqlite3.connect("solo.db")
body = build(db)
msg = MIMEText(body, "plain", "utf-8")
msg["Subject"] = f"[solo-sync] {datetime.now(CN):%m-%d} 摘要"
msg["From"] = "me@example.com"
msg["To"] = "me@example.com"
with smtplib.SMTP_SSL("smtp.example.com", 465) as s:
s.login("me@example.com", "your_password")
s.send_message(msg)
print(body)
```
四个数字各管一类静默故障:
- **最后成功时间 > 2 小时** → 服务挂了(cron 没跑、密钥过期、服务器欠费)
- **最老数据 > 3 天** → 某个平台的增量在悄悄失败,但整体流程还在「成功」
- **卡住的重放 > 0** → 有接口连续 4 次以上失败,需要人看一眼了
- **昨日变更数** → 和你对业务的预期对不上时(比如明明在搞活动,变更却很少),说明数据链路可能有洞
这就是单人版的监控哲学:**不做实时告警,做每天 30 秒的例行检查**。告警系统你配完头一周会认真看,第三周开始划掉通知,第三个月忘了密码。一封邮件你至少会扫一眼标题。
## 四、部署:一小时的收尾,别跳过
代码写完只算一半。独立开发者的项目死于「在我电脑上是好的」,这四步一小时搞定:
```bash
# 1. systemd 用户级单元(比系统级安全,不用 root)
mkdir -p ~/.config/systemd/user
cat > ~/.config/systemd/user/solo-digest.timer <<'EOF'
[Unit]
Description=solo digest daily
[Timer]
OnCalendar=*-*-* 21:00:00
Persistent=true
[Install]
WantedBy=timers.target
EOF
systemctl --user enable --now solo-digest.timer
# 2. 数据库备份:一天一次,拷文件就行(SQLite 的好处)
echo '0 22 * * * cp /opt/solo/solo.db /opt/solo/backup/solo-$(date +\%m\%d).db' | crontab -
# 3. 数据库目录权限:600,里面有 token
chmod 600 /opt/solo/solo.db
# 4. 密钥不进代码:环境变量或独立文件,git 里只有占位符
```
第 3 步单独说一下:`solo.db` 里有 `tokens` 表,**泄露 = 三家店铺的授权全丢**。单人项目最容易死在「数据库和代码一起扔进 git 仓库」。
## 五、放弃线:什么时候该停手
这篇如果只写「怎么接」,不说「什么时候别接」,就不诚实了。三个信号,碰到任何一个就该停:
**① 收入覆盖不了维护时间。** 用第一节的账算:开发 40 小时 + 维护 12~24 小时/年。按独立开发者的时薪折成钱,再加服务器、再加你的注意力成本。**如果你的产品月收入低于这个数的一半,直接买现成的接口服务**——不是你不行,是这笔账本来就不成立。
**② 你开始需要第二个消费方。** 数据一旦要喂两个产品(比如一个网页 + 一个小程序),归一化模型、统一接口层这些东西就从负债变资产了——那是第 2 篇和第 8 篇的领域,说明你该按团队版重做了。
**③ 权限天花板到了。** 个人主体能申请的权限包是有限的(第 17 篇讲过应用类型决定上限)。很多商品、订单类权限对企业主体都要审,个人主体经常直接不可见。**这一条不是技术问题,是资质问题,重写多少代码都解决不了**——也是独立开发者最终走向「买服务」或「注册公司」最常见的分叉点。
三平台接口的门槛里,最硬的一直不是代码。
---
二十篇到这里,正好把「一个人」这个视角补上了。前面十九篇默认读者背后有团队,这篇把约束换成「只有你」,砍法就完全不同了:**团队版优化的是吞吐和稳定,单人版优化的是维护意愿**。你接下来要维护的不是这套代码,是你自己看它的兴趣。