# 代运营公司怎么管 20 个店铺的数据？我们是这样做的（附 Python 源码）
> 代运营的数据问题和卖家不一样。
>
> 卖家的问题只有一个店：取到数据、别出错、别断。代运营的问题是一堆店：**授权会过期、数据不能串、客户要各自的报表、一个店出问题不能影响其他十九个**。
>
> 我们踩过的坑基本都在这四件事上，跟接口本身关系不大。20 个店铺、47 个有效授权（有的店只在两个平台开店）、三个平台三套授权体系——这套东西现在是五个文件的代码在跑，人工每天 10 分钟看一张汇总表。
>
> 这篇把授权、隔离、汇报三块拆开讲，代码可以直接拿去改。
## 一、先把「店铺」当成一级实体
大部分多店铺系统出问题，根子在建模：**把店铺当成了配置，而不是资源**。
配置是没有生命周期的——写进去就不管了。但一个店铺授权是有生命周期的：客户签进来 → 授权 → 用着 → 快到期 → 客户换人/改密码 → 授权失效 → 客户走了 → 撤权。**每一步都有状态，每一步都需要代码处理**。
正确的建模是两层：
```
店铺（shop）
└── 授权（shop_id + platform） ← 唯一的凭证单元
```
20 家店 × 3 平台，不是 20 条记录，是 47 个授权（有 7 家店只在两个平台开店）。凭证表的主键必须是 `(shop_id, platform)`，不能是 `shop_id`——**否则一家店在三个平台的 token 只能存一个，后写的覆盖先写的，你会以为授权好了，其实只好了最后一个平台**。这个 bug 我们见过不止一次。
授权有五个状态，别用布尔值糊过去：
| 状态 | 含义 | 系统该做什么 |
| ---------- | ------------- | ------------ |
| `pending` | 已建店，授权还没完成 | 催授权，不进调度队列 |
| `active` | 正常 | 正常调度 |
| `expiring` | 剩余 < 48 小时 | 提前预警，任务照跑 |
| `expired` | 已过期 | 停止调度，标记客户待处理 |
| `revoked` | 客户主动解绑 / 平台收回 | 停止调度，通知商务 |
`expiring` 这个状态是最值钱的一个。**代运营每月有 3~5 个授权到期**，等过期了再处理，意味着那家店的日报当天就断了，客户第二天早上会来问。提前 48 小时预警，实际处理时间是两天，从容得多。
## 二、token 存储：加密与否，是责任问题
47 个店铺的 refresh token 躺在数据库里，这就是 47 张店铺的钥匙。数据库被拖、备份文件进了 git、开发同学的笔记本被偷——任何一个环节泄露，客户的全店数据就没了。
**这不是「应该加密」，是「必须加密」**，而且要在写代码的第一天做，后面补的成本高得多（要迁移存量、要处理轮换）。
```python
# ops/vault.py
import base64
import os
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
class Vault:
"""token 加解密：AES-GCM，密文入库，密钥永远不进数据库。
keys 是 {key_id: 32字节密钥}，current 是当前用来加密的 key_id。
存 key_id 的目的只有一个：轮换。换密钥时老密文还能用老 key 解开，
解密成功的记录顺手用新 key 重新加密，慢慢滚过去。
"""
def __init__(self, keys: dict[str, bytes], current: str):
self.keys = keys
self.current = current
for k in keys.values():
if len(k) != 32:
raise ValueError("AES-256 需要 32 字节密钥")
def seal(self, plaintext: str) -> tuple[str, str, str]:
"""返回 (密文, nonce, key_id)，三列一起入库"""
nonce = os.urandom(12)
ct = AESGCM(self.keys[self.current]).encrypt(
nonce, plaintext.encode("utf-8"), None)
return (base64.b64encode(ct).decode(),
base64.b64encode(nonce).decode(),
self.current)
def open(self, ct_b64: str, nonce_b64: str, key_id: str) -> str:
return AESGCM(self.keys[key_id]).decrypt(
base64.b64decode(nonce_b64),
base64.b64decode(ct_b64),
None).decode("utf-8")
```
密钥从哪来？三个档，按你的规模选：
| 方式 | 适合 | 注意 |
| -------------- | ---------- | --------------------------- |
| 环境变量 | 小团队、单机 | 进程能读到，宿主机被拿就没了；至少别写进代码和 git |
| 云 KMS / 密钥管理服务 | 十家店以上、多机部署 | 每次解密一次网络调用，要配本地短缓存（5 分钟） |
| 硬件加密机 | 有等保要求 | 代运营基本用不上，成本不划算 |
我们用的是云 KMS + 进程内 5 分钟缓存。缓存的是**解密后的 token**，不是密钥——这个区别很重要，密钥永不落地。
**一个容易漏的地方**：`refresh_token` 也要加密，而且它比 `access_token` 更重要。`access_token` 两小时就过期了，`refresh_token` 能换出无数个新的。只加密前者不加密后者，等于没加密。
## 三、并发刷新撞车：多店铺场景的第一次翻车
单个店铺的时候你不会遇到这个问题。20 个店铺、多个任务并发跑的时候，它一定会来。
场景长这样：某个店的 token 在 10:00:00 过期。10:00:01，订单同步任务和商品同步任务同时发现过期，同时发起刷新。平台的处理是**后刷作废先刷**（京东和 1688 都有这个行为），于是：
- 任务 A 拿到的 token 已经作废了 → 401
- 任务 B 拿到的 token 有效 → 继续跑
- 更糟的情况：两个都拿到「刷新成功」，但只有一个真正有效，另一个在下一次调用时才暴露
结果就是「同一个店，一半任务成功一半失败，重试之后好了」——这种故障最难查，因为它不稳定复现。
解法是**单飞（single-flight）**：同一个 `(shop_id, platform)` 在同一时刻只能有一个刷新在跑，其他协程等它的结果。
```python
# ops/refresher.py
import time
# 抢锁靠这一行 update：只有拿到了刷新权的那个协程，rowcount 才是 1
CLAIM = """
update grants
set locked_until = :until
where shop_id = :shop_id and platform = :platform
and (locked_until is null or locked_until < :now)
"""
def ensure_token(db, vault: Vault, shop_id: str, platform: str,
refresh_fn, ahead: int = 300) -> str:
"""返回可用的 access_token。
ahead：提前多少秒视为「需要刷新」。别卡着过期点刷，
网络一来一回加上平台处理，卡点必翻车。
"""
for _ in range(3): # 最多等三轮，够覆盖一次正常刷新
row = db.execute(
"select access_token, nonce, key_id, refresh_token, "
"refresh_nonce, refresh_key_id, expire_at, state "
"from grants where shop_id=? and platform=?",
(shop_id, platform)).fetchone()
if row is None:
raise LookupError(f"未找到授权 {shop_id}/{platform}")
(ct, nonce, key_id, rct, rnonce, rkey_id, expire_at, state) = row
if state in ("revoked", "expired"):
raise PermissionError(f"授权不可用 {shop_id}/{platform} state={state}")
if expire_at > time.time() + ahead:
return vault.open(ct, nonce, key_id)
now = int(time.time())
cur = db.execute(CLAIM, {"until": now + 30, "shop_id": shop_id,
"platform": platform, "now": now})
db.commit()
if cur.rowcount == 0:
# 别人正在刷，等一下再读它刷好的结果
time.sleep(0.3)
continue
try:
plain_rt = vault.open(rct, rnonce, rkey_id)
new_token, expires_in, new_rt = refresh_fn(plain_rt)
n_ct, n_nonce, n_kid = vault.seal(new_token)
r_ct, r_nonce, r_kid = vault.seal(new_rt or plain_rt)
db.execute(
"update grants set access_token=?, nonce=?, key_id=?, "
"refresh_token=?, refresh_nonce=?, refresh_key_id=?, "
"expire_at=?, state='active', locked_until=null, "
"updated_at=? where shop_id=? and platform=?",
(n_ct, n_nonce, n_kid, r_ct, r_nonce, r_kid,
int(time.time()) + expires_in, now, shop_id, platform))
db.commit()
return new_token
except Exception as e:
# 刷新失败：不清旧 token，只标记状态。旧的可能还能用一会儿，
# 清掉就是立刻断服——代运营断一家店，客户当天就知道。
db.execute(
"update grants set locked_until=null, last_error=?, "
"state=case when expire_at < ? then 'expired' else state end "
"where shop_id=? and platform=?",
(repr(e)[:300], int(time.time()), shop_id, platform))
db.commit()
raise
finally:
# 释放锁。用条件更新而不是直接置 null：
# 万一本进程卡了很久、锁已被别人重新抢到，别把别人的锁解掉
db.execute(
"update grants set locked_until=null "
"where shop_id=? and platform=? and locked_until <= ?",
(shop_id, platform, now + 30))
db.commit()
raise TimeoutError(f"等待刷新超时 {shop_id}/{platform}")
```
三个细节值得说：
1. **抢锁用的是 `update ... where locked_until is null or < now`**，没有引入 Redis。行级条件更新本身就是原子的，`rowcount == 1` 就是拿到锁。多店铺场景下这比引入一套分布式锁简单，也少一个故障点——除非你多机部署，那就换 Redis 或 PG 的 `select ... for update`。
2. **`locked_until` 是超时时间，不是布尔锁**。进程被 kill、机器重启，锁不会永久卡住——30 秒后自动可抢。用布尔锁的人一定会遇到「一次 OOM 之后这个店再也刷不了 token」。
3. **刷新失败不清旧 token**。这条和第 20 篇单人版一致，但理由不同：单人版是「没空管」，代运营是「断一家店客户当天就知道」。让旧 token 多活一会儿，比立刻 401 好。
## 四、隔离：一个店挂掉，不能带走其他十九个
这是多店铺和单店铺最本质的区别。20 家店的调度如果写成这样：
```python
for shop in shops:
await sync(shop) # 一家超时 60 秒，20 家就是 20 分钟
```
那么任何一家店出问题——授权过期、类目权限被收回、平台侧限流——整轮任务就停在那里。更隐蔽的是额度：**20 家店共用一个令牌桶，一家店做大促把额度烧光，其他 19 家的日报全部延迟**。
三件事必须按店铺拆开：
```python
# ops/fanout.py
import asyncio
class ShopWorker:
"""每个店铺一个独立上下文：独立限流桶、独立退避、独立错误计数。
这就是第 9 篇「舱壁隔离」在代运营场景的具体形态。"""
def __init__(self, shop_id: str, gate, backoff_s: float = 0):
self.shop_id = shop_id
self.gate = gate # 按 (shop, platform, api) 分桶
self.backoff_s = backoff_s
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def run_round(workers: list[ShopWorker], task_fn, timeout: float = 300):
"""一轮调度：所有店铺并发，互不阻塞，单店异常只影响自己"""
async def one(w: ShopWorker):
try:
# 单店整体超时：防止一家店拖住整轮
return w.shop_id, await asyncio.wait_for(
task_fn(w), timeout=timeout)
except asyncio.TimeoutError:
return w.shop_id, {"ok": False, "reason": "shop_timeout"}
except PermissionError as e:
return w.shop_id, {"ok": False, "reason": f"auth:{e}"}
except Exception as e:
return w.shop_id, {"ok": False, "reason": repr(e)[:160]}
results = await asyncio.gather(*(one(w) for w in workers))
ok = sum(1 for _, r in results if r.get("ok"))
return {"ok": ok, "fail": len(results) - ok, "detail": dict(results)}
```
关键就一句：`gather(..., return_exceptions=True)` 的效果——**用 `wait_for` 包住单店 + 逐个捕获**，这样一家店超时，其他十九家的结果照常返回。别用 `gather` 的默认行为，它会让第一个异常直接冒出，剩下的协程被取消，等于全挂。
额度也要按店分：
```python
# 每个店铺每天一个额度配额，用完就停，不占别人的
gate = Gate(per_key_quota={
"shop:taobao:item_list": 800,
"shop:jd:item_list": 500,
})
```
这样一家店做活动把额度打满，只是它自己停，其他店照跑。**这个设计在 618、双 11 那种日子价值极高**——那一天你一定会遇到某家店的数据量翻十倍。
## 五、数据隔离：靠 review 记住加 `where` 一定会漏
技术隔离是内部问题，数据隔离是**对客户的责任问题**。你的系统里同时有 20 家店的数据，客户 A 登录进来，绝不能看到客户 B 的任何一行。
两种做法，差别很大：
**做法一（错）**：业务代码里手写 `where shop_id = ?`。写的时候都记得，三个月后新加一个查询，忘了。这类事故在代运营行业不是罕见事件。
**做法二（对）**：在数据访问层强制注入，业务层拿不到「不带 shop 的入口」。
```python
# ops/tenant.py
class TenantRepo:
"""所有查询强制带 shop_id。业务代码只有这一个入口，
想查全库得绕过这个类——那就要 review 了，能拦住。"""
def __init__(self, db, shop_id: str):
if not shop_id:
raise ValueError("shop_id 不能为空") # 空值 = 全库查询，直接炸
self.db, self.shop_id = db, shop_id
def items(self, platform: str, limit: int = 500):
return self.db.execute(
"select item_id, payload, updated_at from items "
"where shop_id=? and platform=? order by updated_at desc limit ?",
(self.shop_id, platform, limit)).fetchall()
def orders(self, platform: str, start: int, end: int):
return self.db.execute(
"select * from orders where shop_id=? and platform=? "
"and created_at between ? and ?",
(self.shop_id, platform, start, end)).fetchall()
# 需要跨店聚合的（内部报表），单独一个类，命名上区分开
```
`shop_id` 为空就抛异常那两行是关键：**空值等于「查全库」**，历史上无数越权事故就是这么来的（参数没传、前端选错、定时任务忘了带）。让它炸，比让它查全库安全。
再配一条测试，比任何 review 都可靠：
```python
def test_no_cross_shop_leak(db):
a = TenantRepo(db, "shop_a").items("taobao")
b_ids = {r[0] for r in TenantRepo(db, "shop_b").items("taobao")}
assert not ({r[0] for r in a} & b_ids), "A 的查询结果里出现了 B 的商品"
```
这条测试跑在 CI 里，任何人写出串店查询都会被拦下。
## 六、店铺生命周期：进来和走的时候最容易乱
20 家店的代运营，每个月都有新客户进来、老客户离开。这两件事没有做成流程，就会留下垃圾和风险。
**进来（onboarding）**，25 分钟能搞定，但要有清单：
| 步骤 | 容易漏的 |
| --------- | ---------------------------------------- |
| 建店记录、绑定平台 | 一个客户在三个平台开店，要建 3 个授权，别只建一个 |
| 商家授权 | **必须商家本人扫码**，不能用员工账号代授——员工离职那天，20 家店一起失效 |
| 核对权限包 | 商品、订单、物流、售后四个包分别确认，缺哪个当场说 |
| 拉一次样本 | 抽 5 个商品 + 3 个订单，跟商家后台对一遍，确认字段和口径 |
| 配日报 | 收件人、时间、关注字段；先发一次给客户确认格式 |
| 记入到期日历 | 授权到期日 + 48 小时预警，写进系统而不是某人的日历 |
那次「员工用老板账号授权」值得单独说。代运营公司里常见做法是运营用客户的主账号扫码授权，图省事。后果是：客户改密码、老板换手机、员工离职交接不清，任何一种都会让授权失效，而且**失效的时候你不会收到通知**，只会发现数据不更新了。正确做法是让商家用自己的账号授权给你的服务商应用，然后给你开子账号。
**离开（offboarding）**，比进来更重要，因为涉及数据责任：
1. **立即 revoke**：调平台的解绑接口，别只是把状态改成 `revoked`——平台侧的授权还在，只是你不用，等于钥匙还在门上。
2. **数据保留期写进合同**：常见是 30 天，到期物理删除。别口头说「我们会删」。
3. **密钥销毁**：这家店的 token 密文连同 `key_id` 一起删。留着解不开也没意义，但审计时会问。
4. **归档报表**：客户可能要历史数据，导出成文件交给他，然后删库。
第 1 条我们踩过：客户走了三个月，我们只是标了 `revoked`，平台侧授权还在。客户的 IT 后来做安全审计，发现我们的应用还在他的授权列表里，直接投诉到了平台。**这件事的后果不是技术问题，是信任问题。**
## 七、20 家店的日报：一张表看完，异常行标记
最后是每天真正要看的东西。47 个授权，不可能一个个点进去看，所以汇总成一张表，一屏装下：
```python
# ops/dashboard.py
import time
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def daily_board(db) -> list[dict]:
"""一行 = 一个授权。异常行带 reason，前端标红。"""
rows = db.execute("""
select g.shop_id, g.platform, g.state,
(g.expire_at - strftime('%s','now')) / 86400.0 as days_left,
(select count(*) from items i
where i.shop_id=g.shop_id and i.platform=g.platform
and i.updated_at > strftime('%s','now') - 86400) as changed,
(select max(updated_at) from items i
where i.shop_id=g.shop_id and i.platform=g.platform) as last_ok
from grants g
order by g.state desc, days_left asc
""").fetchall()
out = []
for shop_id, platform, state, days_left, changed, last_ok in rows:
reason = ""
if state in ("expired", "revoked"):
reason = "授权不可用"
elif days_left is not None and days_left < 2:
reason = f"授权 {days_left:.1f} 天后到期"
elif last_ok is None or (time.time() - last_ok) > 86400 * 2:
reason = "超过 2 天没有新数据"
elif changed == 0:
reason = "昨日无变更" # 不一定是错，但值得看一眼
out.append({"shop": shop_id, "platform": platform, "state": state,
"days_left": round(days_left, 1) if days_left else None,
"changed": changed, "reason": reason})
return out
```
这张表每天 9:00 推到运营群，20 行，异常行带原因。运营的实际工作量从「逐店巡检 2 小时」变成「看 10 分钟，处理 1~3 个异常」。
要注意 `changed == 0` 这一条：**它不是错误，但它是信号**。一家正常营业的店，一天商品和订单一定有变化。连续两天为 0，八成是拉取失败但没报错（第 14 篇讲过的空值问题）。这条规则帮我们抓到过两次静默故障。
## 八、六个坑，按踩到的频率排
| 坑 | 症状 | 解法 |
| ------------------ | ------------------------- | --------------------------- |
| 员工用客户主账号授权 | 员工离职 / 客户改密码，一批店同时失效，且无通知 | 商家本人授权 + 子账号操作 |
| 凭证表主键用 `shop_id` | 一家店三个平台只有一个 token 生效 | 主键必须是 `(shop_id, platform)` |
| 只加密 access_token | 数据库泄露 = 永久授权泄露 | refresh_token 一起加密 |
| 全局令牌桶 | 一家店大促把额度吃光，其他店全延迟 | 按店铺分桶 |
| 手写 `where shop_id` | 某次查询漏了，串店 | 数据层强制注入 + CI 测试 |
| 离开时只标状态不解绑 | 客户审计发现授权还在 | 调平台解绑接口 |
## 九、这套东西的规模效应
47 个授权，每天跑一轮增量：
- 全量拉取：约 18,800 次/天（47 × 平均 400 商品）
- 改增量后：约 2,300 次/天（只有变动的商品和订单）
人工部分：巡检从 2 小时/天降到 10 分钟/天，月底对账从半天降到 1 小时。真正省下的不是时间，是**「哪家店可能出问题」这件事不再依赖某个人记得**。
---
多店铺管理这件事，技术上没有什么新鲜东西——加密、单飞、隔离、强制注入，都是老办法。它难的地方在于：**每一个细节都对应一个真实客户的损失**。串店是信任问题，断服是续费问题，授权过期是关系问题。
所以这套代码里我写得最仔细的，不是并发和性能，是那几行「失败时不清旧 token」「shop_id 为空就抛异常」「离开时必须真的解绑」。
