去年年底帮朋友搭一套校园里的无人超市原型机,前端结算屏、后台进销存、门禁联动都要有,项目排期压得紧,最后用了Django加Flask这套Python双框架组合:Django管运营后台和核心数据,Flask跑门禁接口和轻量服务,两边共用同一个MySQL库。做完之后不少同行来问这套东西怎么落地,今天就把整个项目的设计思路、核心代码、部署要点一次性讲清楚。这篇内容既是项目复盘,也可以直接当一份“从零搭建无人超市管理系统”的实操手册来用,适合刚学完Python基础、想拿真实业务练手的同学,也适合准备做相关毕设或者内部演示系统的开发者参考。
1. 项目整体设计与技术选型思路
1.1 为什么是Django + Flask的双框架组合
先说选型。市面上做无人超市管理系统,基本是三种路子:全用Django、全用Flask、或者像我这套Django + Flask混合。前两种不是不行,但做下来你会发现各自的边界很明显。
Django的优势是“开箱即用全家桶”:自带Admin后台、ORM、认证体系、表单处理、中间件机制。无人超市最核心的运营端——商品录入、库存管理、订单查询、会员数据、报表统计——这些全是标准的CRUD加权限控制,Django的Admin后台能省掉80%的重复开发。你只需要定义好Model,注册到Admin,运营人员立刻就能上手维护数据,连前端页面都不用单独写。
Flask的优势是“轻、快、自由”。无人超市场景里有大量面向设备端的接口:扫码进门鉴权、结算台数据上报、门禁状态查询、摄像头回调。这些接口逻辑简单、要求响应快、不需要页面渲染,用Flask写一个路由加一个JSON返回就行。如果用Django来跑这类接口,要么挂到Django的URLconf里跟业务路由混在一起,要么单独起一个Django项目,反而笨重。
选双框架还有一个现实原因:两个服务可以分别部署、分别扩容。Flask服务挂了对核心运营数据没影响,Django挂了对门禁控制不影响,互不拖累。实际项目里我经常接到类似的咨询,很多人担心两个框架写同一个业务会乱,其实只要边界划清楚——Django管“人”(运营人员、后台管理),Flask管“物”(设备、门禁、识别终端)——代码结构会非常清晰。
1.2 无人超市的业务闭环与模块划分
无人超市跟普通电商系统的本质区别,在于它要把“线上购物流程”搬到“线下物理空间”里。用户在店里拿起商品就走,系统得自动知道他拿了什么、该扣多少钱。所以整套系统的业务闭环可以拆成五个环节:
拿货进场、挑选商品、自动结算、抵扣扣款、离店核验。对应到系统里,就是五大模块:门禁管理、商品管理、购物车与订单、会员账户、库存预警。下面这张表是每个模块的职责划分和技术落点:
| 模块 | 核心职责 | 主要技术落点 |
|---|---|---|
| 门禁管理 | 扫码进店、出场结算校验 | Flask接口 + 二维码Token + 继电器控制 |
| 商品管理 | 商品信息、分类、价格、上下架 | Django Model + Admin后台 |
| 购物车与订单 | 生成虚拟购物车、结算、支付模拟 | Redis缓存 + Django事务 + Flask结算接口 |
| 会员账户 | 余额、积分、消费记录 | Django Auth扩展 + Member Model |
| 库存预警 | 库存实时扣减、补货提醒 | Django信号 + 定时任务 + 库存日志表 |
这里要补充一个容易踩坑的点:无人超市的“购物车”并不像网页端那样存在Session里,而是以“用户身份 + 进场时间”为维度,结合RFID或者视觉识别系统上报的数据实时生成。换句话说,结算台不是自己算用户拿了什么,而是等设备告诉它用户拿了什么,系统再把这些条目组装成订单。所以购物车模块在设计时,要预留设备数据上报的接口通道。
1.3 数据模型设计:五张核心表
数据库直接用MySQL,字符集必须显式指定utf8mb4,否则中文商品名会乱码。项目里最核心的五张表,我直接贴出Django Model的写法,大家可以参考:
from django.db import models class Category(models.Model): name = models.CharField(max_length=50, unique=True, verbose_name="分类名称") sort = models.IntegerField(default=0, verbose_name="排序") created_at = models.DateTimeField(auto_now_add=True) class Meta: verbose_name = "商品分类" verbose_name_plural = verbose_name def __str__(self): return self.name class Product(models.Model): name = models.CharField(max_length=100, verbose_name="商品名称") category = models.ForeignKey(Category, on_delete=models.PROTECT, verbose_name="商品分类") price = models.DecimalField(max_digits=8, decimal_places=2, verbose_name="售价") stock = models.IntegerField(default=0, verbose_name="库存") warning_threshold = models.IntegerField(default=10, verbose_name="库存预警值") rfid_code = models.CharField(max_length=32, blank=True, null=True, unique=True, verbose_name="RFID编码") status = models.BooleanField(default=True, verbose_name="是否上架") image = models.ImageField(upload_to="products/", blank=True, null=True, verbose_name="商品图片") updated_at = models.DateTimeField(auto_now=True) class Meta: verbose_name = "商品" verbose_name_plural = verbose_name def __str__(self): return self.name class Order(models.Model): STATUS_CHOICES = ( ("pending", "待支付"), ("paid", "已支付"), ("canceled", "已取消"), ) order_no = models.CharField(max_length=32, unique=True, verbose_name="订单号") member = models.ForeignKey("Member", on_delete=models.SET_NULL, null=True, blank=True, verbose_name="会员") total_amount = models.DecimalField(max_digits=10, decimal_places=2, verbose_name="订单总额") status = models.CharField(max_length=10, choices=STATUS_CHOICES, default="pending", verbose_name="订单状态") created_at = models.DateTimeField(auto_now_add=True) paid_at = models.DateTimeField(null=True, blank=True, verbose_name="支付时间") class Meta: verbose_name = "订单" verbose_name_plural = verbose_name class OrderItem(models.Model): order = models.ForeignKey(Order, on_delete=models.CASCADE, related_name="items", verbose_name="所属订单") product = models.ForeignKey(Product, on_delete=models.PROTECT, verbose_name="商品") quantity = models.IntegerField(default=1, verbose_name="购买数量") price = models.DecimalField(max_digits=8, decimal_places=2, verbose_name="成交单价") class Meta: verbose_name = "订单明细" verbose_name_plural = verbose_name class Member(models.Model): openid = models.CharField(max_length=64, unique=True, verbose_name="会员标识") nickname = models.CharField(max_length=50, blank=True, verbose_name="昵称") balance = models.DecimalField(max_digits=10, decimal_places=2, default=0, verbose_name="账户余额") points = models.IntegerField(default=0, verbose_name="积分") created_at = models.DateTimeField(auto_now_add=True) class Meta: verbose_name = "会员" verbose_name_plural = verbose_name有一个细节必须说明:外键关联的on_delete参数,我特意把Product关联Category设成PROTECT,OrderItem关联Product也设成PROTECT。这意味着有订单的商品不允许直接删分类或删商品,只能下架。这是真实业务里必须守住的底线——历史订单的明细不能被物理删除,否则对账全乱了。很多人图省事设成CASCADE,一旦运营误删商品,连带历史订单全没了,这种事故在真实项目里属于P0级别。
库存字段用IntegerField而不是PositiveIntegerField,是因为后面要做库存扣减的并发控制,可能出现负数中间态,虽然最终会回滚,但不限制正数更灵活。价格必须用DecimalField,严禁用FloatField,浮点数算钱会在小数点后出现莫名其妙的0.001误差,这在结算环节不可接受。
2. 核心业务模块的实操实现
2.1 商品管理与库存预警机制
商品管理这部分,Django的Admin后台几乎是零成本实现。在应用的admin.py里写上:
from django.contrib import admin from .models import Category, Product, Order, OrderItem, Member @admin.register(Product) class ProductAdmin(admin.ModelAdmin): list_display = ("name", "category", "price", "stock", "status", "updated_at") list_filter = ("category", "status") search_fields = ("name", "rfid_code") list_editable = ("price", "stock", "status") list_per_page = 20list_editable这个配置强烈建议加上,运营人员在列表页直接改价格、库存、上下架状态,不需要点进详情页,工作效率提升非常明显。
库存预警是这个模块的核心功能。我的实现方式是两层:第一层是下单扣库存时的实时判断,第二层是每天凌晨的定时扫描。
下单扣库存的逻辑,直接上代码:
from django.db import transaction from django.db.models import F def deduct_stock(order_items): """ order_items: 列表,每个元素是 {"product_id": 1, "quantity": 2} """ with transaction.atomic(): for item in order_items: product = Product.objects.select_for_update().get(pk=item["product_id"]) if product.stock < item["quantity"]: raise ValueError(f"商品[{product.name}]库存不足") product.stock = F("stock") - item["quantity"] product.save() if product.stock <= product.warning_threshold: StockWarning.objects.create( product=product, current_stock=product.stock, warning_threshold=product.warning_threshold, )这里最关键的调用是select_for_update(),它会在数据库层面给这一行记录加行级锁。为什么要加锁?因为无人超市结算时,两个用户可能同时在两台结算台结账,如果不对商品行加锁,两个请求同时读到stock=5,同时扣掉3件,最后库存显示2,实际却卖出了6件,库存直接变成负数。这是典型的超卖问题,必须用悲观锁来兜底。
还有一个小经验:用F("stock") - item["quantity"]而不是直接读Python变量再赋值,是让扣减操作在数据库层面原子执行,避免读到脏数据。库存扣减完立刻检查预警阈值,触发预警后写一条StockWarning记录,运营后台里新增大屏提醒,比发邮件短信这类传统方式更直观。
定时扫描我用Django自带的manage.py命令配Cron实现,命令逻辑就是遍历所有商品,过滤stock <= warning_threshold且未删除的记录,批量生成预警。不用Celery,因为这个量级和频率用系统级调度就够了,没必要引入额外的消息队列组件。
2.2 购物车、订单与模拟支付闭环
无人超市的购物车,本质是一组设备上报的“用户拿了哪些商品”的数据集合。项目里我预留了一个Flask接口专门接收RFID结算台的清单上报,接口长这样:
from flask import Blueprint, request, jsonify device_api = Blueprint("device_api", __name__) @device_api.route("/api/device/cart", methods=["POST"]) def receive_cart(): """结算台上报用户购物车清单""" data = request.get_json() member_openid = data.get("openid") items = data.get("items") # [{"rfid_code": "xxxx", "quantity": 1}] # 查询商品信息,组装购物车 cart_items = [] total_amount = 0 for raw in items: product = Product.objects.filter(rfid_code=raw["rfid_code"]).first() if not product: return jsonify({"code": 1001, "msg": f"未识别商品: {raw['rfid_code']}"}), 404 cart_items.append({ "product_id": product.id, "name": product.name, "price": str(product.price), "quantity": raw["quantity"], "subtotal": str(product.price * raw["quantity"]), }) total_amount += product.price * raw["quantity"] # 购物车数据写入Redis,有效期约15分钟 cart_key = f"cart:{member_openid}" redis_client.setex(cart_key, 900, json.dumps(cart_items)) return jsonify({ "code": 0, "data": { "items": cart_items, "total_amount": str(total_amount), "expire_seconds": 900, } }), 200这里有个设计细节:设备上报的原始数据只认RFID编码,商品的具体名称、价格、库存全部回到主数据库实时查询。这么做是为了保证价格永远以数据库当前值为准,如果商品刚改过价,结算台不至于用到旧价格。
购物车确认后,用户在小程序或者结算屏上点击“确认支付”,走的是Django这边的下单接口。下单逻辑里,先查Redis拿到购物车数据,再扣库存、生成订单、模拟扣款:
from django.db import transaction from django.utils import timezone import uuid def create_order(member, cart_items): with transaction.atomic(): order = Order.objects.create( order_no=generate_order_no(), member=member, total_amount=sum(item["subtotal"] for item in cart_items), status="paid", paid_at=timezone.now(), ) for item in cart_items: OrderItem.objects.create( order=order, product_id=item["product_id"], quantity=item["quantity"], price=item["price"], ) # 扣减库存(包含预警检查) deduct_stock([{"product_id": item["product_id"], "quantity": item["quantity"]} for item in cart_items]) # 模拟余额扣款 member.balance -= order.total_amount member.points += int(order.total_amount) member.save() return order def generate_order_no(): return timezone.now().strftime("%Y%m%d%H%M%S") + uuid.uuid4().hex[:8].upper()订单号和支付时间放在同一个事务里,保证要么全部成功要么全部回滚。余额扣款这里没有做余额不足的校验,真实项目里必须在事务开始前查一次余额,不够就直接抛异常中止下单。
模拟支付是项目验收阶段最好演示的部分。我没有接真实支付渠道,而是直接在Member余额里扣款,同时在订单表记录支付时间。如果你想做得更逼真,可以接入支付宝沙箱或者微信支付测试号,思路就是在Order里加一个payment_status字段,支付回调里修改状态,再触发库存扣减和积分赠送。
2.3 会员账户与积分沉淀
会员模块相对简单,但也是运营闭环里不可或缺的一环。无人超市没有收银员,会员体系承载着用户身份识别和余额充值的功能。项目里的Member表设计得非常克制,只保留了openid、昵称、余额、积分四个核心字段。
用户第一次扫码进门时,Flask门禁接口会拿着小程序传过来的code去换openid,查不到Member记录就自动注册一个,初始赠送10元体验金。这10元体验金其实是运营策略的体现——让新用户有动力完成第一次进店消费,真实成本不高,但能推动整个结算流程被完整走一遍。
积分规则我设计成消费1元积1分,积分在后台可以兑换优惠券。后端实现非常简单,就是member.points += int(order.total_amount),但要记得在事务里一起提交。优惠券模块我拆成了独立的Coupon表,跟订单关联时校验有效期和使用门槛,这部分属于营销玩法,代码可以在Django Admin里管理,这里不再展开。
会员的消费记录查询,直接复用Order和OrderItem两张表,按Member外键过滤即可。运营后台可以查看每个会员的客单价、购买频次、偏好商品分类,这些数据将来接数据分析系统时会很有价值。
3. 无人值守场景下的进阶能力
3.1 扫码进门与门禁联动
无人超市的门禁控制,是整个系统最体现“无人”二字的部分。我的方案是:用户在门口屏幕扫码,Flask接口生成一个有效期30秒的一次性Token,门禁设备轮询鉴权,通过后由继电器控制电锁开门。
门禁接口的实现:
import time import secrets # Token存储,也可以用Redis door_tokens = {} @device_api.route("/api/door/token", methods=["POST"]) def issue_door_token(): data = request.get_json() openid = data.get("openid") if not openid: return jsonify({"code": 1001, "msg": "缺少openid"}), 400 member = Member.objects.filter(openid=openid).first() if not member: return jsonify({"code": 1002, "msg": "未注册用户"}), 404 token = secrets.token_hex(16) door_tokens[token] = { "openid": openid, "expire_at": time.time() + 30, } return jsonify({"code": 0, "data": {"token": token, "expire_at": door_tokens[token]["expire_at"]}}), 200 @device_api.route("/api/door/check", methods=["GET"]) def check_door_token(): token = request.args.get("token") if not token or token not in door_tokens: return jsonify({"code": 1003, "msg": "无效Token"}), 401 info = door_tokens[token] if time.time() > info["expire_at"]: del door_tokens[token] return jsonify({"code": 1004, "msg": "Token过期,请重新扫码"}), 401 return jsonify({"code": 0, "data": {"openid": info["openid"]}}), 200这里有两个实测下来的注意事项:
Token必须带过期时间,过期后立即从存储中删除。30秒看起来短,但足够完成扫码和门锁响应,太长的Token会给安全带来隐患——有人截获了Token就能随意进入。
门禁权限跟黑名单联动。如果用户的余额小于某个阈值,或者历史订单有恶意欠费记录,应该在issue_door_token这一步就把人拦在门外。这个逻辑我在早期版本里忘了加,后来补上去的。无人店不设防,但不代表可以无限宽纵用户。
如果想接物理门锁,买一个继电器模块,树莓派或者任意支持GPIO的开发板都行,Python里用RPi.GPIO库控制高低电平,接口返回“允许进入”时拉高引脚电平,继电器闭合,完成开锁。演示阶段用舵机搭一个可开合的门模型,视觉冲击力和演示效果都很好。
3.2 RFID与视觉识别方案怎么选
这是做无人超市项目时被问得最多的问题。商品识别到底用RFID还是视觉?我的建议很明确:预算有限、快速演示选RFID;预算充足、追求原生购物体验才考虑视觉。
RFID方案的原理是给每个商品贴一张高频RFID标签,结算台的读写器读取标签编号,编号与商品表里的rfid_code对应,一次批量读取十几件商品只需要两三秒。优势是技术成熟、识别准确率高、开发成本低。劣势是标签本身有成本,大概几毛钱到一块多一张,而且金属材质商品会对信号有干扰。
视觉识别方案分两步:先目标检测(用YOLO系列模型识别商品类别),再通过追踪算法统计拿了哪件、放回哪件。优势是商品无需额外贴标,购物体验自然。劣势是开发难度陡增,需要准备大量的商品图片做训练集,模型推理还得依赖GPU服务器。对于一个练手项目或者毕设项目来说,视觉方案可能的时间成本会超出预期。
我的建议是可以做“混合架构”:采集端预留两套接口,RFID和视觉识别模块上报的数据格式统一成{"rfid_code": "", "quantity": 1}这种结构。将来视觉识别模型成熟了,替换采集端,后端订单逻辑一行代码都不用改。这就是接口抽象带来的好处。
3.3 异常行为与离店防损策略
无人超市真正上了线,防损几乎是生死线。物理防盗设备(比如防盗门)只是一部分,系统层面的策略也很关键。
我做得比较有效的一个策略是重量校验。在结算台和出口设置称重传感器,进场记录一次体重,拿完商品离店再称一次,系统把重量差跟订单商品的理论重量做对比。这个策略能拦住部分“拿而不付”的行为,实现也不复杂,就是Flask接口上传称重数据,后端做阈值判断。
另一个策略是离店不结算拦截。门禁系统在用户出场时会检查该用户是否有未完成订单或者有未支付的购物车,有的话门禁保持关闭,屏幕提示去结算台完成支付。这个逻辑直接复用check_door_token接口,只是多加一个查询条件:
pending_cart = redis_client.get(f"cart:{openid}") if pending_cart: return jsonify({"code": 1005, "msg": "有未结算商品,请先完成支付"}), 402核心思路就是让门禁接口与业务状态联动,而不是独立地开锁关锁。
需要提醒的是,这里的策略都只是软件层的“威慑+止损”,真正上线的大型无人超市还需要结合摄像头和商品识别来做全链路追踪。练习项目做到门禁联动这个程度,已经可以完整演示无人零售的闭环了。
4. 部署上线与常见问题排查
4.1 本地开发环境搭建清单
先把本地环境跑起来。我的建议是全程用虚拟环境,用venv而不是直接装到系统Python里,避免多个项目依赖打架:
# 创建虚拟环境 python3 -m venv vuenv source vuenv/bin/activate # 安装依赖 pip install django flask pymysql gunicorn redis # 创建Django项目 django-admin startproject supermarket cd supermarket python manage.py startapp core python manage.py migrate # 创建Flask服务目录 mkdir browser_app数据库我用的MySQL8.0,Python连接MySQL用pymysql,需要在项目的__init__.py里挂上:
import pymysql pymysql.install_as_MySQLdb()Django的settings.py里,数据库配置、语言时区、静态文件这三个地方最容易出错:
DATABASES = { "default": { "ENGINE": "django.db.backends.mysql", "NAME": "supermarket", "USER": "root", "PASSWORD": "your_password", "HOST": "127.0.0.1", "PORT": "3306", "OPTIONS": { "charset": "utf8mb4", "init_command": "SET NAMES utf8mb4", }, } } LANGUAGE_CODE = "zh-hans" TIME_ZONE = "Asia/Shanghai" USE_TZ = True STATIC_URL = "/static/" STATIC_ROOT = BASE_DIR / "staticfiles" STATICFILES_DIRS = [BASE_DIR / "static"]4.2 部署到服务器的配置要点
本地跑通后,部署到云服务器上我用的是nginx + gunicorn的组合。Flask服务直接用gunicorn启动就行,Django也同样可以用gunicorn,不用刻意上uwsgi,现在的gunicorn性能足够,配置还简单。
gunicorn启动脚本:
gunicorn core.wsgi:application -w 4 -b 127.0.0.1:8000 --timeout 60 gunicorn browser_app.app:app -w 2 -b 127.0.0.1:8080第一行是Django服务,跑在8000端口;第二行是Flask服务,跑在8080端口。两个服务共用同一个MySQL,通过不同端口对外提供能力。
nginx配置里做两件事:一是把不同路径转发到不同服务,二是处理好静态文件和媒体文件。上传的商品图片由Django处理,存放在MEDIA_ROOT目录,nginx需要单独配置一个location来访问:
server { listen 80; server_name your_domain.com; location /static/ { alias /your_project_path/staticfiles/; } location /media/ { alias /your_project_path/media/; } location /api/device/ { proxy_pass http://127.0.0.1:8080; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } }部署完别忘了collectstatic,把所有静态文件收集到STATIC_ROOT目录:
python manage.py collectstatic --noinput4.3 高频报错排查实录
最后整理几个做这个项目时大家最容易踩的坑,全部实测过,直接对照排查:
问题1:Django页面样式全乱,静态文件加载不出来
八成是STATIC_URL配置和目录结构的问题。开发环境跑runserver时要手动加一段静态文件路由:
from django.conf import settings from django.conf.urls.static import static urlpatterns += static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT) urlpatterns += static(settings.STATIC_URL, document_root=settings.STATICFILES_DIRS[0])还有个绕弯的坑:在模板里写img标签引用静态图片,路径用{% static 'products/xxx.jpg' %},而不是直接写相对路径。直接写路径在开发环境可能碰巧能显示,一旦部署到nginx下面全丢。
问题2:Flask端报中文乱码
MySQL建库时必须指定utf8mb4,连接参数里的charset带上,Django和Flask两边都要设置。如果已经建库了,用ALTER DATABASE把字符集改掉,再检查表结构。乱码问题90%出在字符集,不是Python代码问题。
问题3:Django报CSRF验证失败
第一个是表单提交时模板里没加{% csrf_token %};第二个是Flask接口访问Django的接口时没有处理CSRF。解决方案是给Flask调用的那部分接口加上@csrf_exempt装饰器,因为设备端接口本身就不依赖Django的Session认证,没必要做CSRF校验。
from django.views.decorators.csrf import csrf_exempt @csrf_exempt def device_receive_cart(request): pass问题4:库存被扣成负数
排查询条件里是不是漏了select_for_update,或者把商品查询放在了事务外面。事务边界内的所有行锁必须集中在同一个with transaction.atomic()块里执行,提前查出来的对象不会自动加锁。
问题5:Flask上传图片后,附件路径一直报404
这是Flask跟Django的media目录没对齐的问题。Flask服务写文件的时候,保存路径要用Django的MEDIA_ROOT绝对路径,而不是Flask自己的static目录。我在项目里用一个共享环境变量来统一两个框架的媒体目录地址,一劳永逸。
问题6:redis连接被拒
检查密码配置和访问权限,然后确认Redis绑定的端口不是127.0.0.1,如果两台机器不在同一台主机,需要把bind配置改成0.0.0.0并设定密码,最后检查防火墙放行6379端口。
这个项目做到最后,我体会最深的一点是:无人超市管理系统听起来很“硬核”,但拆解下来,核心还是订单、库存、会员这些经典业务模块,加上设备接口的对接。Django负责把经典业务做到扎实,Flask负责把设备接入做到轻快,两者配合,整个闭环就通了。如果你正在准备类似的系统,先把基础模块跑通,再一个个加识别、门禁这些花活,稳扎稳打比什么都重要。