基于Python的Django+Flask畜牧站疾病防控与检测系统实战解析

做基层畜牧站的信息化项目最头疼的不是算法而是把一堆琐碎的防疫流程理顺。最近在开发一个基于Python的畜牧站疾病防控与检测系统技术栈选了DjangoFlask这个组合。很多人第一反应是“一个项目为什么要混用两个框架”其实真正落地之后你会发现这不是炫技而是被业务特点逼出来的选择。畜牧站的业务既有相对稳定的档案管理、任务流转又有变化频繁的采样检测、数据上报用Django承担核心业务系统用Flask单独维护轻量化的检测数据服务既能快速交付又不会让代码纠缠成一团。这个系统能做的事情简单说就是把“养殖档案-免疫记录-疫病上报-采样检测-预警处置”这条链路全部打通。以前工作人员用Excel登记、微信群上报、电话通知检测结果现在所有数据都进系统关键节点自动提醒检测结果自动回传并生成预警站长打开大屏就能看到辖区内的疫情风险。如果你也需要做类似的基层政企管理系统、防疫类信息平台或者只是想看看Django和Flask怎么在一个项目里优雅共存这篇内容应该能给你一个比较完整的参考。1. 整体架构拆解为什么是Django Flask 的组合1.1 拆解业务需求畜牧站真正要解决什么问题我在动手之前先把某畜牧站的实际工作流程完整捋了一遍。他们日常工作大概有这么几块辖区内养殖场户的畜禽存栏档案、疫苗免疫记录、日常巡查上报、可疑疫情报告、采样送检、检测结果登记、阳性处置、以及向上级部门报的各类汇总表。这些流程有个共同特点核心数据模型相对固定但业务状态一直在变。比如一头牛从入场、免疫、检疫、采样到处置中间要经历很多状态流转而检测项目又经常因为上级要求调整有时要增加新的病种检测有时要改变判定标准。如果把这些全部塞进一个高度耦合的系统里每次需求变更都要动核心表结构风险很大。所以我最初定的目标是核心业务模块要稳定、收敛能通过配置适应变化检测相关模块要独立、灵活能快速对接外部设备和系统。说白了这就是典型的“核心系统重稳定、边缘模块重迭代”的思路。Django的Admin、ORM和权限体系非常适合做核心管理端而Flask的轻量、灵活非常适合做数据接收和中间服务。1.2 技术选型逻辑Django负责核心业务Flask负责轻量检测服务先说为什么不用单独的Django或单独的Flask完成全部功能。如果只用Django检测数据的接收和解析逻辑会慢慢被业务代码同化尤其当外部检测仪器通过HTTP回调推送数据时你得在主项目里增加一堆配置代码和异常处理时间久了views.py会非常臃肿。如果只用Flask虽然轻但重型的用户权限管理、数据模型迁移、Admin后台都要自己搭工期会成倍增加。于是我把系统拆成两个服务主服务用Django负责日常业务管理、账号权限、审批流、数据展示检测服务用Flask专门接收检测设备上传的数据、解析样本结果、调用判定规则再通过内部API把判定结果推回Django主服务。两个服务共享同一个基础数据库但Django拥有核心表结构Flask只通过约定好的视图或者服务账号读写检测相关表避免抢Key。这样做的直接好处是Flask服务挂了主业务还能继续录档案、做免疫记录Django服务要升级也不影响检测设备数据持续上传。两个服务可以分别部署、分别扩缩容排查问题的时候边界也清楚。1.3 部署形态与目录结构实际的代码仓库我是分成两个工程的但放在同一个git仓库的apps目录下方便统一管理版本。大概是这样project_root/ ├── django_server/ │ ├── manage.py │ ├── config/ │ │ ├── settings/ │ │ ├── urls.py │ │ └── wsgi.py │ ├── apps/ │ │ ├── farms/ │ │ ├── immunization/ │ │ ├── disease_report/ │ │ └── early_warning/ │ └── static/ ├── flask_server/ │ ├── app.py │ ├── core/ │ │ ├── db.py │ │ ├── validator.py │ │ └── rules.py │ └── routes/ │ ├── sample_upload.py │ └── result_callback.py ├── deploy/ │ ├── docker-compose.yml │ ├── nginx.conf │ └── supervisor/ └── docs/线上部署时Django跑在Gunicorn上Flask跑在单独的一个Gunicorn worker组里前面统一挂Nginx做反向代理和静态文件服务。数据库用的是PostgreSQL缓存和消息队列用Redis。这个架构不复杂但运维起来非常清晰不必为了所谓微服务搞一堆注册中心。2. 核心功能实现从疾病上报到检测闭环2.1 养殖档案与免疫记录建模养殖档案是整个系统的地基。一开始很容易把“养殖场”和“批次”混在一个表里但实际操作中一个养殖场可能有多个批次每个批次的免疫状态、健康状态都不一样。所以我把模型拆成Farms、Batches、Animals三个层次个体动物只有在需要追溯时才会录入一般情况只管理到批次。免疫记录是另一个关键模型直接关系到后面的预警分析。每一批次动物的疫苗种类、免疫日期、疫苗批号、免疫剂量都要记录。这里容易踩坑的是“免疫状态”字段设计不要只存一个布尔值因为实际业务里存在“已免疫、免疫中、免疫失败、已补免”多种状态。我建议用一个整数字段表示状态并且用Django的TextChoices做枚举约束。from django.db import models from django.utils.translation import gettext_lazy as _ class ImmunizationRecord(models.Model): class ImmunizationStatus(models.TextChoices): PENDING pending, _(待免疫) IMMUNIZED immunized, _(已免疫) FAILED failed, _(免疫失败) REWORK rework, _(已补免) batch models.ForeignKey(farms.Batch, on_deletemodels.CASCADE, related_nameimmunizations) vaccine_type models.CharField(疫苗种类, max_length100) vaccine_brand models.CharField(疫苗品牌, max_length100) vaccine_batch_no models.CharField(疫苗批号, max_length100) immunized_at models.DateField(免疫日期) status models.CharField(免疫状态, max_length20, choicesImmunizationStatus.choices, defaultImmunizationStatus.PENDING) operator models.ForeignKey(users.User, on_deletemodels.PROTECT, verbose_name操作人) remark models.TextField(备注, blankTrue) created_at models.DateTimeField(auto_now_addTrue) class Meta: db_table immunization_record indexes [ models.Index(fields[batch, vaccine_type, immunized_at]), ]免疫记录为什么要单独建索引因为实际查询中最常见的就是“某个批次最近一次打了什么疫苗”“某段时间内哪些批次该补免了”。加了联合索引之后这类查询速度会快很多。基层系统数据量不算大但索引用不用差距还是很明显的尤其到年底汇总报表的时候。2.2 疾病上报与防控任务流转疫病上报模块是整个系统里最需要“流程思维”的部分。一个报上来的可疑疫情不是记录一下就完事它要经过“待审核、待采样、检测中、待处置、已处置、已结案”等多个状态。我在设计时没有用复杂的流程引擎而是用了一个state字段外加一张流转记录表来存操作轨迹。class DiseaseReport(models.Model): batch models.ForeignKey(farms.Batch, on_deletemodels.CASCADE, related_namedisease_reports) disease_name models.CharField(疑似病种, max_length100) symptom models.TextField(临床症状描述) report_org models.CharField(报告单位, max_length200) report_time models.DateTimeField(报告时间) state models.CharField(流转状态, max_length20, defaultpending) current_handler models.ForeignKey(users.User, on_deletemodels.PROTECT, related_namedisease_tasks, verbose_name当前处理人) ... class DiseaseFlowLog(models.Model): report models.ForeignKey(DiseaseReport, on_deletemodels.CASCADE, related_namelogs) from_state models.CharField(原状态, max_length20) to_state models.CharField(新状态, max_length20) operator models.ForeignKey(users.User, on_deletemodels.PROTECT, verbose_name操作人) operation models.CharField(操作类型, max_length50) comment models.TextField(办理意见, blankTrue) created_at models.DateTimeField(auto_now_addTrue)状态流转我没有硬编码在views里而是抽了一个简单的状态机工具类每个状态定义了允许执行的操作和下个状态。这样好处是需求方说要增加一个“待复核”状态时只需改状态机配置不用动一堆页面逻辑。防控任务是根据上报单自动或人工生成的。比如某批次确诊某种传染病系统会自动生成“紧急免疫任务”和“场所消毒任务”并分配到对应责任人。这里要特别注意的是任务生成必须支持重复性因为一个疫情往往对应多个任务如果任务表直接挂在上报单下一个上报单就只会生成一条记录后面扩展就麻烦了。2.3 采样检测数据接入Flask API与异步任务检测数据是整个系统中变化最频繁的部分。仪器设备的接口不同有的传JSON有的传XML有的甚至只传一个CSV文件。如果把这些解析逻辑全部放进Django主工程每次对接一种新设备就得改业务代码风险太大。所以我用Flask单独做了一个检测数据接收服务。设计思路是这样的采样人员在Django系统里登记一个“送检单”系统为每个样本生成唯一sample_code二维码打印出来贴到采样管上。检测实验室拿到样本后通过扫码枪或设备接口把样本号和检测结果传入Flask服务。Flask服务只做三件事认证请求来源并校验sample_code是否存在。解析检测数据按配置的判定规则计算阴阳性。将结果写入检测结果表并通过内部回调通知Django更新送检单状态。# flask_server/routes/sample_upload.py from flask import Blueprint, request, jsonify from core.validator import validate_sample_code, validate_result_data from core.rules import judge_result from core.db import get_db sample_bp Blueprint(sample, __name__) sample_bp.route(/api/v1/results, methods[POST]) def upload_result(): req request.get_json() if not req: return jsonify({code: 400, msg: 请求体不能为空}), 400 sample_code req.get(sample_code) raw_result req.get(raw_result) if not validate_sample_code(sample_code): return jsonify({code: 404, msg: 样本编号不存在}), 404 if not validate_result_data(raw_result): return jsonify({code: 422, msg: 检测数据不完整}), 422 final_result judge_result(raw_result) db get_db() db.execute( UPDATE sample_result SET raw_result ?, final_result ?, updated_at NOW() WHERE sample_code ?, (raw_result, final_result, sample_code), ) db.commit() # 通知Django服务刷新送检单状态 notify_django(sample_code, final_result) return jsonify({code: 200, msg: ok, data: {sample_code: sample_code, final_result: final_result}})这里有几个细节值得注意。第一样本编号一旦生成在生命周期内不允许重复数据库里一定要加唯一约束。第二检测结果解析不能直接在请求里同步做复杂判定否则设备多、并发高的时候Flask服务会被拖垮。实际项目中我会把判定规则收敛到一个独立模块里先同步校验数据完整性再把详细判定任务交给Celery去执行。2.4 疫情预警与可视化展示预警模块是这套系统最能体现价值的地方。每天凌晨Django的定时任务会扫描免疫记录、检测结果、上报单三个表根据预设规则产生预警消息。比如某批次距离上次免疫超过建议间隔天数触发“补免提醒”。某养殖场7天内连续出现2例同类疾病上报触发“聚集性疫情预警”。某样本检测阳性按病种和养殖场历史记录自动提高风险等级。这些规则的阈值没有写死在代码里而是放在数据库配置表中允许管理员在后台调整。这样畜牧站可以根据本辖区实际情况设置更敏感的预警线不用每次改阈值都发版。可视化部分我用的是EChartsDjango只提供JSON数据接口。主页面展示本辖区养殖场分布热力图、免疫完成率趋势、检测阳性率柱状图和待处置任务列表。预警消息通过WebSocket推送到后台页面工作人员一登录就能看到浏览器右上角的弹窗不用再每天翻微信群。3. 关键环节实操数据建模、接口权限与部署3.1 数据模型设计与循环引用问题在做这种多模块系统时很容易掉进循环外键的坑。比如疾病上报需要关联批次、批次的养殖场养殖场又需要记录最近的疾病上报情况。如果一路加外键模型之间会形成一个环迁移时会遇到各种顺序问题。我的建议是尽量让每个核心模型只持有必要的外键。比如Batch表里保存farm_id和current_disease_status字段但不保存最新上报单id避免Batch-DiseaseReport和DiseaseReport-Batch双向强引用。如果要从上报单反查批量信息通过ForeignKey查询完全可以胜任没必要额外存冗余字段。另外不要过度使用GenericForeignKey。基层系统里字段含义比较固定使用通用外键虽然灵活但查询时会丢掉很多数据库层的关联优化写起来也容易出bug。除非你是做一个类似工单系统的多实体评论功能否则尽量少用。上了PostgreSQL之后还有一个性能优化点对数据量增长很快的日志表比如操作日志、检测结果历史表可以使用表分区。Django原生不直接支持分区表但可以通过数据库中手动创建分区表然后让模型inspect读取或者使用第三方扩展包。这个在数据量到百万级之后帮助非常大建议提前规划。3.2 权限控制与操作日志畜牧站系统和其他业务系统不同用户角色很杂有畜牧站管理员、防疫员、实验室检测员、乡镇巡查员。每个角色能看到的数据范围不一样。乡村巡查员只能看到自己负责的养殖场实验室检测员只能看到送检样本相关数据畜牧站站长可以看全辖区所有内容。最简单有效的做法是用Django自带的auth框架结合一个UserProfile字段来限定数据范围。也就是说用户登录后我们不仅要判断”有没有某个权限“还要在查询集上强制过滤”能看到哪些养殖场“。def visible_batches(user): if user.role station_admin: return Batch.objects.all() if user.role inspector: return Batch.objects.filter(area__inuser.managed_areas.all()) return Batch.objects.none()这种过滤逻辑最好封装成QuerySet Manager方法使其在views和模板中复用。实际踩坑最多的不是权限查询而是权限测试不充分。我见过很多系统普通用户一改URL参数就能看到其他人的数据。这不是改前端能解决的必须在后端每个查询里加数据范围过滤最好写一套自动化测试把所有角色对应的关键接口都跑一遍。操作日志单独建了一张表用middleware统一记录。记录内容包括操作人、IP、请求路径、请求参数、返回状态码、耗时。这个日志不只是为了追溯还可以用来做接口性能分析。有一次系统反应慢我直接把最近一周耗时Top10的接口从日志里拉出来很快就找到了瓶颈。3.3 检测数据异步入库与结果推送在前面Flask接收检测数据时我提到会把详细判定丢给Celery异步处理。实际项目中我用的是Redis作为BrokerCelery worker单独跑一个进程。这一步主要是为了防止这样的场景某次检测设备批量上传几千个样本如果所有判定都在HTTP请求里同步做Flask服务会长时间占用worker导致其他请求排不上队。异步任务还有一个好处可以可靠重试。仪器上传数据时偶发网络超时或者样本编号格式不规范任务执行失败后会自动重试不会因为一次解析失败就丢失整批检测数据。结果推送给Django主服务时我并没有直接调用Django的内部接口而是通过Redis发布订阅消息先发一条事件通知Django这边收到后主动去Flask查询最新结果。这样两个服务不会在调用链上强耦合Flask也不需要知道Django的接口地址。当然如果你的部署环境比较简单直接HTTP调用也行但一定要加接口鉴权不能裸奔。最后整个异步链路要加一个兜底定时任务比如每隔30分钟扫描一下“检测结果已生成但送检单未更新状态”的样本自动补齐状态。技术上这个东西叫“对账任务”非常实用。否则发布订阅偶尔丢消息用户是不会管你是网络问题还是代码问题只会看到状态没更新然后来投诉。3.4 部署与配置要点Gunicorn Nginx部署这块很多人图省事直接用Python自带的runserver跑生产服务这是大坑。Django和Flask开发服务器都是单进程的并发能力极差稍微有点流量就直接卡死。我用Gunicorn起多个worker配置大致如下# Django服务 gunicorn config.wsgi:application \ --bind 0.0.0.0:8000 \ --workers 4 \ --threads 2 \ --timeout 60 \ --access-logfile - \ --error-logfile - # Flask检测服务 gunicorn app:app \ --bind 0.0.0.0:5000 \ --workers 2 \ --threads 4 \ --timeout 30 \ --access-logfile -Nginx配置里我会把/api/detect开头的请求转发到Flask服务其他业务请求转发到Django服务。静态文件和上传的文件直接由Nginx处理减轻Python进程负担。server { listen 80; server_name livestock.example.com; location /static/ { alias /var/www/django_server/static/; } location /media/ { alias /var/www/django_server/media/; } location /api/detect/ { proxy_pass http://127.0.0.1:5000; 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; } }线上环境还有一个容易忽略的问题时区。畜牧站检测结果有时间属性服务器默认时区往往是UTC直接保存会导致凌晨上报的时间变成前一天。我在Django的配置文件里把TIME_ZONE设为自己的业务时区同时把USE_TZ保留为True这样数据库存的是带时区的UTC时间展示时再转成业务时间避免不同模块时间不一致。4. 常见问题与踩坑实录4.1 跨域与前后端联调问题因为Django和Flask分属两个端口前端页面在调试时必然遇到跨域问题。我第一次上线时没注意Flask服务的CORS配置结果浏览器里检测结果回传一直报错。排查了半天才发现是响应头里缺少Access-Control-Allow-Origin。解决方式其实很简单Flask装一个Flask-CORS扩展按路由配置允许来源。如果前后端同域部署Nginx转发了就不存在跨域问题但如果开发环境是前端工程单独跑在localhost:8080后端在localhost:8000那必须在后端配置CORS。from flask_cors import CORS CORS(app, origins[http://localhost:8080], supports_credentialsTrue)这里有个细节supports_credentials选项只能和明确的origins列表一起使用不能写成“”。如果写“”浏览器在携带Cookie时会被拒绝。调试跨域问题最有效的工具是看浏览器Network里的完整响应头能直接看出Access-Control-Allow-Origin有没有正常返回。4.2 数据库并发冲突与事务基层系统虽然并发不高但在特定时间段会有集中操作。比如早上上班后多个防疫员同时给同一个批次的动物做免疫登记或者扫码设备连续推送检测结果。此时如果不加事务控制很容易出现重复插入、状态被覆盖之类的问题。最典型的一个坑是两个请求同时检测到上报单处于”待审核“状态然后分别执行了“通过”和“驳回”操作最后数据库里显示的状态是后提交的那次操作但业务上先提交的其实已经生效。解决这个问题不能只靠前端禁用按钮必须在后端使用select_for_update锁住对应记录让两个操作串行化。from django.db import transaction def handle_audit(report_id, action, user): with transaction.atomic(): report DiseaseReport.objects.select_for_update().get(idreport_id) if report.state ! pending: raise BusinessException(该上报单已处理禁止重复操作) if action approve: report.state processing elif action reject: report.state rejected report.save()同样的问题也会出现在送检单状态更新上。所以我把“修改核心业务状态”的代码统一封装成Service层方法不允许在views里直接save。这样所有状态变更都经过同一套并发控制后面扩展也方便。4.3 定时任务与预警误报预警任务看似简单其实是最容易出乱子的地方。第一个坑是重复执行。如果定时任务跑得慢后一个任务又启动了就可能产生重复预警。我在实现时给每条预警消息加了一个业务唯一键比如“补免提醒”的唯一键是batch_idvaccine_type提醒日期如果当天已经存在同样的提醒就不再重复插入。第二个坑是时区问题定时任务最好使用本地时区时间。如果任务定义的是“每天零点跑”但服务器时区是UTC实际执行时间是早上8点工作人员看到的预警永远是8点后产生的会错过早间巡查。这些细节都要在配置里对清楚。第三个坑是规则误报。刚开始我把阈值设得很灵敏比如“连续两天出现同类上报”就预警结果一个养殖户只是为了报备连续两天上报了两只鸡的轻微症状系统就拉响了一次一级预警搞得大家虚惊一场。后来我把规则加了约束条件必须是同批次、临床表现符合某种特征才触发聚集性预警。业务规则一定要跟畜牧站的人反复确认不能只凭技术逻辑拍脑袋。4.4 数据量增长后的归档策略基层系统的数据量增长其实比想象中快。每一条免疫记录、每一次检测结果、每一个操作日志一年下来也是几十万条。刚开始没做归档到了年底汇总报表时一些联表查询从毫秒级变成了秒级页面直接超时。我的做法是设计一个year字段在业务表中按年度分表归档。甲方不需要查询特别老的历史数据时默认只查询当年和上一年的数据。查询接口里自动带上年份范围历史数据通过后台单独页面导出或查询。这样既不影响日常使用又不用动不动就上分库分表。另外监控数据库慢查询日志也很有必要。PostgreSQL默认会记录执行时间超过阈值的SQL我会定期看这些慢查询然后针对性加索引或改写查询逻辑。这套闭环做下来系统跑几年都不会有明显卡顿。说句实在话Django和Flask的组合并不是什么高深技术但它非常适合像畜牧站疾病防控与检测系统这种“核心业务稳、边缘模块活”的项目。Django帮你把用户、权限、迁移、后台这些繁琐事情管好Flask给你留出一个轻量、干净的区域去应对不断变化的检测需求。如果你也在做类似的政务信息化项目不妨先把业务状态机理清楚再决定模块边界架构自然就浮现出来了。最后再分享一个小技巧两个服务之间传递业务状态时别只传一个状态码要把变化前后的状态都记录到日志里否则排查流程纠纷时你根本说不清问题是哪一步出的。