原创 · 跨境电商数据开发实战系列

关键词：FastAPI、SQLAlchemy、MySQL、JWT 认证、数据回滚、操作日志、跨境电商、SKU 费用核算、Ozon、Wildberries、Yandex

---

一、业务背景：为什么跨境电商需要这套系统

做俄语区跨境电商（Wildberries / Ozon / Yandex Market / FBP）的朋友应该深有体会：

1. 费用口径极其复杂。一个 SKU 在一个平台上有十几项费用：仓内操作费、头程费用、尾程出库操作费、平台佣金、站内营销、资金占用、售后资损……4 个平台加起来 47 个费用字段。
2. 费用按"周"滚动核算。平台费用每周都在变，利润核算必须精确到周度，跨周算就是错的。
3. 改一周 = 改全年。9 月第 1 周采购价上涨，那第 1 周到年末所有周都要跟着变——手工 Excel 一列列拖，改错一行就漏一列。
4. 没有审计就没人敢改数。谁改了哪个 SKU 的哪个费用、影响了哪些周、改之前是什么值——必须全部留痕，错了还能一键回滚。

这就是 SOCT（SKU Cost Track）系统的由来：一套用 FastAPI 构建的、面向跨境多平台场景的 SKU 周度费用维护与审计系统，把"费用录入 → 周度覆盖 → 批量更新 → 审计回滚"整条链路做成一个开箱即用的 Web 系统。

---

二、技术选型：为什么是 FastAPI

维度	Flask	FastAPI（最终选择）	Django
数据校验	手动	声明式 Pydantic，开箱即用	Form/DRF
API 文档	无	自动生成 Swagger UI	需额外配置
依赖注入	无	原生支持，权限控制干净	无
异步支持	弱	原生 async	弱
适合场景	小工具	API 服务 + 内部系统	大而全的 Web 应用

三个实打实的收益：

1. Pydantic 声明式校验：47 个平台费用字段 + 17 个公共字段全部声明式描述，类型和必填在接口层卡死，脏数据进不了服务层；
2. 依赖注入（DI）：Depends(get_current_user) 一个依赖统一做登录校验，业务函数完全不用关心"当前用户是谁"；
3. 自动 Swagger 文档：启动后访问 /docs 就是完整接口文档，团队零成本对接。

---

三、系统整体架构

分层架构：前端（原生 HTML/JS）→ 路由层 → 服务层（业务逻辑）→ 数据访问层 → MySQL（ODS/DIM 分层）。

flowchart TB
    subgraph 前端["前端（原生HTML+JS，FastAPI静态托管）"]
        A1[登录页]
        A2[综合查询/导出]
        A3[修改费用]
        A4[平台批量更新]
        A5[新增SKU/批量导入]
        A6[SKU全年变化]
        A7[操作日志/回滚]
    end

    subgraph 路由层["FastAPI 路由层（3个Router）"]
        B1["auth_router<br/>/api/auth 登录·用户管理"]
        B2["sku_router<br/>/api/sku 查询·修改·新增·批量"]
        B3["log_router<br/>/api/log 日志·回滚"]
    end

    subgraph 服务层["服务层 services.py"]
        C1[modify_cost 向后覆盖]
        C2[modify_by_platform 平台批量]
        C3[add_sku 全年生成]
        C4[batch_add_sku 批量导入]
        C5[rollback_operation 回滚]
    end

    subgraph 数据层["数据访问层"]
        D1[SQLAlchemy ORM + 连接池]
        D2[原生SQL批量UPDATE]
    end

    subgraph 存储["MySQL（多库）"]
        E1["e_kj_soct SKU周度费用主表<br/>复合主键(biz_week_id, SKU)"]
        E2["dim.rt_calendar_week_dim<br/>周度维度表"]
        E3["soct_user 用户表"]
        E4["soct_operation_log 操作日志表"]
    end

    A1 --> B1
    A2 --> B2
    A3 --> B2
    A4 --> B2
    A5 --> B2
    A6 --> B2
    A7 --> B3
    B1 --> C1
    B2 --> C1 & C2 & C3 & C4
    B3 --> C5
    C1 & C2 & C3 & C4 & C5 --> D1 & D2
    D1 --> E1 & E3 & E4
    D2 --> E1
    E2 -.-> C3 & C4

