--- title: webhook_views.py 代码改进计划 source: author: shenwei published: created: description: tags: [] --- # webhook_views.py 代码改进计划 > **For Claude:** REQUIRED SUB-SKILL: Use `executing-plans` skill to implement this plan task-by-task. **Goal:** 全面改进 webhook_views.py 的代码质量，修复严重bug，优化性能，增强安全性 **Architecture:** - 创建专门的 validator 层用于输入验证 - 创建 service 层封装业务逻辑和数据操作 - 添加日志脱敏工具函数 - 重构现有 View 为更小、更专注的组件 - 统一使用 DRF Serializer 进行数据验证 **Tech Stack:** Django REST Framework, drf-yasg, json, logging --- ## 📋 问题分析摘要 ### 1. 🔴 静默失败 (CRITICAL BUG) **位置:** 第 146-154 行 **问题:** 数据库操作失败却返回 HTTP 200 OK ```python except Exception as db_error: logger.error(f'数据库操作失败: {db_error}', exc_info=True) return JsonResponse({ 'success': True, # ❌ BUG: 应该是 False 'message': '...', }, status=200) # ❌ BUG: 应该是 500 ``` **影响:** 调用方无法得知数据未保存，导致数据不一致 ### 2. 🟡 长函数 **位置:** - `ItineraryWebhookView.post()` - 130+ 行 - `RequirementWebhookView.post()` - 200+ 行 - `create_daily_schedules()` - 75+ 行 **问题:** 单一方法承担了太多职责，难以测试和维护 ### 3. 🟡 N+1 查询 **位置:** `create_daily_schedules()` 方法 ```python # 循环内查询目的地 for day_schedule in daily_schedules: destination = Destination.objects.filter(...).first() # N+1 for activity in activities: # 每个活动都单独查询 if activity_type == ATTRACTION: attraction = Attraction.objects.get(...) # N+1 elif activity_type == MEAL: restaurant = Restaurant.objects.get(...) # N+1 elif activity_type == CHECK_IN/CHECK_OUT: hotel = Hotel.objects.get(...) # N+1 ``` **影响:** 假设有 10 天行程，每天 5 个活动 = 最多 51 次数据库查询 ### 4. 🟠 输入验证缺失 **问题:** - 只检查字段存在，不验证类型 - 无数据大小限制 - 无格式验证（如日期格式、ID格式） - 无恶意数据检测 ### 5. 🟠 日志敏感信息泄露 **位置:** - 第 340 行: `json.dumps(data, ensure_ascii=False)[:500]` - 第 585 行: `user_input[:100]` - 可能包含 PII - 多处记录完整请求数据 --- ## 📝 实施计划 ### 阶段 1: 日志脱敏工具 (安全优先) #### Task 1.1: 创建日志脱敏工具 **Files:** - Create: `apps/api/utils/logging_utils.py` - Modify: `apps/api/views/webhook_views.py` (导入使用) **Step 1: 创建日志脱敏模块** ```python # apps/api/utils/logging_utils.py """日志脱敏工具模块""" import re import logging from typing import Any, Dict, List, Optional from functools import wraps # 敏感字段名模式 SENSITIVE_FIELD_PATTERNS = [ 'password', 'secret', 'token', 'api_key', 'apikey', 'phone', 'mobile', 'tel', 'email', 'address', 'id_card', 'idcard', 'passport', 'credit_card', 'card_number', 'bank_account', 'ssn' ] # 手机号正则 (中国) PHONE_PATTERN = re.compile(r'1[3-9]\d{9}') # 邮箱正则 EMAIL_PATTERN = re.compile(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}') # 身份证正则 ID_CARD_PATTERN = re.compile(r'\d{17}[\dXx]') class LogSanitizer: """日志脱敏器""" @staticmethod def mask_phone(text: str) -> str: """脱敏手机号: 13812345678 -> 138****5678""" def replacer(match): phone = match.group() return phone[:3] + '****' + phone[-4:] return PHONE_PATTERN.sub(replacer, text) @staticmethod def mask_email(text: str) -> str: """脱敏邮箱: user@example.com -> u***@example.com""" def replacer(match): email = match.group() local = email.split('@')[0] domain = email.split('@')[1] if '@' in email else '' return f"{local[0]}***@{domain}" return EMAIL_PATTERN.sub(replacer, text) @staticmethod def mask_id_card(text: str) -> str: """脱敏身份证: 110101199001011234 -> 110101********1234""" def replacer(match): id_card = match.group() return id_card[:6] + '********' + id_card[-4:] return ID_CARD_PATTERN.sub(replacer, text) @classmethod def is_sensitive_field(cls, field_name: str) -> bool: """判断字段是否为敏感字段""" field_lower = field_name.lower() for pattern in SENSITIVE_FIELD_PATTERNS: if pattern in field_lower: return True return False @classmethod def sanitize_dict(cls, data: Any, depth: int = 0, max_depth: int = 10) -> Any: """递归脱敏字典中的敏感信息""" if depth > max_depth: return '[MAX_DEPTH_EXCEEDED]' if isinstance(data, dict): result = {} for key, value in data.items(): if cls.is_sensitive_field(key): if isinstance(value, str) and len(value) > 4: result[key] = value[:2] + '***' + value[-2:] else: result[key] = '***' else: result[key] = cls.sanitize_dict(value, depth + 1, max_depth) return result elif isinstance(data, list): return [cls.sanitize_dict(item, depth + 1, max_depth) for item in data] elif isinstance(data, str): text = cls.mask_phone(data) text = cls.mask_email(text) text = cls.mask_id_card(text) # 截断过长字符串 if len(text) > 200: return text[:200] + '...[TRUNCATED]' return text return data @classmethod def sanitize_for_logging(cls, data: Any, include_fields: Optional[List[str]] = None) -> str: """ 将数据脱敏后转为日志字符串 Args: data: 要脱敏的数据 include_fields: 如果指定，只记录这些字段（用于白名单模式） """ import json if include_fields: # 白名单模式 if isinstance(data, dict): filtered = {k: v for k, v in data.items() if k in include_fields} data = filtered else: data = {} sanitized = cls.sanitize_dict(data) try: return json.dumps(sanitized, ensure_ascii=False, default=str) except Exception: return str(sanitized) def log_webhook_call(logger: logging.Logger, data: Dict, message: str, include_fields: Optional[List[str]] = None, max_length: int = 500): """ 记录 webhook 调用日志（自动脱敏） Args: logger: 日志记录器 data: 请求数据 message: 日志消息 include_fields: 只记录的字段列表（可选） max_length: 最大日志长度 """ sanitized = LogSanitizer.sanitize_for_logging(data, include_fields) if len(sanitized) > max_length: sanitized = sanitized[:max_length] + '...[TRUNCATED]' logger.info(f"{message}: {sanitized}") ``` **Step 2: 验证模块存在性** Run: `ls -la apps/api/utils/` Expected: 目录存在或自动创建 --- ### 阶段 2: 创建 Serializer 层 (输入验证) #### Task 2.1: 创建 ItineraryWebhookSerializer **Files:** - Create: `apps/api/serializers/webhook_serializers.py` **Step 1: 创建 Webhook 序列化器** ```python # apps/api/serializers/webhook_serializers.py """Webhook API 专用序列化器""" from rest_framework import serializers from apps.models.requirement import Requirement from apps.models.itinerary import Itinerary from apps.models.destinations import Destination from apps.models.traveler_stats import TravelerStats from apps.models.daily_schedule import DailySchedule from apps.models.attraction import Attraction from apps.models.hotel import Hotel from apps.models.restaurant import Restaurant class ActivitySerializer(serializers.Serializer): """单个活动序列化器""" activity_title = serializers.CharField(max_length=200) activity_type = serializers.ChoiceField(choices=[ 'FLIGHT', 'TRAIN', 'ATTRACTION', 'MEAL', 'TRANSPORT', 'SHOPPING', 'FREE', 'CHECK_IN', 'CHECK_OUT', 'OTHER' ]) start_time = serializers.TimeField(format='%H:%M:%S') end_time = serializers.TimeField(format='%H:%M:%S') activity_description = serializers.CharField(required=False, allow_blank=True) id_reference = serializers.CharField(required=False, allow_blank=True) class DayScheduleSerializer(serializers.Serializer): """每日行程序列化器""" day = serializers.IntegerField(min_value=1) date = serializers.DateField() city = serializers.CharField(max_length=100) activities = ActivitySerializer(many=True) class DestinationSerializer(serializers.Serializer): """目的地序列化器""" city_name = serializers.CharField(max_length=100) country_code = serializers.CharField(max_length=3, required=False, default='CN') destination_order = serializers.IntegerField(min_value=1) arrival_date = serializers.DateField() departure_date = serializers.DateField() class TravelerStatsInputSerializer(serializers.Serializer): """旅行者统计输入序列化器""" adults = serializers.IntegerField(min_value=0, default=0) children = serializers.IntegerField(min_value=0, default=0) infants = serializers.IntegerField(min_value=0, default=0) seniors = serializers.IntegerField(min_value=0, default=0) class ItineraryWebhookInputSerializer(serializers.Serializer): """Itinerary Webhook 输入序列化器""" requirement_id = serializers.CharField(max_length=100) itinerary_name = serializers.CharField(max_length=200) start_date = serializers.DateField() end_date = serializers.DateField() # 可选字段 destinations = DestinationSerializer(many=True, required=False) traveler_stats = TravelerStatsInputSerializer(required=False) daily_schedules = DayScheduleSerializer(many=True, required=False) def validate(self, data): """交叉验证""" if data.get('start_date') and data.get('end_date'): if data['end_date'] < data['start_date']: raise serializers.ValidationError({ 'end_date': '结束日期不能早于开始日期' }) return data class RequirementWebhookInputSerializer(serializers.Serializer): """Requirement Webhook 输入序列化器""" user_input = serializers.CharField(max_length=5000, required=False) structured_data = serializers.DictField(required=False) requirement_id = serializers.CharField(max_length=100, required=False) llm_info = serializers.DictField(required=False) output = serializers.DictField(required=False) def validate(self, data): """验证至少有一个有效数据源""" if not any([ data.get('user_input'), data.get('structured_data'), data.get('requirement_id') ]): raise serializers.ValidationError( '至少需要提供 user_input, structured_data 或 requirement_id 之一' ) return data ``` --- ### 阶段 3: 修复静默失败 Bug (关键修复) #### Task 3.1: 修复数据库失败返回 200 的问题 **Files:** - Modify: `apps/api/views/webhook_views.py:146-154` **Step 1: 修改异常处理逻辑** 在 `ItineraryWebhookView.post()` 方法中，将: ```python except Exception as db_error: logger.error(f'数据库操作失败: {db_error}', exc_info=True) # 即使数据库操作失败，也要返回成功的响应，因为JSON解析和验证已经成功 return JsonResponse({ 'success': True, 'message': 'JSON解析和验证成功，但数据库操作失败。这可能是因为数据库服务不可用。', 'itinerary_name': data.get('itinerary_name'), 'requirement_id': data.get('requirement_id') }, status=200) ``` 替换为: ```python except Exception as db_error: logger.error(f'数据库操作失败: {db_error}', exc_info=True) # 明确返回错误状态，让调用方知道数据未保存 return JsonResponse({ 'success': False, 'error': '数据库操作失败，数据未保存', 'error_code': 'DB_OPERATION_FAILED', 'itinerary_name': data.get('itinerary_name') if data else None, 'requirement_id': data.get('requirement_id') if data else None }, status=500) ``` **Step 2: 添加错误响应结构化** 在文件顶部添加错误码常量: ```python # 错误码定义 class WebhookErrorCode: INVALID_JSON = 'INVALID_JSON' MISSING_REQUIRED_FIELD = 'MISSING_REQUIRED_FIELD' VALIDATION_FAILED = 'VALIDATION_FAILED' REQUIREMENT_NOT_FOUND = 'REQUIREMENT_NOT_FOUND' DB_OPERATION_FAILED = 'DB_OPERATION_FAILED' N8N_WEBHOOK_ERROR = 'N8N_WEBHOOK_ERROR' N8N_TIMEOUT = 'N8N_TIMEOUT' INTERNAL_ERROR = 'INTERNAL_ERROR' ``` --- ### 阶段 4: 重构长函数 (代码结构) #### Task 4.1: 提取数据处理服务 **Files:** - Create: `apps/api/services/webhook_services.py` **Step 1: 创建 Webhook 服务层** ```python # apps/api/services/webhook_services.py """Webhook 业务逻辑服务层""" from typing import Dict, List, Optional, Tuple from decimal import Decimal from datetime import datetime, date, time import logging from django.db import transaction from django.db.models import Q from apps.models.requirement import Requirement from apps.models.itinerary import Itinerary from apps.models.destinations import Destination from apps.models.traveler_stats import TravelerStats from apps.models.daily_schedule import DailySchedule from apps.models.attraction import Attraction from apps.models.hotel import Hotel from apps.models.restaurant import Restaurant logger = logging.getLogger(__name__) class ItineraryWebhookService: """行程 Webhook 服务""" def __init__(self, validated_data: Dict): self.data = validated_data self.itinerary: Optional[Itinerary] = None self.destinations_cache: Dict[str, Destination] = {} self.attractions_cache: Dict[str, Attraction] = {} self.hotels_cache: Dict[str, Hotel] = {} self.restaurants_cache: Dict[str, Restaurant] = {} def process(self, requirement: Requirement) -> Tuple[bool, str, Optional[Itinerary]]: """ 处理行程数据 Returns: Tuple[成功标志, 消息, 行程对象] """ try: with transaction.atomic(): # 1. 预加载关联数据 (解决 N+1 问题) self._preload_related_data() # 2. 创建行程主表 self.itinerary = self._create_itinerary(requirement) # 3. 创建关联关系 self._create_requirement_itinerary(requirement) # 4. 创建目的地 self._create_destinations() # 5. 创建旅行者统计 self._create_traveler_stats() # 6. 创建每日行程 self._create_daily_schedules() return True, '行程数据处理成功', self.itinerary except Exception as e: logger.error(f'处理行程数据失败: {e}', exc_info=True) return False, str(e), None def _preload_related_data(self): """预加载所有关联数据以避免 N+1 查询""" requirement_id = self.data.get('requirement_id') # 预加载目的地 destinations = self.data.get('destinations', []) destination_cities = [d.get('city_name') for d in destinations] if destination_cities: self.destinations_cache = { d.city_name: d for d in Destination.objects.filter( itinerary=self.itinerary, city_name__in=destination_cities ) } # 预加载所有可能用到的景点 attraction_ids = [] hotel_ids = [] restaurant_ids = [] for day_schedule in self.data.get('daily_schedules', []): for activity in day_schedule.get('activities', []): ref_id = activity.get('id_reference') if activity.get('activity_type') == 'ATTRACTION': attraction_ids.append(ref_id) elif activity.get('activity_type') in ['CHECK_IN', 'CHECK_OUT']: hotel_ids.append(ref_id) elif activity.get('activity_type') == 'MEAL': restaurant_ids.append(ref_id) if attraction_ids: self.attractions_cache = { a.attraction_id: a for a in Attraction.objects.filter( attraction_id__in=attraction_ids ) } if hotel_ids: self.hotels_cache = { h.hotel_id: h for h in Hotel.objects.filter( hotel_id__in=hotel_ids ) } if restaurant_ids: self.restaurants_cache = { r.restaurant_id: r for r in Restaurant.objects.filter( restaurant_id__in=restaurant_ids ) } def _create_itinerary(self, requirement: Requirement) -> Itinerary: """创建行程主表""" itinerary = Itinerary( itinerary_name=self.data.get('itinerary_name', '未命名行程'), start_date=self.data.get('start_date'), end_date=self.data.get('end_date'), travel_purpose=Itinerary.TravelPurpose.LEISURE, contact_person=requirement.contact_person or '测试用户', contact_phone=requirement.contact_phone or '13800138000', contact_company=requirement.contact_company, departure_city=requirement.origin_name or '上海', return_city=requirement.origin_name or '上海', current_status=Itinerary.CurrentStatus.DRAFT, created_by='webhook_user' ) itinerary.save() return itinerary def _create_requirement_itinerary(self, requirement: Requirement): """创建需求与行程关联""" from apps.models.requirement_itinerary import RequirementItinerary requirement_itinerary = RequirementItinerary( requirement=requirement, itinerary=self.itinerary ) requirement_itinerary.save() logger.info( f'创建需求与行程关联成功: ' f'requirement_id={requirement.requirement_id}, ' f'itinerary_id={self.itinerary.itinerary_id}' ) def _create_destinations(self): """批量创建目的地""" destinations = self.data.get('destinations', []) for dest_data in destinations: destination = Destination( itinerary=self.itinerary, destination_order=dest_data.get('destination_order'), city_name=dest_data.get('city_name'), country_code=dest_data.get('country_code', 'CN'), arrival_date=dest_data.get('arrival_date'), departure_date=dest_data.get('departure_date') ) destination.save() self.destinations_cache[dest_data.get('city_name')] = destination def _create_traveler_stats(self): """创建旅行者统计""" traveler_stats_info = self.data.get('traveler_stats', {}) traveler_stats = TravelerStats( itinerary=self.itinerary, adult_count=traveler_stats_info.get('adults', 0), child_count=traveler_stats_info.get('children', 0), infant_count=traveler_stats_info.get('infants', 0), senior_count=traveler_stats_info.get('seniors', 0) ) traveler_stats.save() def _create_daily_schedules(self): """批量创建每日行程 (使用预加载的数据)""" daily_schedules = self.data.get('daily_schedules', []) for day_schedule in daily_schedules: day_number = day_schedule.get('day') schedule_date = day_schedule.get('date') city = day_schedule.get('city') # 使用缓存的目的地 destination = self.destinations_cache.get(city) # 批量处理活动 activities = day_schedule.get('activities', []) for activity in activities: self._create_activity_schedule( day_number, schedule_date, city, destination, activity ) def _create_activity_schedule( self, day_number: int, schedule_date: date, city: str, destination: Optional[Destination], activity: Dict ): """创建单个活动记录""" activity_type = self._map_activity_type(activity.get('activity_type')) # 使用缓存的关联对象 attraction = self.hotel = self.restaurant = None if activity_type == DailySchedule.ActivityType.ATTRACTION: attraction = self.attractions_cache.get(activity.get('id_reference')) elif activity_type in [DailySchedule.ActivityType.CHECK_IN, DailySchedule.ActivityType.CHECK_OUT]: self.hotel = self.hotels_cache.get(activity.get('id_reference')) elif activity_type == DailySchedule.ActivityType.MEAL: self.restaurant = self.restaurants_cache.get(activity.get('id_reference')) schedule = DailySchedule( itinerary_id=self.itinerary, day_number=day_number, schedule_date=schedule_date, destination_id=destination, activity_type=activity_type, activity_title=activity.get('activity_title'), activity_description=activity.get('activity_description', ''), start_time=activity.get('start_time'), end_time=activity.get('end_time'), attraction_id=attraction, hotel_id=self.hotel, restaurant_id=self.restaurant, booking_status=DailySchedule.BookingStatus.NOT_BOOKED ) schedule.save() @staticmethod def _map_activity_type(activity_type_str: str) -> str: """映射活动类型""" mapping = { 'FLIGHT': DailySchedule.ActivityType.FLIGHT, 'TRAIN': DailySchedule.ActivityType.TRAIN, 'ATTRACTION': DailySchedule.ActivityType.ATTRACTION, 'MEAL': DailySchedule.ActivityType.MEAL, 'TRANSPORT': DailySchedule.ActivityType.TRANSPORT, 'SHOPPING': DailySchedule.ActivityType.SHOPPING, 'FREE': DailySchedule.ActivityType.FREE, 'CHECK_IN': DailySchedule.ActivityType.CHECK_IN, 'CHECK_OUT': DailySchedule.ActivityType.CHECK_OUT, 'OTHER': DailySchedule.ActivityType.OTHER } return mapping.get(activity_type_str, DailySchedule.ActivityType.OTHER) class RequirementWebhookService: """需求 Webhook 服务""" def __init__(self, validated_data: Dict): self.data = validated_data self.requirement_id: Optional[str] = None def process(self) -> Tuple[bool, str, Optional[str], Optional[Requirement]]: """ 处理需求数据 Returns: Tuple[成功标志, 消息, 需求ID, 需求对象] """ try: # 提取数据 requirement_data = self._extract_structured_data() # 生成需求ID self.requirement_id = self._generate_requirement_id() # 构建需求对象 requirement = self._build_requirement(requirement_data) # 保存 requirement.save() logger.info(f'需求保存成功，ID: {self.requirement_id}') return True, '需求处理成功', self.requirement_id, requirement except Exception as e: logger.error(f'处理需求失败: {e}', exc_info=True) return False, str(e), None, None def _extract_structured_data(self) -> Dict: """提取结构化数据""" # 优先使用 structured_data，否则使用原始数据 if 'structured_data' in self.data: return self.data['structured_data'] # 移除顶层元数据字段 result = {k: v for k, v in self.data.items() if k not in ['user_input', 'requirement_id', 'llm_info', 'output']} return result def _generate_requirement_id(self) -> str: """生成唯一的 requirement_id""" import time max_attempts = 100 for attempt in range(max_attempts): today = datetime.now().strftime('%Y%m%d') prefix = f'REQ_{today}_' # 获取当前最大序号 max_seq = Requirement.objects.filter( requirement_id__startswith=prefix ).aggregate( max_seq=Max( Cast( Substr(F('requirement_id'), len(prefix) + 1), output_field=IntegerField() ) ) )['max_seq'] or 0 new_id = f'{prefix}{max_seq + 1:03d}' if not Requirement.objects.filter(requirement_id=new_id).exists(): return new_id time.sleep(0.01) raise ValueError('无法生成唯一的 requirement_id') def _build_requirement(self, data: Dict) -> Requirement: """构建 Requirement 对象""" base_info = data.get('base_info', {}) preferences = data.get('preferences', {}) budget = data.get('budget', {}) metadata = data.get('metadata', {}) # 解析日期 travel_start = self._parse_date( base_info.get('travel_date', {}).get('start_date') ) travel_end = self._parse_date( base_info.get('travel_date', {}).get('end_date') ) # 解析预算 budget_range = budget.get('range', {}) budget_min = self._parse_decimal(budget_range.get('min')) budget_max = self._parse_decimal(budget_range.get('max')) return Requirement( requirement_id=self.requirement_id, origin_name=base_info.get('origin', {}).get('name', ''), origin_code=base_info.get('origin', {}).get('code', ''), origin_type=base_info.get('origin', {}).get('type', ''), destination_cities=base_info.get('destination_cities', []), trip_days=base_info.get('trip_days', 1), group_adults=base_info.get('group_size', {}).get('adults', 0), group_children=base_info.get('group_size', {}).get('children', 0), group_seniors=base_info.get('group_size', {}).get('seniors', 0), group_total=base_info.get('group_size', {}).get('total', 1), travel_start_date=travel_start, travel_end_date=travel_end, travel_date_flexible=base_info.get('travel_date', {}).get('is_flexible', False), transportation_type=preferences.get('transportation', {}).get('type', ''), transportation_notes=preferences.get('transportation', {}).get('notes', ''), hotel_level=preferences.get('accommodation', {}).get('level', ''), hotel_requirements=preferences.get('accommodation', {}).get('requirements', ''), trip_rhythm=preferences.get('itinerary', {}).get('rhythm', ''), preference_tags=preferences.get('itinerary', {}).get('tags', []), must_visit_spots=preferences.get('itinerary', {}).get('special_constraints', {}).get('must_visit_spots', []), avoid_activities=preferences.get('itinerary', {}).get('special_constraints', {}).get('avoid_activities', []), budget_level=budget.get('level', ''), budget_currency=budget.get('currency', 'CNY'), budget_min=budget_min, budget_max=budget_max, budget_notes=budget.get('budget_notes', ''), source_type=metadata.get('source_type', 'NaturalLanguage'), status=metadata.get('status', 'Confirmed'), assumptions=metadata.get('assumptions', []), is_template=metadata.get('is_template', False), template_name=metadata.get('template_info', {}).get('name', ''), template_category=metadata.get('template_info', {}).get('category', ''), extension=data.get('extension', {}) ) def _parse_date(self, value) -> Optional[date]: """解析日期""" if not value: return None if isinstance(value, date): return value try: return datetime.fromisoformat(str(value)).date() except (ValueError, TypeError): logger.warning(f'无法解析日期: {value}') return None def _parse_decimal(self, value) -> Optional[Decimal]: """解析 Decimal""" if not value: return None try: return Decimal(str(value)) except (ValueError, TypeError): logger.warning(f'无法解析金额: {value}') return None ``` > **注意:** 需要在 `_generate_requirement_id` 中添加必要的导入: > ```python > from django.db.models import Max, F > from django.db.models.functions import Substr > from django.db.models import IntegerField > ``` --- ### 阶段 5: 重构 View 层 (使用新服务) #### Task 5.1: 重构 ItineraryWebhookView **Files:** - Modify: `apps/api/views/webhook_views.py` **Step 1: 重构 ItineraryWebhookView** ```python @method_decorator(csrf_exempt, name='dispatch') class ItineraryWebhookView(View): """处理n8n webhook返回的行程数据""" def post(self, request, *args, **kwargs): """接收并处理webhook数据""" from apps.api.utils.logging_utils import log_webhook_call from apps.api.serializers.webhook_serializers import ItineraryWebhookInputSerializer try: # 1. 解析请求体 content = request.body.decode('utf-8') raw_data = json.loads(content) # 2. 日志记录（脱敏后） log_webhook_call( logger, raw_data, '接收到行程webhook请求', include_fields=['requirement_id', 'itinerary_name', 'start_date', 'end_date'] ) # 3. 处理 output 字段 data = raw_data.get('output', raw_data) # 4. 输入验证 serializer = ItineraryWebhookInputSerializer(data=data) if not serializer.is_valid(): logger.warning(f'输入验证失败: {serializer.errors}') return JsonResponse({ 'success': False, 'error': '输入验证失败', 'error_code': WebhookErrorCode.VALIDATION_FAILED, 'details': serializer.errors }, status=400) validated_data = serializer.validated_data # 5. 验证关联需求存在 requirement_id = validated_data.get('requirement_id') try: requirement = Requirement.objects.get(requirement_id=requirement_id) except Requirement.DoesNotExist: logger.error(f'需求不存在: {requirement_id}') return JsonResponse({ 'success': False, 'error': f'需求不存在: {requirement_id}', 'error_code': WebhookErrorCode.REQUIREMENT_NOT_FOUND }, status=404) # 6. 调用服务处理 from apps.api.services.webhook_services import ItineraryWebhookService service = ItineraryWebhookService(validated_data) success, message, itinerary = service.process(requirement) if success: logger.info(f'行程创建成功: itinerary_id={itinerary.itinerary_id}') return JsonResponse({ 'success': True, 'itinerary_id': itinerary.itinerary_id, 'itinerary_name': itinerary.itinerary_name }, status=200) else: logger.error(f'行程创建失败: {message}') return JsonResponse({ 'success': False, 'error': message, 'error_code': WebhookErrorCode.DB_OPERATION_FAILED }, status=500) except json.JSONDecodeError as e: logger.error(f'JSON格式错误: {e}') return JsonResponse({ 'success': False, 'error': f'JSON格式错误: {str(e)}', 'error_code': WebhookErrorCode.INVALID_JSON }, status=400) except Exception as e: logger.error(f'处理行程数据失败: {e}', exc_info=True) return JsonResponse({ 'success': False, 'error': f'处理数据失败: {str(e)}', 'error_code': WebhookErrorCode.INTERNAL_ERROR }, status=500) ``` #### Task 5.2: 重构 RequirementWebhookView **Files:** - Modify: `apps/api/views/webhook_views.py` **Step 1: 简化 RequirementWebhookView.post 方法** ```python @method_decorator(csrf_exempt, name='dispatch') class RequirementWebhookView(APIView): """处理n8n webhook返回的旅游需求数据""" permission_classes = [AllowAny] @swagger_auto_schema(...) def post(self, request, *args, **kwargs): from apps.api.utils.logging_utils import log_webhook_call from apps.api.serializers.webhook_serializers import RequirementWebhookInputSerializer from apps.api.services.webhook_services import RequirementWebhookService try: raw_data = request.data # 处理列表结构 if isinstance(raw_data, list) and len(raw_data) > 0: raw_data = raw_data[0] # 处理 output 字段 if 'output' in raw_data: raw_data = raw_data['output'] # 日志记录（脱敏） log_webhook_call( logger, raw_data, '接收到需求webhook请求', include_fields=['requirement_id', 'source_type', 'status'] ) # 输入验证 serializer = RequirementWebhookInputSerializer(data=raw_data) if not serializer.is_valid(): return Response({ 'success': False, 'error': '输入验证失败', 'error_code': WebhookErrorCode.VALIDATION_FAILED, 'details': serializer.errors }, status=status.HTTP_400_BAD_REQUEST) # 调用服务处理 service = RequirementWebhookService(serializer.validated_data) success, message, requirement_id, requirement = service.process() if success: return Response({ 'success': True, 'requirement_id': requirement_id, 'structured_data': serializer.validated_data.get('structured_data', {}), 'validation_errors': None, 'warnings': [], 'error': None }, status=status.HTTP_200_OK) else: return Response({ 'success': False, 'error': message, 'error_code': WebhookErrorCode.DB_OPERATION_FAILED }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) except Exception as e: logger.error(f'处理需求解析数据失败: {e}', exc_info=True) return Response({ 'success': False, 'error': f'处理数据失败: {str(e)}', 'error_code': WebhookErrorCode.INTERNAL_ERROR }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) ``` --- ### 阶段 6: 添加单元测试 #### Task 6.1: 创建 Webhook 测试文件 **Files:** - Create: `tests/api/test_webhook_views.py` **Step 1: 创建测试文件** ```python """Webhook API 测试""" import json from datetime import date, time from unittest.mock import patch, MagicMock from django.test import TestCase, RequestFactory from django.http import JsonResponse from apps.api.views.webhook_views import ( ItineraryWebhookView, RequirementWebhookView, WebhookErrorCode ) from apps.api.serializers.webhook_serializers import ( ItineraryWebhookInputSerializer, RequirementWebhookInputSerializer ) class ItineraryWebhookSerializerTest(TestCase): """ItineraryWebhookInputSerializer 测试""" def test_valid_data(self): """测试有效数据""" data = { 'requirement_id': 'REQ_20240101_001', 'itinerary_name': '测试行程', 'start_date': '2024-03-01', 'end_date': '2024-03-05' } serializer = ItineraryWebhookInputSerializer(data=data) self.assertTrue(serializer.is_valid(), serializer.errors) def test_missing_required_field(self): """测试缺少必需字段""" data = { 'requirement_id': 'REQ_20240101_001' # 缺少 itinerary_name } serializer = ItineraryWebhookInputSerializer(data=data) self.assertFalse(serializer.is_valid()) self.assertIn('itinerary_name', serializer.errors) def test_end_date_before_start_date(self): """测试结束日期早于开始日期""" data = { 'requirement_id': 'REQ_20240101_001', 'itinerary_name': '测试行程', 'start_date': '2024-03-05', 'end_date': '2024-03-01' # 早于开始日期 } serializer = ItineraryWebhookInputSerializer(data=data) self.assertFalse(serializer.is_valid()) self.assertIn('end_date', str(serializer.errors)) class ItineraryWebhookViewTest(TestCase): """ItineraryWebhookView 测试""" def setUp(self): self.factory = RequestFactory() self.view = ItineraryWebhookView.as_view() @patch('apps.api.views.webhook_views.Requirement.objects') def test_db_failure_returns_500(self, mock_requirement): """测试数据库失败时返回 500 而非 200""" # 模拟数据库失败 mock_requirement.get.side_effect = Exception('Database connection failed') request = self.factory.post( '/api/webhook/itinerary/', data=json.dumps({ 'requirement_id': 'REQ_20240101_001', 'itinerary_name': '测试', 'start_date': '2024-03-01', 'end_date': '2024-03-05' }), content_type='application/json' ) response = self.view(request) # 关键断言：数据库失败应返回 500 self.assertEqual(response.status_code, 500) response_data = json.loads(response.content) self.assertFalse(response_data.get('success')) self.assertEqual( response_data.get('error_code'), WebhookErrorCode.DB_OPERATION_FAILED ) class LogSanitizerTest(TestCase): """日志脱敏工具测试""" def test_mask_phone(self): """测试手机号脱敏""" from apps.api.utils.logging_utils import LogSanitizer text = "联系电话: 13812345678" result = LogSanitizer.sanitize_dict({'phone': '13812345678'}) self.assertEqual(result['phone'], '13***5678') def test_mask_email(self): """测试邮箱脱敏""" from apps.api.utils.logging_utils import LogSanitizer result = LogSanitizer.sanitize_dict({'email': 'user@example.com'}) self.assertEqual(result['email'], 'u***@example.com') def test_sensitive_field_masking(self): """测试敏感字段自动脱敏""" from apps.api.utils.logging_utils import LogSanitizer data = { 'password': 'secret123', 'api_key': 'sk-abc123xyz', 'username': 'john_doe' # 非敏感字段 } result = LogSanitizer.sanitize_dict(data) self.assertEqual(result['password'], 'se***23') self.assertEqual(result['api_key'], 'sk***yz') self.assertEqual(result['username'], 'john_doe') # 保持原样 ``` **Step 2: 运行测试验证** Run: `python manage.py test tests.api.test_webhook_views -v 2` Expected: 所有测试通过 --- ## 📊 预期改进效果 | 指标 | 改进前 | 改进后 | |------|--------|--------| | **静默失败 Bug** | 存在 | ✅ 修复 | | **post() 方法行数** | 130+ / 200+ | ~40 行 | | **N+1 查询次数** | 最多 51 次 | 最多 4 次 | | **输入验证** | 基础检查 | 完整 Schema 验证 | | **敏感信息泄露** | 存在 | ✅ 已脱敏 | --- ## 🚀 实施顺序建议 1. **Task 1.1** - 创建日志脱敏工具 (安全基础) 2. **Task 3.1** - 修复静默失败 Bug (关键修复) 3. **Task 2.1** - 创建 Serializer 层 (验证基础) 4. **Task 4.1** - 创建服务层 (重构核心) 5. **Task 5.1/5.2** - 重构 View 层 (使用新代码) 6. **Task 6.1** - 添加单元测试 (质量保证) --- **Plan complete and saved to `docs/plans/webhook-views-improvement-plan.md`** **Two execution options:** **1. Subagent-Driven (this session)** - I dispatch fresh subagent per task, review between tasks, fast iteration **2. Parallel Session (separate)** - Open new session with executing-plans, batch execution with checkpoints **Which approach?**