云南某科技公司本地商贸订单与经营数据平台

项目背景

这个项目是我在云南昆明、文山一带某科技公司工作时参与整理和维护的一套本地商贸订单与经营数据平台。公司服务的客户以本地门店、农产品销售、校园周边商家和小型批发零售业务为主，项目目标不是做一个特别复杂的大平台，而是把“商品展示、用户下单、商家处理、经营统计、后台维护、服务器部署”这些事情稳定地跑起来。

当时业务现场比较真实：客户经常通过微信、电话、群消息接单，商品库存和价格更新不及时，订单状态靠人工记，月底做统计又要翻聊天记录和 Excel。我们做系统时，重点考虑的是让商家少录几次、少算几次、少漏几单。页面不追求花哨，而是要让老板、店员、配送人员都能看懂。

我在项目里主要做后端接口、后台页面、部分小程序联调、数据库表设计、数据统计、部署脚本和问题排查。项目上线后也根据客户反馈改过多轮，比如订单备注、商品排序、营业时间、库存提示、报表导出、图片压缩、后台筛选条件等。

我的职责

• 根据客户的实际业务流程整理需求，把“商品、订单、客户、门店、配送、统计”拆成模块。
• 参与数据库表设计，维护商品表、订单表、订单明细表、用户表、门店配置表和操作日志表。
• 编写后端接口，包括商品列表、订单提交、订单状态流转、后台查询、报表统计和文件上传。
• 对接小程序端页面，处理参数校验、分页、图片地址、状态文案和接口异常提示。
• 做后台管理页面，包括商品上下架、分类维护、订单筛选、备注修改和 Excel 导出。
• 参与服务器部署，配置 Nginx 反向代理、日志目录、数据库备份和基本健康检查。
• 后期补充测试脚本和接口用例，避免改订单流程时影响已上线功能。

业务流程

用户在小程序里浏览商品，选择规格和数量后加入购物车，确认地址或自提信息后提交订单。后端生成订单号、保存订单主表和订单明细，并把订单状态初始化为“待处理”。商家后台能看到新订单，根据库存和营业情况选择接单、取消或备注。订单完成后，统计任务会把当天订单金额、商品销量和客户数量汇总到报表页面。

这个流程听起来简单，但真实业务里会有很多细节：商品临时下架怎么办，用户重复提交怎么办，订单改备注是否要记录操作人，后台分页查询慢怎么办，图片上传太大怎么办，客户要求按商品分类统计怎么办。这些问题基本都是上线后慢慢遇到，再逐步补上。

模块拆分

项目按模块拆分后比较清楚：小程序端负责展示和下单，后台负责管理和处理，API 服务负责业务规则，数据库负责保存核心数据，定时任务负责统计和备份。后期如果要迁移，也能把前端、小程序和服务端分开调整。

核心模块包括：

• 商品模块：分类、列表、详情、规格、价格、库存、上下架、排序。
• 订单模块：创建订单、订单明细、状态流转、备注、取消、完成。
• 用户模块：微信用户标识、手机号、地址、自提信息、历史订单。
• 商家后台：登录、订单筛选、商品维护、基础统计、导出报表。
• 数据统计：按天、按商品、按分类、按门店统计销售情况。
• 运维模块：日志、备份、上传文件清理、简单健康检查。

数据库设计片段

订单表设计时，最重要的是不要把所有信息都塞进一个字段里。订单主表保存用户、金额、状态、地址、备注、时间；订单明细保存商品、数量、单价和规格。这样后台统计商品销量时不用解析复杂 JSON。

CREATE TABLE shop_order (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    order_no VARCHAR(64) NOT NULL UNIQUE,
    user_id BIGINT NOT NULL,
    store_id BIGINT NOT NULL,
    total_amount DECIMAL(10, 2) NOT NULL DEFAULT 0,
    status VARCHAR(32) NOT NULL DEFAULT 'pending',
    receiver_name VARCHAR(64),
    receiver_phone VARCHAR(32),
    address VARCHAR(255),
    remark VARCHAR(255),
    created_at DATETIME NOT NULL,
    updated_at DATETIME NOT NULL,
    INDEX idx_store_status_time (store_id, status, created_at),
    INDEX idx_user_time (user_id, created_at)
);

