×

写给独立开发者:一个人怎么把三平台数据接口接完(附 Python 源码)

万邦科技Lex 万邦科技Lex 发表于2026-11-03 11:05:47 浏览22 评论0

抢沙发发表评论

# 写给独立开发者:一个人怎么把三平台数据接口接完(附 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 篇讲过应用类型决定上限)。很多商品、订单类权限对企业主体都要审,个人主体经常直接不可见。**这一条不是技术问题,是资质问题,重写多少代码都解决不了**——也是独立开发者最终走向「买服务」或「注册公司」最常见的分叉点。

三平台接口的门槛里,最硬的一直不是代码。

---

二十篇到这里,正好把「一个人」这个视角补上了。前面十九篇默认读者背后有团队,这篇把约束换成「只有你」,砍法就完全不同了:**团队版优化的是吞吐和稳定,单人版优化的是维护意愿**。你接下来要维护的不是这套代码,是你自己看它的兴趣。


群贤毕至

访客