1. 项目背景与需求分析
城市花园小区维修管理系统是一个典型的物业信息化解决方案，旨在通过数字化手段提升小区设备维护、报修响应和资源调度的效率。传统物业维修管理普遍存在工单流转慢、维修进度不透明、历史记录难追溯等问题。我们设计的系统需要实现以下核心功能：
- 业主端：在线报修（文字/图片/视频）、进度查询、服务评价
- 物业端：工单智能分派、维修资源管理、数据统计分析
- 维修端：移动接单、维修记录上传、配件申领
- 管理端：KPI考核、费用结算、供应商管理
2. 技术选型对比
2.1 Django vs Flask框架对比
通过实际项目验证，两个框架在维修管理系统中的表现差异明显：
| 特性 | Django优势 | Flask优势 |
|---|---|---|
| 开发速度 | 内置Admin后台可节省30%开发时间 | 需要自行搭建后台管理系统 |
| 数据建模 | ORM支持完善的关系型数据建模 | 需搭配SQLAlchemy等第三方库 |
| API开发 | 需配合DRF(Django REST Framework) | 原生支持RESTful API开发更轻量 |
| 微服务适配 | 单体架构改造成本高 | 天然适合微服务架构 |
| 学习曲线 | 需要掌握MTV全套理念 | 仅需理解路由和视图函数 |
2.2 混合架构设计思路
基于项目实际需求，我们采用Django+Flask混合架构：
- 使用Django构建核心业务模块（用户中心、工单管理、权限系统）
- 采用Flask开发AI服务接口（智能分单、图像识别）
- 通过RabbitMQ实现跨框架消息通信
3. 核心模块实现
3.1 智能工单分派系统
# Django模型设计 class RepairOrder(models.Model): PRIORITY_CHOICES = [ (0, '常规'), (1, '紧急'), (2, '特急') ] order_no = models.CharField(max_length=20, unique=True) location = models.ForeignKey(Location, on_delete=models.PROTECT) reporter = models.ForeignKey(User, on_delete=models.PROTECT) category = models.ForeignKey(RepairCategory, on_delete=models.PROTECT) priority = models.SmallIntegerField(choices=PRIORITY_CHOICES) ai_score = models.FloatField(null=True) # AI评估分值 # Flask实现的AI分单服务 @app.route('/api/ai/assign', methods=['POST']) def ai_assign(): data = request.get_json() # 使用预训练的XGBoost模型 model = joblib.load('ai_model/xgboost_v1.pkl') features = [ data['category'], data['priority'], datetime.now().hour, # 报修时间 len(data['description']), # 描述长度 1 if data['has_image'] else 0 ] score = model.predict([features])[0] return jsonify({'score': float(score)})3.2 移动端图片识别报修
# Django接收图片的视图 class ImageUploadView(APIView): parser_classes = [MultiPartParser] def post(self, request): image_file = request.FILES['image'] img_path = f"media/temp/{uuid.uuid4()}.jpg" # 调用Flask图像识别服务 response = requests.post( 'http://ai-service:5000/api/ai/recognize', files={'image': image_file} ) result = response.json() # 自动填充报修类别 category = RepairCategory.objects.get( name=result['predicted_category']) return Response({ 'category_id': category.id, 'confidence': result['confidence'] }) # Flask图像识别服务 @app.route('/api/ai/recognize', methods=['POST']) def recognize(): if 'image' not in request.files: abort(400) # 使用预训练的ResNet50模型 model = tf.keras.models.load_model('ai_model/resnet50.h5') img = Image.open(request.files['image']) img = img.resize((224, 224)) img_array = np.array(img) / 255.0 img_array = np.expand_dims(img_array, axis=0) predictions = model.predict(img_array) class_idx = np.argmax(predictions[0]) return jsonify({ 'predicted_category': CLASS_NAMES[class_idx], 'confidence': float(predictions[0][class_idx]) })4. 系统集成关键点
4.1 跨框架会话保持方案
# settings.py 共享配置 SECRET_KEY = 'django-insecure-+your-secret-key-here' SESSION_COOKIE_NAME = 'repair_system_session' SESSION_COOKIE_DOMAIN = '.example.com' # Flask配置同步 app.secret_key = SECRET_KEY app.config['SESSION_COOKIE_NAME'] = SESSION_COOKIE_NAME app.config['SESSION_COOKIE_DOMAIN'] = SESSION_COOKIE_DOMAIN4.2 统一认证中间件
# Django认证中间件 class CrossAuthMiddleware: def __init__(self, get_response): self.get_response = get_response def __call__(self, request): if request.path.startswith('/api/flask/'): session_key = request.COOKIES.get(SESSION_COOKIE_NAME) if session_key: engine = importlib.import_module('django.contrib.sessions.backends.db') session = engine.SessionStore(session_key) if 'auth_user_id' in session: user = User.objects.get(pk=session['auth_user_id']) request.user = user return self.get_response(request)5. 性能优化实践
5.1 数据库查询优化
# 错误示例（N+1查询问题） orders = RepairOrder.objects.all() for order in orders: print(order.reporter.username) # 每次循环都查询数据库 # 优化方案 orders = RepairOrder.objects.select_related( 'reporter', 'location', 'category' ).prefetch_related( 'attachments' ).filter(status='pending')5.2 缓存策略设计
# Django缓存配置 CACHES = { 'default': { 'BACKEND': 'django_redis.cache.RedisCache', 'LOCATION': 'redis://redis:6379/1', 'OPTIONS': { 'CLIENT_CLASS': 'django_redis.client.DefaultClient', } } } # Flask缓存扩展 cache = RedisCache( host='redis', port=6379, db=2, default_timeout=300 ) @app.route('/api/stats') @cache.cached(timeout=60) def get_stats(): # 复杂统计计算 return jsonify(statistics)6. 部署架构设计
前端Nginx ├── 静态文件服务 ├── 负载均衡 │ ├── Django容器组（Gunicorn） │ └── Flask容器组（Gunicorn） ├── Redis缓存 └── PostgreSQL集群关键部署命令示例：
# Django迁移 docker-compose exec django python manage.py migrate # Flask服务启动 gunicorn -w 4 -b :5000 --access-logfile - "app:create_app()" # Celery定时任务 celery -A repair_system beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler7. 踩坑经验总结
跨域会话问题：
- 现象：Django设置的session在Flask端无法读取
- 解决方案：确保SECRET_KEY一致，配置SESSION_COOKIE_DOMAIN为父域名
静态文件冲突：
- 现象：两个框架的static目录命名冲突
- 解决方案：Django使用/static/，Flask改用/flask_static/
AI模型热更新：
- 现象：模型更新需要重启服务
- 解决方案：使用watchdog监控模型文件变化，自动重新加载
数据库连接泄漏：
- 现象：Flask端出现大量空闲连接
- 解决方案：配置SQLAlchemy的pool_recycle参数
日志统一收集：
- 使用ELK栈集中处理日志
- 关键配置示例：
LOGGING = { 'version': 1, 'handlers': { 'logstash': { 'class': 'logstash.TCPLogstashHandler', 'host': 'logstash', 'port': 5959, 'version': 1, } } }
这套系统在实际部署后，将平均报修响应时间从48小时缩短至4小时，业主满意度提升35%。特别在暴雨等突发事件期间，智能分单系统能自动识别紧急情况，优先处理电梯故障、电力中断等关键问题。