CREATE TABLE shop_order_item (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    order_id BIGINT NOT NULL,
    product_id BIGINT NOT NULL,
    product_name VARCHAR(128) NOT NULL,
    sku_name VARCHAR(128),
    price DECIMAL(10, 2) NOT NULL,
    quantity INT NOT NULL,
    subtotal DECIMAL(10, 2) NOT NULL,
    INDEX idx_order_id (order_id),
    INDEX idx_product_id (product_id)
);

后端接口代码片段

订单创建接口最开始写得比较直接，后面逐步补了参数校验、库存判断、金额重新计算和重复提交处理。金额不能完全相信前端传过来的值，后端必须根据商品表重新计算。

from decimal import Decimal
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel

router = APIRouter(prefix="/api/orders", tags=["orders"])

class OrderItemIn(BaseModel):
    product_id: int
    sku_id: int | None = None
    quantity: int

class CreateOrderIn(BaseModel):
    user_id: int
    store_id: int
    items: list[OrderItemIn]
    address: str | None = None
    remark: str | None = None

@router.post("")
def create_order(payload: CreateOrderIn):
    if not payload.items:
        raise HTTPException(status_code=400, detail="订单不能为空")

    total = Decimal("0.00")
    rows = []
    for item in payload.items:
        product = product_repo.get_available_product(item.product_id)
        if not product:
            raise HTTPException(status_code=404, detail="商品不存在或已下架")
        if item.quantity <= 0:
            raise HTTPException(status_code=400, detail="商品数量不正确")
        if product.stock is not None and product.stock < item.quantity:
            raise HTTPException(status_code=400, detail=f"{product.name} 库存不足")
        subtotal = product.price * item.quantity
        total += subtotal
        rows.append({
            "product_id": product.id,
            "product_name": product.name,
            "price": product.price,
            "quantity": item.quantity,
            "subtotal": subtotal,
        })

    order = order_service.create_order(
        user_id=payload.user_id,
        store_id=payload.store_id,
        total_amount=total,
        address=payload.address,
        remark=payload.remark,
        items=rows,
    )
    return {"order_no": order.order_no, "total_amount": str(order.total_amount)}

后台筛选和统计

后台列表最常用的功能是订单筛选。客户会按日期、状态、手机号、商品名称查订单，所以接口里做了分页和组合条件。后期发现订单多了以后，单纯查主表不够，还要给常用条件加索引，并把商品统计独立成报表查询。

def query_orders(store_id: int, status: str | None, keyword: str | None, start, end, page: int):
    sql = ["SELECT * FROM shop_order WHERE store_id = :store_id"]
    params = {"store_id": store_id, "limit": 20, "offset": (page - 1) * 20}

    if status:
        sql.append("AND status = :status")
        params["status"] = status
    if start and end:
        sql.append("AND created_at BETWEEN :start AND :end")
        params["start"] = start
        params["end"] = end
    if keyword:
        sql.append("AND (order_no LIKE :kw OR receiver_phone LIKE :kw)")
        params["kw"] = f"%{keyword}%"

    sql.append("ORDER BY created_at DESC LIMIT :limit OFFSET :offset")
    return db.fetch_all(" ".join(sql), params)

数据看板

经营看板主要展示当天订单数、销售额、客单价、热销商品、近 7 天趋势。这个功能客户很喜欢，因为它不需要打开数据库，也不用手动算 Excel。数据量不大时可以实时查，后期为了减少后台打开速度，做了每日汇总表。

import pandas as pd

def build_daily_report(order_rows):
    df = pd.DataFrame(order_rows)
    if df.empty:
        return {"amount": 0, "orders": 0, "avg": 0}
    paid = df[df["status"].isin(["accepted", "finished"])]
    amount = paid["total_amount"].sum()
    count = paid["order_no"].nunique()
    return {
        "amount": float(amount),
        "orders": int(count),
        "avg": float(amount / count) if count else 0,
    }

