去年年底帮朋友搭一套校园里的无人超市原型机前端结算屏、后台进销存、门禁联动都要有项目排期压得紧最后用了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_length50, uniqueTrue, verbose_name分类名称) sort models.IntegerField(default0, verbose_name排序) created_at models.DateTimeField(auto_now_addTrue) class Meta: verbose_name 商品分类 verbose_name_plural verbose_name def __str__(self): return self.name class Product(models.Model): name models.CharField(max_length100, verbose_name商品名称) category models.ForeignKey(Category, on_deletemodels.PROTECT, verbose_name商品分类) price models.DecimalField(max_digits8, decimal_places2, verbose_name售价) stock models.IntegerField(default0, verbose_name库存) warning_threshold models.IntegerField(default10, verbose_name库存预警值) rfid_code models.CharField(max_length32, blankTrue, nullTrue, uniqueTrue, verbose_nameRFID编码) status models.BooleanField(defaultTrue, verbose_name是否上架) image models.ImageField(upload_toproducts/, blankTrue, nullTrue, verbose_name商品图片) updated_at models.DateTimeField(auto_nowTrue) 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_length32, uniqueTrue, verbose_name订单号) member models.ForeignKey(Member, on_deletemodels.SET_NULL, nullTrue, blankTrue, verbose_name会员) total_amount models.DecimalField(max_digits10, decimal_places2, verbose_name订单总额) status models.CharField(max_length10, choicesSTATUS_CHOICES, defaultpending, verbose_name订单状态) created_at models.DateTimeField(auto_now_addTrue) paid_at models.DateTimeField(nullTrue, blankTrue, verbose_name支付时间) class Meta: verbose_name 订单 verbose_name_plural verbose_name class OrderItem(models.Model): order models.ForeignKey(Order, on_deletemodels.CASCADE, related_nameitems, verbose_name所属订单) product models.ForeignKey(Product, on_deletemodels.PROTECT, verbose_name商品) quantity models.IntegerField(default1, verbose_name购买数量) price models.DecimalField(max_digits8, decimal_places2, verbose_name成交单价) class Meta: verbose_name 订单明细 verbose_name_plural verbose_name class Member(models.Model): openid models.CharField(max_length64, uniqueTrue, verbose_name会员标识) nickname models.CharField(max_length50, blankTrue, verbose_name昵称) balance models.DecimalField(max_digits10, decimal_places2, default0, verbose_name账户余额) points models.IntegerField(default0, verbose_name积分) created_at models.DateTimeField(auto_now_addTrue) class Meta: verbose_name 会员 verbose_name_plural verbose_name有一个细节必须说明外键关联的on_delete参数我特意把Product关联Category设成PROTECTOrderItem关联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(pkitem[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( productproduct, current_stockproduct.stock, warning_thresholdproduct.warning_threshold, )这里最关键的调用是select_for_update()它会在数据库层面给这一行记录加行级锁。为什么要加锁因为无人超市结算时两个用户可能同时在两台结算台结账如果不对商品行加锁两个请求同时读到stock5同时扣掉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_coderaw[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 fcart:{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_nogenerate_order_no(), membermember, total_amountsum(item[subtotal] for item in cart_items), statuspaid, paid_attimezone.now(), ) for item in cart_items: OrderItem.objects.create( orderorder, product_iditem[product_id], quantityitem[quantity], priceitem[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(openidopenid).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(fcart:{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.0Python连接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 高频报错排查实录最后整理几个做这个项目时大家最容易踩的坑全部实测过直接对照排查问题1Django页面样式全乱静态文件加载不出来八成是STATIC_URL配置和目录结构的问题。开发环境跑runserver时要手动加一段静态文件路由from django.conf import settings from django.conf.urls.static import static urlpatterns static(settings.MEDIA_URL, document_rootsettings.MEDIA_ROOT) urlpatterns static(settings.STATIC_URL, document_rootsettings.STATICFILES_DIRS[0])还有个绕弯的坑在模板里写img标签引用静态图片路径用{% static products/xxx.jpg %}而不是直接写相对路径。直接写路径在开发环境可能碰巧能显示一旦部署到nginx下面全丢。问题2Flask端报中文乱码MySQL建库时必须指定utf8mb4连接参数里的charset带上Django和Flask两边都要设置。如果已经建库了用ALTER DATABASE把字符集改掉再检查表结构。乱码问题90%出在字符集不是Python代码问题。问题3Django报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()块里执行提前查出来的对象不会自动加锁。问题5Flask上传图片后附件路径一直报404这是Flask跟Django的media目录没对齐的问题。Flask服务写文件的时候保存路径要用Django的MEDIA_ROOT绝对路径而不是Flask自己的static目录。我在项目里用一个共享环境变量来统一两个框架的媒体目录地址一劳永逸。问题6redis连接被拒检查密码配置和访问权限然后确认Redis绑定的端口不是127.0.0.1如果两台机器不在同一台主机需要把bind配置改成0.0.0.0并设定密码最后检查防火墙放行6379端口。这个项目做到最后我体会最深的一点是无人超市管理系统听起来很“硬核”但拆解下来核心还是订单、库存、会员这些经典业务模块加上设备接口的对接。Django负责把经典业务做到扎实Flask负责把设备接入做到轻快两者配合整个闭环就通了。如果你正在准备类似的系统先把基础模块跑通再一个个加识别、门禁这些花活稳扎稳打比什么都重要。
基于Django+Flask的无人超市管理系统架构与实现
去年年底帮朋友搭一套校园里的无人超市原型机前端结算屏、后台进销存、门禁联动都要有项目排期压得紧最后用了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_length50, uniqueTrue, verbose_name分类名称) sort models.IntegerField(default0, verbose_name排序) created_at models.DateTimeField(auto_now_addTrue) class Meta: verbose_name 商品分类 verbose_name_plural verbose_name def __str__(self): return self.name class Product(models.Model): name models.CharField(max_length100, verbose_name商品名称) category models.ForeignKey(Category, on_deletemodels.PROTECT, verbose_name商品分类) price models.DecimalField(max_digits8, decimal_places2, verbose_name售价) stock models.IntegerField(default0, verbose_name库存) warning_threshold models.IntegerField(default10, verbose_name库存预警值) rfid_code models.CharField(max_length32, blankTrue, nullTrue, uniqueTrue, verbose_nameRFID编码) status models.BooleanField(defaultTrue, verbose_name是否上架) image models.ImageField(upload_toproducts/, blankTrue, nullTrue, verbose_name商品图片) updated_at models.DateTimeField(auto_nowTrue) 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_length32, uniqueTrue, verbose_name订单号) member models.ForeignKey(Member, on_deletemodels.SET_NULL, nullTrue, blankTrue, verbose_name会员) total_amount models.DecimalField(max_digits10, decimal_places2, verbose_name订单总额) status models.CharField(max_length10, choicesSTATUS_CHOICES, defaultpending, verbose_name订单状态) created_at models.DateTimeField(auto_now_addTrue) paid_at models.DateTimeField(nullTrue, blankTrue, verbose_name支付时间) class Meta: verbose_name 订单 verbose_name_plural verbose_name class OrderItem(models.Model): order models.ForeignKey(Order, on_deletemodels.CASCADE, related_nameitems, verbose_name所属订单) product models.ForeignKey(Product, on_deletemodels.PROTECT, verbose_name商品) quantity models.IntegerField(default1, verbose_name购买数量) price models.DecimalField(max_digits8, decimal_places2, verbose_name成交单价) class Meta: verbose_name 订单明细 verbose_name_plural verbose_name class Member(models.Model): openid models.CharField(max_length64, uniqueTrue, verbose_name会员标识) nickname models.CharField(max_length50, blankTrue, verbose_name昵称) balance models.DecimalField(max_digits10, decimal_places2, default0, verbose_name账户余额) points models.IntegerField(default0, verbose_name积分) created_at models.DateTimeField(auto_now_addTrue) class Meta: verbose_name 会员 verbose_name_plural verbose_name有一个细节必须说明外键关联的on_delete参数我特意把Product关联Category设成PROTECTOrderItem关联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(pkitem[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( productproduct, current_stockproduct.stock, warning_thresholdproduct.warning_threshold, )这里最关键的调用是select_for_update()它会在数据库层面给这一行记录加行级锁。为什么要加锁因为无人超市结算时两个用户可能同时在两台结算台结账如果不对商品行加锁两个请求同时读到stock5同时扣掉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_coderaw[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 fcart:{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_nogenerate_order_no(), membermember, total_amountsum(item[subtotal] for item in cart_items), statuspaid, paid_attimezone.now(), ) for item in cart_items: OrderItem.objects.create( orderorder, product_iditem[product_id], quantityitem[quantity], priceitem[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(openidopenid).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(fcart:{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.0Python连接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 高频报错排查实录最后整理几个做这个项目时大家最容易踩的坑全部实测过直接对照排查问题1Django页面样式全乱静态文件加载不出来八成是STATIC_URL配置和目录结构的问题。开发环境跑runserver时要手动加一段静态文件路由from django.conf import settings from django.conf.urls.static import static urlpatterns static(settings.MEDIA_URL, document_rootsettings.MEDIA_ROOT) urlpatterns static(settings.STATIC_URL, document_rootsettings.STATICFILES_DIRS[0])还有个绕弯的坑在模板里写img标签引用静态图片路径用{% static products/xxx.jpg %}而不是直接写相对路径。直接写路径在开发环境可能碰巧能显示一旦部署到nginx下面全丢。问题2Flask端报中文乱码MySQL建库时必须指定utf8mb4连接参数里的charset带上Django和Flask两边都要设置。如果已经建库了用ALTER DATABASE把字符集改掉再检查表结构。乱码问题90%出在字符集不是Python代码问题。问题3Django报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()块里执行提前查出来的对象不会自动加锁。问题5Flask上传图片后附件路径一直报404这是Flask跟Django的media目录没对齐的问题。Flask服务写文件的时候保存路径要用Django的MEDIA_ROOT绝对路径而不是Flask自己的static目录。我在项目里用一个共享环境变量来统一两个框架的媒体目录地址一劳永逸。问题6redis连接被拒检查密码配置和访问权限然后确认Redis绑定的端口不是127.0.0.1如果两台机器不在同一台主机需要把bind配置改成0.0.0.0并设定密码最后检查防火墙放行6379端口。这个项目做到最后我体会最深的一点是无人超市管理系统听起来很“硬核”但拆解下来核心还是订单、库存、会员这些经典业务模块加上设备接口的对接。Django负责把经典业务做到扎实Flask负责把设备接入做到轻快两者配合整个闭环就通了。如果你正在准备类似的系统先把基础模块跑通再一个个加识别、门禁这些花活稳扎稳打比什么都重要。