分层要点： 路由层只做参数接收（薄）；服务层承载全部业务规则（核心）；ORM 管单条读写、原生 SQL 管批量 UPDATE（性能）；周度维度表 dim.rt_calendar_week_dim 全局驱动"生成到年末""影响哪些周"，全年口径统一。

---

四、核心数据模型

4.1 主表：SKU × 周度的二维费用矩阵

核心设计是复合主键 (biz_week_id, SKU)：一个 SKU 一年 48~52 个周度记录，每个周度一条，字段就是该 SKU 当周在 4 个平台上的全部费用。

# models.py（核心结构）
class SoctCost(Base):
    """SKU 周度成本主表 e_kj_soct"""
    __tablename__ = "e_kj_soct"

    biz_week_id = Column(Integer, primary_key=True)   # 周序号（来自周度维度表）
    week_label  = Column(String(255))                 # 如 26-09月-W1
    SKU = Column(String(255), primary_key=True)       # 复合主键 (biz_week_id, SKU)
    品牌 = Column(String(255))
    运营人员 = Column(String(255))
    采购价CNY = Column(Double)

    # 4 个平台共 47 个费用字段，例如：
    yandex_平台佣金 = Column(Double)
    yandex_资金占用 = Column(Double)
    wb_平台佣金 = Column(Double)
    wb_平台送货费 = Column(Double)
    ozon_平台佣金 = Column(Double)
    ozon_配送费用 = Column(Double)
    fbp_平台订单处理费 = Column(Double)
    fbp_平台送货费 = Column(Double)
    # ...（其余费用字段同构，共47个）

中文列名是故意的：运营、财务直接用中文沟通字段，表结构对齐业务语言，报表导出、排查问题零翻译成本。

4.2 平台字段分组配置：加平台只改一个文件

47 个费用字段按平台分组管理，所有"按平台筛选、批量更新、默认填 0"都读这份配置：

# platform_fields.py（核心思路）
PLATFORM_FIELDS = {
    "yandex": ["yandex_香港仓仓内操作费", "yandex_平台佣金", "yandex_资金占用", "...共13项"],
    "wb":     ["wb_香港仓仓内操作费", "wb_平台佣金", "wb_平台送货费", "...共12项"],
    "ozon":   ["ozon_香港_坪山调拨费用", "ozon_平台佣金", "ozon_配送费用", "...共11项"],
    "fbp":    ["fbp_香港_坪山调拨费用", "fbp_平台订单处理费", "fbp_平台送货费", "...共11项"],
}
VARCHAR_FIELDS = {"yandex_tfs头程费用", "yandex_平台中段运输费用"}  # 字符串类型字段白名单
ALL_PLATFORMS = ["yandex", "wb", "ozon", "fbp"]

能力： 按平台筛选 → 取字段列表渲染表格；按平台批量更新 → 校验字段归属后 UPDATE；新增 SKU 未选平台 → 遍历其他平台字段自动填 0。以后加平台只改这一个文件，全系统生效。

4.3 周度维度表：全系统时间基准

def get_all_weeks(db):
    """从周度维度表获取全年周度列表（按biz_week_id排序）"""
    rows = db.execute(text(
        "SELECT DISTINCT biz_week_id, week_label "
        "FROM dim.rt_calendar_week_dim "
        "WHERE week_label IS NOT NULL AND week_label != '' "
        "ORDER BY biz_week_id"
    )).fetchall()
    return [{"biz_week_id": r[0], "week_label": str(r[1])} for r in rows if r[0] is not None]

周度由维表驱动而非代码硬编码，"第 901 周是 26-09月-W1"这种口径全系统唯一——这是数仓分层思路（DIM 维表 + ODS 数据）在业务系统里的直接落地。

---