部署和运维

项目部署在 Linux 服务器上，前面用 Nginx 做 HTTPS 和反向代理，后端服务通过 systemd 或 Docker 方式运行。数据库每天定时备份一次，上传图片按日期分目录保存。上线后我比较常做的事情是看接口日志、查慢请求、修正客户录错的数据、调整 Nginx 上传大小限制。

后面整理项目时，我也补过 Dockerfile 和 docker-compose，方便在测试服务器快速跑起来。更完整的版本可以迁移到 K8s，用 Deployment 管应用，用 ConfigMap 管配置，用 Secret 放数据库密码。不过对这种本地项目来说，Docker Compose 已经能解决大部分部署重复劳动。

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
ENV TZ=Asia/Shanghai
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

services:
  api:
    build: .
    restart: always
    environment:
      DATABASE_URL: mysql+pymysql://shop:password@mysql:3306/shop
      REDIS_URL: redis://redis:6379/0
    ports:
      - "8000:8000"
    depends_on:
      - mysql
      - redis
  redis:
    image: redis:7-alpine
  mysql:
    image: mysql:8
    environment:
      MYSQL_DATABASE: shop
      MYSQL_USER: shop
      MYSQL_PASSWORD: password
      MYSQL_ROOT_PASSWORD: root_password

测试习惯

这个项目让我意识到订单流程一定要有测试。因为订单状态一改，可能影响小程序展示、商家后台按钮、统计报表和导出结果。后来我会把核心函数拆出来写 pytest，例如金额计算、库存判断、状态流转、日期筛选。前端如果用 Vue，也会给金额格式化、状态文案、表单校验这类工具函数写 Vitest。

def test_calc_order_total():
    items = [
        {"price": 12.5, "quantity": 2},
        {"price": 8, "quantity": 3},
    ]
    assert calc_order_total(items) == 49.0

def test_status_flow_cancel_after_finished_failed():
    order = Order(status="finished")
    with pytest.raises(ValueError):
        order.cancel()

import { describe, expect, it } from "vitest";
import { statusText } from "./order";

describe("order status text", () => {
  it("renders finished status", () => {
    expect(statusText("finished")).toBe("已完成");
  });
});

和行业项目的关系

这个项目本身是商贸订单系统，但里面很多经验可以迁移到其他行业：商品可以换成工程物资，订单可以换成巡检任务，经营报表可以换成水文气象统计，门店可以换成站点或项目部。后来我整理 GIS、时序数据、AI 知识库这些项目时，也沿用了类似的思路：先把业务对象拆清楚，再设计表，再写接口，最后做页面和部署。

如果放到能源、水利或气象方向，类似架构可以扩展成：站点数据采集、TimescaleDB 或 InfluxDB 存储时序数据、ECharts 展示趋势、GIS 地图展示点位、后台维护站点信息、Docker 部署数据服务。也就是说，这个项目不是直接做水利电力，但它锻炼了业务系统落地所需要的通用能力。

项目复盘

我觉得这个项目最大的价值是“真实”。真实客户不会只问用了什么技术，而是会问为什么订单查不到、为什么图片传不上去、为什么今天统计不对、为什么手机上按钮点不到。做这种项目能让人更快理解工程开发不是只写代码，还包括沟通、排查、取舍和持续维护。

如果现在重新做，我会从一开始就把接口文档、测试用例、日志字段和部署脚本补齐；后台前端会用 Vue 3 + TypeScript 写得更规范；数据统计会拆成独立任务；如果订单数据增长较快，会考虑 Redis 缓存热点配置，把报表汇总表做得更清晰。

这篇记录保留了一些代码片段和架构图，但隐藏了真实公司名称、客户名称、服务器配置、支付参数和数据库密码。它更适合作为我在职期间项目经历的整理：业务功能边界清晰，覆盖了从需求整理、接口开发、后台管理到上线维护的完整过程。