简介:面向具备Python基础、数据库与Web开发经验的研发人员及信息系统专业学生,这份docx文档完整呈现了古城文创商品订单与库存管理系统的项目实例。内容围绕商品资料统一管理、订单与库存联动、库存预警与经营分析、文化内容融合四条主线展开,采用Flask/FastAPI作后端、SQLite/MySQL作数据库、Tkinter实现桌面端GUI,并借助事务机制保障订单扣减与库存恢复的一致性,配合安全库存与移动平均算法给出补货建议。资源包仅1个docx文件,约119KB,集中承载系统设计说明、数据库脚本、API接口规范与关键代码示例,结构紧凑便于通读。目前已有226人学习,适合作为Python全栈教学案例,也可为古城景区、博物馆、非遗中心的轻量进销存原型开发提供参考,帮助读者掌握事务处理、接口设计与前后端交互逻辑。
1. 古城文创商品订单与库存管理系统要解决的三件事
古城景区里一家文创店,货架上摆着一两百个 SKU:书签、冰箱贴、丝绸手帕、印章、明信片。节假日一天能出几百单,老板通常先用 Excel 记账,做着做着就会撞上三个问题:同一个书签在两个渠道同时卖出,表格显示还剩 3 件,实际早就卖空;月底盘点时账面和实物差十几件,说不清到底哪一单出问题;新招的店员不愿意在一堆表格里翻商品编码。
基于 Python 的古城文创商品订单与库存管理系统,对准的就是这三件事。商品档案、订单、订单明细、库存流水放进同一个本地数据库,下单时库存扣减和订单写入必须同时成功或同时失败;图形界面把下单、改价、查库存这些动作变成点几下按钮就能完成的操作;每一次库存变动都留一条流水,月底对不上账时能顺着流水倒查到具体订单号。
这套东西并不复杂,Python 标准库里自带的 sqlite3 和 tkinter 就撑得起一个单机版本,不用额外装数据库服务,也不会卡在 python 安装教程的配置环节上。正在写数据库课程设计的学生,或者真想给一家小店做个内部工具的人,都能照着走一遍。下面按数据库设计、GUI 搭建、订单与库存核心逻辑、进阶技巧的顺序展开,每一步都给能直接跑的代码。
2. 数据库设计:商品、订单、明细、库存流水四张表怎么落
数据层如果一开始做偏,后面 GUI 和业务逻辑都得返工。我一般先把表固定下来,再往上盖界面。古城文创的特性和普通零售不太一样:SKU 不多个别卖断货快,节假日单量是平时的好几倍,而且老板非常在意“哪一单把库存吃掉多少”。所以除了常规的商品表和订单表,库存流水这张表不能省。
2.1 四张主表的字段规划与类型选择
四张表的分工是这样的:product存商品档案和当前库存,orders存订单主表,order_item存每单买了哪些商品,stock_log存每次库存变动的流水。订单和明细拆开,是为了一个订单能装多个商品;明细里存一份下单时的单价快照,是因为文创商品经常调价,如果只存商品 ID,改价之后老订单金额就对不上了。
| 表名 | 作用 | 关键字段 | 注意点 |
|---|---|---|---|
| product | 商品档案与当前库存 | id, sku, name, category, price, stock, warn_stock | price 以「分」为单位存整数 |
| orders | 订单主表 | id, order_no, customer, total_amount, status | status 用状态机控制流转 |
| order_item | 订单明细 | order_id, product_id, qty, unit_price | unit_price 是下单时的单价快照 |
| stock_log | 库存流水 | product_id, change_qty, biz_type, ref_no | 只增不改,用于对账 |
金额字段这里要重点说一句。用REAL存价格看着方便,3.9 + 4.1这种运算在浮点数下会算出7.999999999999999,一个月下来累计误差可能就差出几块钱。稳妥做法是统一存整数「分」,显示时再除以 100。库存字段加CHECK (stock >= 0),是在数据库层面再加一道保险,即使业务代码写漏了,也不会出现负库存这种脏数据。
stock_log里的biz_type我一般固定成四个值:SALE销售出库、CANCEL取消回补、INBOUND采购入库、ADJUST手工盘点调整。ref_no记订单号或者盘点单号,将来对账时直接按这个字段分组就够用了。
2.2 用 sqlite3 跑通建库与样例数据脚本
环境用 Python 3.10 以上、PyCharm 或者 VSCode 都行,sqlite3 是标准库,不需要 pip 安装。下面这段脚本一次跑完就能得到建好表、灌好样例数据的gucheng.db。
import sqlite3 from pathlib import Path DB_PATH = Path(__file__).parent / "gucheng.db" DDL = """ CREATE TABLE IF NOT EXISTS product ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku TEXT NOT NULL UNIQUE, -- 商品编码,如 GC-BS-001 name TEXT NOT NULL, category TEXT NOT NULL DEFAULT '文创', price INTEGER NOT NULL CHECK (price >= 0), -- 单位:分 stock INTEGER NOT NULL DEFAULT 0 CHECK (stock >= 0), warn_stock INTEGER NOT NULL DEFAULT 5, -- 低于此值触发预警 updated_at TEXT NOT NULL DEFAULT (datetime('now','localtime')) ); CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_no TEXT NOT NULL UNIQUE, customer TEXT NOT NULL DEFAULT '散客', total_amount INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL DEFAULT 'CREATED', -- CREATED/PAID/CANCELLED created_at TEXT NOT NULL DEFAULT (datetime('now','localtime')) ); CREATE TABLE IF NOT EXISTS order_item ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL REFERENCES orders(id), product_id INTEGER NOT NULL REFERENCES product(id), qty INTEGER NOT NULL CHECK (qty > 0), unit_price INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS stock_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_id INTEGER NOT NULL REFERENCES product(id), change_qty INTEGER NOT NULL, -- 出库为负,入库为正 biz_type TEXT NOT NULL, -- SALE/CANCEL/INBOUND/ADJUST ref_no TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now','localtime')) ); CREATE INDEX IF NOT EXISTS idx_item_order ON order_item(order_id); CREATE INDEX IF NOT EXISTS idx_item_product ON order_item(product_id); CREATE INDEX IF NOT EXISTS idx_log_product ON stock_log(product_id, created_at); """ SEED = """ INSERT OR IGNORE INTO product (sku, name, category, price, stock, warn_stock) VALUES ('GC-BS-001', '古城手绘书签·城墙', '书签', 1280, 60, 10), ('GC-BS-002', '古城手绘书签·钟楼', '书签', 1280, 45, 10), ('GC-FR-001', '古城冰箱贴·四景套装', '冰箱贴', 3900, 30, 5), ('GC-SK-001', '丝绸手帕·青砖纹', '丝织品', 8900, 12, 3), ('GC-ST-001', '古城印章·旅行款', '文具', 2600, 8, 5); """ def init_db() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH) conn.execute("PRAGMA foreign_keys = ON") conn.executescript(DDL) conn.executescript(SEED) conn.commit() return conn if __name__ == "__main__": init_db() print("建库完成:", DB_PATH)这段脚本里有几个地方值得停一下。INSERT OR IGNORE配合sku上的UNIQUE,重复执行脚本不会灌出多份样例数据,方便反复调试。executescript一次提交多条 DDL,比一条一条execute干净。datatime('now','localtime')用的是本地时间,如果服务器在 UTC 时区,记得改成显式写入时间戳,否则盘点时段会对不齐。
2.3 三个必须打开的 SQLite 开关:外键、WAL 与忙等待
SQLite 默认不开外键约束,REFERENCES只写在注释里有效。每次连接后第一件事就是PRAGMA foreign_keys = ON,删商品时如果有明细引用就会直接报错,防止删出孤儿数据。第二个是 WAL 模式,写入时读操作不被阻塞,对单机 GUI 意义不大,但如果店员用软件的同时老板那边跑一个报表查询,体验会明显不同。第三个是busy_timeout,设成 5000 毫秒,两个连接撞到一起时自动等待而不是立刻抛database is locked。
def connect(): conn = sqlite3.connect(DB_PATH, timeout=5.0) conn.execute("PRAGMA foreign_keys = ON") conn.execute("PRAGMA journal_mode = WAL") conn.execute("PRAGMA busy_timeout = 5000") conn.row_factory = sqlite3.Row # 查询结果支持按列名取值 return connrow_factory = sqlite3.Row这行是省事的关键,后面写界面代码时直接row["name"]就行,不用靠下标记字段顺序。索引方面,order_item(order_id)和stock_log(product_id, created_at)这两条是必须的,前者撑订单详情页,后者撑按商品按时间的流水查询,缺了它们数据到几千条时界面就会明显卡顿。
3. Tkinter GUI:把订单与库存管理做成店员能用的界面
单机版没必要上 Web 框架,tkinter 加上 ttk 主题控件足够。界面我一般切成三块:左边是商品列表(用ttk.Treeview),右边是下单面板,底部留一条状态栏显示库存预警。关键不是画得好看,而是让店员不用记商品编码——点一行商品,编码自动填进下单框。
3.1 主窗口骨架与 ttk.Treeview 商品列表
先把窗口和表格搭起来,数据从 service 层取,界面本身不碰 SQL。这样做的好处是以后想把数据源换成 MySQL,界面代码一行都不用动。
import tkinter as tk from tkinter import ttk, messagebox class InventoryApp(tk.Tk): def __init__(self, service): super().__init__() self.service = service self.title("古城文创商品订单与库存管理系统") self.geometry("1040x640") self._build_product_table() self._build_order_panel() self.refresh_products() def _build_product_table(self): frame = ttk.LabelFrame(self, text="商品库存") frame.place(x=10, y=10, width=640, height=600) cols = ("sku", "name", "category", "price", "stock") self.tree = ttk.Treeview(frame, columns=cols, show="headings", height=22) headers = (("sku", "编码", 120), ("name", "名称", 210), ("category", "分类", 90), ("price", "单价", 90), ("stock", "库存", 70)) for key, text, width in headers: self.tree.heading(key, text=text) self.tree.column(key, width=width, anchor="center") self.tree.column("name", anchor="w") self.tree.pack(fill="both", expand=True, padx=6, pady=6) self.tree.bind("<<TreeviewSelect>>", self.on_pick_product) def refresh_products(self): self.tree.delete(*self.tree.get_children()) for row in self.service.list_products(): self.tree.insert("", "end", values=( row["sku"], row["name"], row["category"], f'{row["price"] / 100:.2f}', row["stock"]))show="headings"隐藏了 Treeview 默认的第一列空树列,视觉上干净。price / 100在这里把「分」转回元显示,只在界面层做格式化。<<TreeviewSelect>>是 Treeview 的选中事件,绑定到on_pick_product后,点哪一行都能拿到对应商品。
3.2 下单面板与库存实时校验联动
右侧下单面板只放三个输入项:商品编码、数量、客户名称,加一个提交按钮和一个流水查询按钮。库存校验分两次,一次在点击提交时由业务层校验,一次在输入数量失焦时给个即时提示,后者只是提醒,不能替代前者。
def _build_order_panel(self): panel = ttk.LabelFrame(self, text="下单") panel.place(x=660, y=10, width=370, height=300) ttk.Label(panel, text="商品编码").grid(row=0, column=0, padx=8, pady=8, sticky="e") self.var_sku = tk.StringVar() ttk.Entry(panel, textvariable=self.var_sku, width=22).grid(row=0, column=1) ttk.Label(panel, text="数量").grid(row=1, column=0, padx=8, pady=8, sticky="e") self.var_qty = tk.StringVar(value="1") ttk.Entry(panel, textvariable=self.var_qty, width=22).grid(row=1, column=1) ttk.Label(panel, text="客户").grid(row=2, column=0, padx=8, pady=8, sticky="e") self.var_customer = tk.StringVar(value="散客") ttk.Entry(panel, textvariable=self.var_customer, width=22).grid(row=2, column=1) ttk.Button(panel, text="提交订单", command=self.on_submit).grid( row=3, column=0, columnspan=2, pady=14) ttk.Button(panel, text="查看库存流水", command=self.on_show_log).grid( row=4, column=0, columnspan=2) def on_pick_product(self, _event): item = self.tree.selection() if item: self.var_sku.set(self.tree.item(item[0], "values")[0]) def on_submit(self): try: qty = int(self.var_qty.get()) order_no = self.service.create_order( self.var_sku.get().strip(), qty, self.var_customer.get().strip()) except ValueError: messagebox.showwarning("输入有误", "数量必须是整数") return except Exception as exc: messagebox.showerror("下单失败", str(exc)) return messagebox.showinfo("下单成功", f"订单号:{order_no}") self.refresh_products()提交后的流程是:业务层抛出异常,界面统一用messagebox提示,不把堆栈直接甩到用户面前。这里except ValueError要放在通用Exception前面,否则输入框里敲了字母也会走到下单逻辑里。商品编码这里用的是「点选自动填充 + 手工可改」的组合,比下拉框更适合一屏能看完全部商品的小店。
3.3 GUI 与数据层解耦:Service 层接口约定
界面代码里一行 SQL 都不该出现。我一般定一个InventoryService类,把界面需要的动作包成方法,界面只认方法名和返回值。这样换数据库、加缓存、写单元测试都从这一层下手。
| Service 方法 | 入参 | 返回 | 界面调用点 |
|---|---|---|---|
| list_products | 无 | sqlite3.Row 列表 | refresh_products |
| create_order | sku, qty, customer | 订单号字符串 | on_submit |
| cancel_order | order_no | 无 | 订单查询窗口 |
| list_stock_log | sku, limit | Row 列表 | on_show_log |
| low_stock_list | 无 | Row 列表 | 启动时预警弹窗 |
list_products返回的是sqlite3.Row,界面直接按列名取值,不做二次封装,少一层拷贝。create_order的入参用 sku 而不是 product_id,是为了让界面不依赖数据库主键,即使以后换库、换自增策略,界面代码也不受影响。
4. 订单与库存核心逻辑:事务边界、条件扣减与库存回补
超卖是这类系统里最要命的 bug:库存显示还有 5 件,实际卖出了 7 件。根因几乎都一样——读库存和改库存分成了两条独立语句,中间被别人插了一脚。解决办法是把「检查够不够」和「扣减」合并成一条带条件的 UPDATE,并用事务包住整条下单链路。
4.1 下单链路的事务边界怎么划
一个订单可能包含多个商品,要么全部扣减成功,要么一件都不扣。这决定了事务必须包住整个订单,而不是每个商品一个事务。BEGIN IMMEDIATE在这里比默认的延迟事务更合适,它在事务开始时就拿到写锁,避免读到库存后、准备写时才发现拿不到锁而回滚重来。
事务里的步骤是固定的:先插入orders拿到order_id,再逐个商品执行条件扣减,扣成功就往order_item和stock_log各写一条,最后更新订单总金额和状态。任何一步失败,整段rollback,数据库回到下单前的样子,不会出现「扣了库存却没有订单」的中间态。
class InventoryService: def __init__(self, conn_factory): self._connect = conn_factory def create_order(self, sku: str, qty: int, customer: str = "散客") -> str: conn = self._connect() try: conn.execute("BEGIN IMMEDIATE") row = conn.execute("SELECT id, price FROM product WHERE sku = ?", (sku,)).fetchone() if row is None: raise ValueError(f"商品不存在:{sku}") order_no = self._gen_order_no(conn) cur = conn.execute( "INSERT INTO orders(order_no, customer, total_amount, status) VALUES (?,?,?,?)", (order_no, customer, 0, "PAID")) order_id = cur.lastrowid # 条件扣减:只有库存足够时才会更新到 1 行 cur = conn.execute( "UPDATE product SET stock = stock - ?, " "updated_at = datetime('now','localtime') " "WHERE id = ? AND stock >= ?", (qty, row["id"], qty)) if cur.rowcount != 1: raise ValueError(f"库存不足:{sku} 当前仅剩 " f"{conn.execute('SELECT stock FROM product WHERE id=?', (row['id'],)).fetchone()[0]} 件") conn.execute("INSERT INTO order_item(order_id, product_id, qty, unit_price) " "VALUES (?,?,?,?)", (order_id, row["id"], qty, row["price"])) conn.execute("INSERT INTO stock_log(product_id, change_qty, biz_type, ref_no) " "VALUES (?,?,?,?)", (row["id"], -qty, "SALE", order_no)) conn.execute("UPDATE orders SET total_amount = ? WHERE id = ?", (row["price"] * qty, order_id)) conn.commit() return order_no except Exception: conn.rollback() raise finally: conn.close() def _gen_order_no(self, conn) -> str: n = conn.execute("SELECT COUNT(*) + 1 FROM orders").fetchone()[0] return f"GC{datetime.now():%Y%m%d}{n:04d}"rowcount != 1这个判断是整段的核心:库存不足时 UPDATE 影响 0 行,直接抛错,事务回滚。这种方式在 SQLite 里天然安全,因为它对写事务是串行的;换成 MySQL 时同样的写法要配合 InnoDB 行锁,思路完全一致。订单号用「日期 + 当日序号」拼,简单直观,缺点是并发下单时可能撞号——单机场景问题不大,多机部署时换成数据库自增或雪花 ID 更稳。
4.2 库存扣减的三种写法对比
同样的扣减逻辑,用不同写法,风险和性能差别很大。把常见写法放一起对比更清楚。
| 写法 | SQL 形态 | 是否防超卖 | 适用场景 |
|---|---|---|---|
| 先查后改 | SELECT 后 UPDATE | 否 | 单用户本地工具 |
| 条件更新 | UPDATE ... WHERE stock >= ? | 是 | 单机与多机通用,推荐 |
| 悲观锁 | SELECT ... FOR UPDATE 后 UPDATE | 是 | MySQL 高并发,SQLite 不支持 |
第二种是这里采用的方式,两次操作合并成一条,靠WHERE stock >= ?做判定。SQLite 没有SELECT ... FOR UPDATE,所以第三种写法在这儿用不上,但如果是把数据层换到 MySQL,行锁版本能扛住更密集的并发,配合连接池效果更好。
4.3 订单取消、库存回补与幂等
取消订单要比下单更小心,因为回补库存必须幂等——重复点两次取消按钮,库存只能回来一次。做法是先把订单状态从PAID改成CANCELLED,用条件更新保证只有第一次能改成功,改成功才继续回补库存。
def cancel_order(self, order_no: str) -> None: conn = self._connect() try: conn.execute("BEGIN IMMEDIATE") cur = conn.execute( "UPDATE orders SET status = 'CANCELLED' " "WHERE order_no = ? AND status = 'PAID'", (order_no,)) if cur.rowcount != 1: raise ValueError("订单不存在或已取消") order_id = conn.execute( "SELECT id FROM orders WHERE order_no = ?", (order_no,)).fetchone()[0] for it in conn.execute( "SELECT product_id, qty FROM order_item WHERE order_id = ?", (order_id,)): conn.execute( "UPDATE product SET stock = stock + ?, " "updated_at = datetime('now','localtime') WHERE id = ?", (it["qty"], it["product_id"])) conn.execute( "INSERT INTO stock_log(product_id, change_qty, biz_type, ref_no) " "VALUES (?,?,?,?)", (it["product_id"], it["qty"], "CANCEL", order_no)) conn.commit() except Exception: conn.rollback() raise finally: conn.close()注意WHERE status = 'PAID'这个条件,它就是幂等的开关:第二次调用时状态已经是CANCELLED,rowcount为 0,直接抛错,后面的回补逻辑根本不会执行。这个套路在电商系统里普遍使用,改成「已支付/已发货/已收货」的多状态机也是一样的写法,只需要把状态白名单换掉。
5. 库存预警、对账查询与打包分发的进阶技巧
系统能跑起来只是开始,真正让老板觉得有用的,是每天早上打开软件那一眼预警,以及月底对账时能不能三分钟说清差在哪。
5.1 库存预警与销量 TopN 的两条查询
预警查询很简单,一句 SQL 就能把低于阈值的商品捞出来,界面上可以直接弹窗或高亮行。销量排行用GROUP BY加窗口函数,SQLite 3.25 以上都支持。
-- 库存预警:低于预警线,按缺口从大到小 SELECT sku, name, stock, warn_stock FROM product WHERE stock <= warn_stock ORDER BY (warn_stock - stock) DESC; -- 近 30 天销量 TopN,排除已取消订单 SELECT p.sku, p.name, SUM(i.qty) AS sold FROM order_item i JOIN orders o ON o.id = i.order_id JOIN product p ON p.id = i.product_id WHERE o.status = 'PAID' AND o.created_at >= datetime('now', '-30 days', 'localtime') GROUP BY p.id ORDER BY sold DESC LIMIT 10;第一条要留意<=而不是<,预警线本身就应该触发提示。第二条里o.status = 'PAID'不能漏,否则取消掉的订单也会算进销量,补货就会多补。
5.2 账实不符时按流水倒查的三步
盘点发现实物比系统少,先别急着改库存。用stock_log把某个商品从某个时间点开始的流水全部拉出来,肉眼扫一遍通常就能看出问题:一笔异常大的 SALE,或者一条重复的 CANCEL。
-- 1) 看某个商品最近的全部流水 SELECT created_at, biz_type, change_qty, ref_no FROM stock_log WHERE product_id = (SELECT id FROM product WHERE sku = 'GC-ST-001') ORDER BY created_at DESC LIMIT 50; -- 2) 流水累计是否等于当前库存(应为 0,不为 0 说明有手工改过库存) SELECT p.stock - IFNULL(SUM(l.change_qty), 0) AS diff FROM product p LEFT JOIN stock_log l ON l.product_id = p.id WHERE p.sku = 'GC-ST-001'; -- 3) 找出只扣库存没生成订单的孤儿流水 SELECT l.* FROM stock_log l LEFT JOIN orders o ON o.order_no = l.ref_no WHERE l.biz_type = 'SALE' AND o.id IS NULL;第二条查询是这套系统的体检项:diff为 0 说明库存完全由流水推导出来,一分不差;不为 0 说明有人直接改过product.stock而没写流水,这类改动以后应该一律走ADJUST类型。第三条专门抓孤儿流水,通常来自事务中途崩溃或者老版本的代码缺陷。
改库存的正确姿势是在 Service 层加一个adjust_stock(sku, new_qty, reason)方法:算出差额,更新product.stock的同时补一条ADJUST流水,把reason写进ref_no。这样即使半年后有人问「这件丝绸手帕怎么突然多了 3 件」,流水里也留着一句「2024 盘点调整」。
软件交到店员手里之前,用 PyInstaller 打包成单个 exe,注意把gucheng.db放在 exe 同级的目录而不是打进包内,否则每次启动都是初始库存;--add-data只用来带图标和只读资源。命令大体是pyinstaller -F -w --name 古城文创库存 main.py,-w关掉黑窗口,-F出单文件。首次运行时如果数据库文件不存在,就在代码里调一次init_db()自动建库,比让店员手动放文件靠谱得多。
本文还有配套的精品资源,点击获取