五、工程骨架：3 个文件把服务跑起来

5.1 入口 main.py：一个服务托管前端 + API + 文档

# main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
from routers.auth_router import router as auth_router
from routers.sku_router import router as sku_router
from routers.log_router import router as log_router
from config import settings

app = FastAPI(title="SOCT - SKU周度费用维护系统", version="1.0.0")

app.add_middleware(CORSMiddleware,
    allow_origins=["*"], allow_credentials=True,
    allow_methods=["*"], allow_headers=["*"])

app.include_router(auth_router)   # /api/auth
app.include_router(sku_router)    # /api/sku
app.include_router(log_router)    # /api/log

app.mount("/static", StaticFiles(directory="static"), name="static")

@app.get("/")
def index():
    """根路径直接返回前端页面，部署=启动一个服务"""
    return FileResponse("static/index.html")

@app.get("/health")
def health():
    return {"status": "ok", "service": "SOCT"}

配置全部走 .env（数据库连接、JWT 密钥、端口），代码里零密钥；前端是原生 HTML+JS 单页（约 65KB），部署 = 启动一个服务，不需要 Nginx 配两个端口。

5.2 数据库连接：两个参数防"断连"

# database.py
engine = create_engine(
    settings.DATABASE_URL,
    pool_pre_ping=True,   # 取连接前探活，避免拿到失效连接
    pool_recycle=3600,    # 连接复用1小时即回收，防 MySQL wait_timeout 断连
)

def get_db():
    """FastAPI 依赖注入：获取数据库会话"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

---

六、JWT 认证：接口层的"门禁"

# auth.py（核心）
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi.security import OAuth2PasswordBearer

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")

def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db),
) -> SoctUser:
    """从 Token 解析当前用户（FastAPI 依赖注入）"""
    credentials_exception = HTTPException(status_code=401, detail="登录状态已失效")
    try:
        payload = jwt.decode(token, settings.JWT_SECRET_KEY,
                             algorithms=[settings.JWT_ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = db.query(SoctUser).filter(SoctUser.username == username).first()
    if user is None:
        raise credentials_exception
    return user

角色权限的两种用法：

# 普通登录即可访问：Depends 注入身份
@router.post("/modify")
def modify_cost(req: ModifyRequest, db: Session = Depends(get_db),
                current_user: SoctUser = Depends(get_current_user)):
    ...   # 操作人自动记录 current_user.username

# 仅管理员可访问：路由内二次校验
@router.post("/rollback")
def rollback(req: RollbackRequest, db: Session = Depends(get_db),
             current_user: SoctUser = Depends(get_current_user)):
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="只有管理员才能执行回滚")

---

七、核心业务实现：这 4 个逻辑是系统的灵魂

7.1 修改费用：2 条 UPDATE 实现"向后覆盖"

业务规则：改第 N 周，第 N 周及以后所有周同步更新。运营改一次采购价，全年毛利自动重算。

# services.py 核心
set_clause = ", ".join([f"`{f}` = :v{i}" for i, f in enumerate(all_updates.keys())])

# ★ 2条批量UPDATE搞定"当前周 + 后续所有周"
db.execute(text(
    f"UPDATE e_kj_soct SET {set_clause} "
    f"WHERE biz_week_id = :wid AND {where_extra}"), where_params)     # 当前周
db.execute(text(
    f"UPDATE e_kj_soct SET {set_clause} "
    f"WHERE biz_week_id > :wid AND {where_extra}"), where_params)     # 后续所有周

性能说明： 不用逐周逐 SKU 循环 UPDATE，两条 SQL 直接完成全量覆盖。实测单次批量修改覆盖 2300+ 个 SKU、几十个周度，毫秒级完成。

细节： 支持三种粒度——单个 SKU / 某品牌下所有 SKU / 某运营下所有 SKU；字符串字段（VARCHAR_FIELDS）不走 float 强转，空值落 NULL，不让脏数据进库。

7.2 平台批量更新：1 条 UPDATE 改全平台

场景：WB 平台佣金上调 0.5%，要影响所有在售 SKU 的所有后续周。

# services.py 核心
conds = ["biz_week_id >= :wid"]   # ★ 当前周及以后所有周
params = {"wid": biz_week_id, "val": value_to_set}
# 可选：品牌/运营过滤
if brand:
    conds.append("品牌 = :brand")
    params["brand"] = brand
where_clause = " AND ".join(conds)

# ★ 1条UPDATE搞定所有
db.execute(text(f"UPDATE e_kj_soct SET `{field}` = :val WHERE {where_clause}"), params)

# 统计影响范围用于日志
sku_count = db.execute(text(
    f"SELECT COUNT(DISTINCT SKU) FROM e_kj_soct WHERE {where_clause}"), params).scalar()

实测一次平台批量更新覆盖 2300+ 个 SKU、横跨 20+ 周度，秒级完成，且每一步都写日志、可回滚。

7.3 新增 SKU：从起始周自动生成到年末

新增一个 SKU 只需填基础信息和起始周，系统会：从周度维度表定位起始周 → 生成到年末，每个周度一条记录；选中平台按模板填值，未选平台费用自动填 0（防止跨平台报表出现 NULL 空洞）。

# 未选平台的所有费用字段自动填 0
for p in [p for p in ALL_PLATFORMS if p not in platforms]:
    for f in get_platform_fields(p):
        record_data[f] = 0.0 if f not in VARCHAR_FIELDS else "0"

# 从起始周生成到年末
for i in range(start_idx, len(all_weeks)):
    w = all_weeks[i]
    rec = SoctCost()
    for f in COMMON_FIELDS:
        if f in record_data and f not in ("biz_week_id", "week_label", "biz_week_yue"):
            setattr(rec, f, record_data[f])
    rec.biz_week_id = w["biz_week_id"]
    rec.week_label = w["week_label"]
    db.add(rec)

批量导入（Excel 上传 / 单行粘贴）复用同一套生成逻辑，每个 SKU 用独立 Session 处理——一个失败不影响其他，返回"成功 N 个、失败 M 个 + 失败原因清单（精确到 Excel 行号）"。

7.4 审计日志 + 数据回滚：每一步修改都有"后悔药"

日志表设计：

class SoctOperationLog(Base):
    __tablename__ = "soct_operation_log"
    id = Column(Integer, primary_key=True, autoincrement=True)
    operator = Column(String(64), nullable=False, index=True)   # 操作人
    action = Column(String(32), nullable=False)                 # modify / add / rollback
    sku = Column(String(255), nullable=False, index=True)
    biz_week_id = Column(Integer, nullable=False)
    platform = Column(String(32))                               # yandex / wb / ozon / fbp
    before_data = Column(JSON)     # ★ 修改前的数据快照（只存有变动的字段）
    after_data = Column(JSON)      # ★ 修改后的数据快照
    affected_weeks = Column(JSON)  # ★ 受影响的周度列表
    remark = Column(Text)          # 变更说明：字段 旧值→新值
    rollback_from = Column(Integer)  # 回滚来源日志ID（审计闭环）

回滚策略（两种动作两种策略）：

if log.action == "modify":
    # 策略1：恢复该SKU修改前数据，并向后覆盖所有后续周
    before_data = log.before_data or {}
    for rec in db.query(SoctCost).filter(
            and_(SoctCost.SKU == sku, SoctCost.biz_week_id >= biz_week_id)).all():
        for f, v in before_data.items():
            setattr(rec, f, v)

elif log.action == "add":
    # 策略2：删除该SKU从起始周起的全部记录
    deleted = db.query(SoctCost).filter(
        and_(SoctCost.SKU == sku, SoctCost.biz_week_id >= biz_week_id)
    ).delete(synchronize_session=False)

# ★ 回滚操作本身也记录日志（rollback_from 指向被回滚的日志），审计链路闭环

为什么回滚在跨境费用场景这么重要？ 费用数据直接决定毛利报表、广告 ROI、定价决策。一次误操作（比如批量更新选错平台）如果改完就没了，轻则报表错一周，重则定价决策全错。有了回滚，运营敢放手操作，权限也敢下放。

---

八、查询、导出与前端

• 综合查询：周度 + 品牌 + 运营 + 平台 + SKU 组合查询，分页展示；指定平台时只返回公共字段 + 该平台费用字段，前端不冗余；
• SKU 全年变化：逐周对比自动标出"变化周 + 变化字段"，运营一眼看到"这个 SKU 从第 901 周起佣金变了"；
• CSV 导出：charset=utf-8-sig（带 BOM），Excel 打开中文不乱码——这个坑务必注意；
• 前端：原生 HTML+JS 单页（约 65KB），六个 Tab：综合查询 / 修改费用 / 平台批量更新 / 新增SKU（模板下载、Excel 粘贴解析、批量上传）/ SKU全年变化 / 操作日志（仅管理员，含详情与回滚按钮）。

---

九、项目结构一览

kj_cost/                          # SOCT 系统根目录
├── main.py                       # FastAPI 入口：CORS、路由、静态托管
├── config.py                     # 配置（.env 读取，代码零密钥）
├── database.py                   # SQLAlchemy 引擎 + Session + 依赖注入
├── models.py                     # 数据模型：SoctCost / SoctUser / SoctOperationLog
├── schemas.py                    # Pydantic 请求/响应模型
├── platform_fields.py            # ★ 平台字段分组配置（核心可扩展点）
├── services.py                   # ★ 核心业务：修改/批量/新增/日志/回滚
├── auth.py                       # JWT 认证 + 密码哈希 + 当前用户依赖
├── init_db.py                    # 数据库初始化（建表 + 默认管理员）
├── routers/
│   ├── auth_router.py            # /api/auth 登录、用户管理
│   ├── sku_router.py             # /api/sku 查询、修改、新增、批量、导出
│   └── log_router.py             # /api/log 日志查询、详情、回滚（仅admin）
└── static/
    └── index.html                # 极简 HTML 前端（约65KB，无框架）

---

十、部署与启动

pip install -r requirements.txt   # fastapi / uvicorn / sqlalchemy / pymysql / python-jose / passlib[bcrypt]
python init_db.py                 # 建表 + 默认管理员
python main.py                    # 开发模式（reload=True）
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4   # 生产模式
# 前端：http://localhost:8000/    Swagger：http://localhost:8000/docs

---

十一、写在最后：这套系统的分量在哪

对跨境电商团队来说，费用核算不是"记账"，而是定价、备货、广告投放的决策底座。SOCT 的价值四点：

1. 口径统一：周度维度表 + 47 个平台字段集中管理，全公司一套费用口径，告别"各运营各一张 Excel，对不上数"；
2. 效率跃迁：改一次采购价自动向后覆盖全年；平台批量更新一次覆盖 2300+ 个 SKU，秒级完成；
3. 安全可控：JWT 登录 + 角色权限 + 全量操作日志 + 数据回滚，敢改数、改错了能还原；
4. 开箱即用：FastAPI 一个服务托管前端 + API + 自动文档，部署即用，对接 FineBI / PowerBI 直接走 API 或读库都行。

技术高光点（面试和文章最有价值的部分）：

• 复合主键 (biz_week_id, SKU) 二维费用模型，一个表装下全年 × 全 SKU × 全平台的费用矩阵；
• "修改向后覆盖"批量 UPDATE 策略：2 条 SQL 搞定当前周 + 后续所有周；
• 配置驱动的平台字段管理：加平台只改一个文件；
• JSON 快照式审计日志 + 双策略回滚：modify 恢复并覆盖、add 整段删除、回滚自身也留痕；
• 独立 Session 批量导入：部分成功部分失败，失败明细精确到 Excel 行号。

如果你也在做跨境电商的数据系统，或者想了解 FastAPI 在真实业务中的落地姿势，欢迎评论区交流。

---

本文代码为脱敏后的核心逻辑示意，数据库连接、JWT 密钥、真实业务数据均以占位符替代。