import re import uuid from django.db.models import Q from .models import City, Attraction import logging logger = logging.getLogger(__name__) import json from django.conf import settings from openai import OpenAI from .models import City, Attraction from django.core.cache import cache import traceback # services.py class MoonshotAIService: RED_TOURISM_TAGS = ['红色旅游', '革命', '烈士', '纪念馆', '党史'] BATCH_SIZE = 3 # 分批处理的天数 @staticmethod def generate_travel_plan(preference): """ 红色旅游专用生成方法 - 增强版，支持长行程分批次处理 """ try: # 1. 获取城市和景点 cities = City.objects.filter(id__in=preference['city_ids']) if not cities.exists(): return {'error': '未找到指定城市'} days = preference['days'] # 2. 获取所有景点（一次性查询提高效率） all_attractions = list(Attraction.objects.filter(city__in=cities)) # 3. 根据天数决定生成方式 return MoonshotAIService._generate_long_plan_in_batches(cities, preference, all_attractions) except Exception as e: logger.error(f"生成行程失败: {str(e)}", exc_info=True) return {'error': str(e)} @staticmethod def _generate_long_plan_in_batches(cities, preference, all_attractions): """处理2天及以上的长行程（分批生成），解决景点重复问题""" total_days = preference['days'] batches = (total_days + MoonshotAIService.BATCH_SIZE - 1) // MoonshotAIService.BATCH_SIZE full_plan = { "title": f"{total_days}天红色文化之旅", "description": "通过参观革命历史遗址学习党史", "days": [], "education_goals": [] } # 维护已使用的景点ID集合和名称集合（使用字典提高查找效率） used_attractions = { 'ids': set(), 'names': set() } for batch_num in range(batches): start_day = batch_num * MoonshotAIService.BATCH_SIZE + 1 end_day = min((batch_num + 1) * MoonshotAIService.BATCH_SIZE, total_days) current_batch_days = end_day - start_day + 1 try: # 为当前批次筛选景点（严格排除已使用的景点） batch_attractions = [ att for att in all_attractions if (att.id not in used_attractions['ids'] and att.name not in used_attractions['names']) ] # 确保每批有足够的红色景点 red_attractions = [ att for att in batch_attractions if any(tag in att.tags for tag in MoonshotAIService.RED_TOURISM_TAGS) ] # 计算需要的红色景点数（每天至少2个） needed_red_attractions = current_batch_days * 2 if len(red_attractions) < needed_red_attractions: # 如果红色景点不足，从所有景点中补充 extra_attractions = [ att for att in batch_attractions if att not in red_attractions ] batch_attractions = red_attractions + extra_attractions[ :needed_red_attractions - len(red_attractions)] else: batch_attractions = red_attractions[:needed_red_attractions] + [ att for att in batch_attractions if att not in red_attractions ] # 构建当前批次的提示词 - 添加景点去重提示 batch_prompt = f""" 你是一个专业的红色旅游路线规划AI助手。请为第{start_day}天到第{end_day}天规划红色旅游路线，共{current_batch_days}天。 重要要求： 1. 必须严格避免选择重复景点（已使用景点：{list(used_attractions['names'])[:10]}...） 2. 必须返回严格符合JSON格式的数据 3. 所有字符串值必须用双引号括起来 4. 确保所有JSON字段都有正确的闭合 5. 不要包含任何注释或额外文本 6. 每天必须安排1-5个景点 7. 每个景点必须与红色主题有关 8. 景点之间要考虑地理位置和交通时间 9. 确保每个JSON对象属性后面都有逗号分隔符，除了最后一个属性 10. 所有字符串值必须在一行内完成，不能换行 11. 字符串值中不能包含未转义的引号 特别注意： - 严格避免景点重复（已使用景点：{list(used_attractions['names'])[:5]}...） - "description"、"history_significance"字段必须详细描述(15字左右) - "history_significance"应详细描述该地点在党史中的具体作用和事件 - 所有字符串值必须用双引号正确闭合 - 确保没有缺失的逗号或引号 城市范围：{[city.name for city in cities]} 交通方式：{preference['transport']} 特殊要求：{preference.get('custom_requirements', '无')} 返回格式示例： {{ "days": [ {{ "day": 1, "theme": "主题", "attractions": [ {{ "id": "景点ID", "name": "景点名称", "address": "详细地址", "description": "这里应该提供详细的景点描述，包括建筑特色、主要展区、重要展品等。例如：井冈山革命博物馆始建于1958年，馆藏文物3万余件，珍贵文献资料和历史图片2万余份，保存毛泽东、朱德等领导人重上井冈山时的影视资料数百件。", "history_significance": "这里应该详细描述景点的历史意义。例如：井冈山是中国革命的摇篮，1927年10月，毛泽东率领秋收起义部队到达井冈山，创建了中国第一个农村革命根据地，开辟了农村包围城市、武装夺取政权的中国革命道路。", "open_hours": "08:00-17:00", "ticket_info": "门票信息", "is_red_tourism": true, "latitude": 35.1234, "longitude": 116.5678, "visit_time": "09:00-11:00", "duration": 120 }} ] }} ] }} """ # 调用AI接口 client = OpenAI( api_key=settings.MOONSHOT_API_KEY, base_url="https://api.moonshot.cn/v1" ) response = client.chat.completions.create( model="moonshot-v1-8k", messages=[ { "role": "system", "content": "你是专业的红色旅游规划专家，必须严格按要求返回有效的JSON格式数据。特别注意：\n" "1. 严格避免景点重复（已使用景点：" + ", ".join( list(used_attractions['names'])[:5]) + "...)\n" "2. 提供详细的景点描述、历史意义和学习要点\n" "3. 所有字符串用双引号\n" "4. 确保所有引号正确闭合\n" "5. 不要有注释\n" "6. 确保JSON格式完整正确\n" "7. 对于红色教育基地，必须详细描述其历史背景和教育意义" }, {"role": "user", "content": batch_prompt} ], response_format={"type": "json_object"}, temperature=0.3, max_tokens=6000 ) # 获取并预处理响应内容 response_content = response.choices[0].message.content.strip() logger.debug(f"原始AI响应: {response_content}") # 增强的JSON修复处理 def fix_json(json_str): json_str = json_str.strip() start_idx = json_str.find('{') end_idx = json_str.rfind('}') if start_idx == -1 or end_idx == -1: raise ValueError("无效的JSON结构") json_str = json_str[start_idx:end_idx + 1] # 替换中文标点 json_str = json_str.replace('"', '"').replace('"', '"') json_str = json_str.replace(''', "'").replace(''', "'") json_str = json_str.replace('，', ',').replace('：', ':') # 修复未闭合的字符串 json_str = re.sub(r'(? close_braces: json_str += '}' * (open_braces - close_braces) return json_str # 尝试解析JSON try: batch_data = json.loads(response_content) except json.JSONDecodeError as e: logger.warning(f"首次JSON解析失败，尝试修复... 错误: {str(e)}") try: fixed_content = fix_json(response_content) batch_data = json.loads(fixed_content) logger.info("JSON修复成功") except Exception as fix_error: logger.error(f"JSON修复失败: {str(fix_error)}") logger.error(f"问题内容: {response_content[max(0, e.pos - 50):e.pos + 50]}") raise ValueError(f"无法解析AI响应，请检查响应格式。错误：{str(e)}") # 验证数据结构 if not isinstance(batch_data, dict) or 'days' not in batch_data: raise ValueError("AI返回的数据结构无效，缺少'days'字段") # 处理AI响应并检查景点重复 processed_data = MoonshotAIService._process_red_tourism_response(batch_data, batch_attractions) # 记录已使用的景点 for day in processed_data.get('days', []): for attraction in day.get('attractions', []): if 'id' in attraction: used_attractions['ids'].add(attraction['id']) if 'name' in attraction: used_attractions['names'].add(attraction['name']) # 合并到完整行程中 full_plan['days'].extend(processed_data['days']) if processed_data.get('education_goals'): full_plan['education_goals'].extend(processed_data['education_goals']) except Exception as e: logger.error(f"处理第{batch_num + 1}批行程时出错: {str(e)}", exc_info=True) if len(full_plan['days']) == 0: raise Exception(f"生成行程失败: {str(e)}") else: logger.warning(f"部分行程生成失败，已生成{len(full_plan['days'])}天") # 最终验证和去重检查 if len(full_plan['days']) < total_days: logger.warning(f"生成的行程不完整，只有{len(full_plan['days'])}天/{total_days}天") full_plan['warning'] = f"行程不完整，只生成{len(full_plan['days'])}天" # 最终检查所有景点是否重复 all_attractions_in_plan = [] for day in full_plan['days']: for attraction in day['attractions']: all_attractions_in_plan.append((attraction.get('id'), attraction.get('name'))) unique_attractions = set(all_attractions_in_plan) if len(all_attractions_in_plan) != len(unique_attractions): duplicate_count = len(all_attractions_in_plan) - len(unique_attractions) logger.warning(f"最终检查发现{duplicate_count}个重复景点") # 进行最终去重处理 seen = set() for day in full_plan['days']: day['attractions'] = [ att for att in day['attractions'] if (att.get('id'), att.get('name')) not in seen and not seen.add((att.get('id'), att.get('name'))) ] return full_plan @staticmethod def regenerate_red_tourism_plan(preference): """ 重新规划红色旅游行程 - 增强版，支持长行程分批次处理 """ try: # 1. 获取城市和景点 cities = City.objects.filter(id__in=preference['city_ids']) if not cities.exists(): return {'error': '未找到指定城市'} days = preference['days'] previous_plan = preference.get('previous_plan', {}) # 2. 获取所有景点（一次性查询提高效率） all_attractions = list(Attraction.objects.filter(city__in=cities)) # 3. 获取已使用的景点ID和名称（从原行程中） used_attractions = { 'ids': set(), 'names': set() } if previous_plan and 'days' in previous_plan: for day in previous_plan['days']: for attraction in day.get('attractions', []): if 'id' in attraction: used_attractions['ids'].add(str(attraction['id'])) if 'name' in attraction: used_attractions['names'].add(attraction['name']) # 4. 根据天数决定生成方式 return MoonshotAIService._regenerate_long_plan_in_batches( cities, preference, all_attractions, used_attractions ) except Exception as e: logger.error(f"重新规划行程失败: {str(e)}", exc_info=True) return {'error': str(e)} @staticmethod def _regenerate_long_plan_in_batches(cities, preference, all_attractions, used_attractions): """处理2天及以上的长行程（分批重新生成），解决景点重复问题""" total_days = preference['days'] batches = (total_days + MoonshotAIService.BATCH_SIZE - 1) // MoonshotAIService.BATCH_SIZE full_plan = { "title": f"{total_days}天红色文化之旅(重新规划)", "description": "通过参观革命历史遗址学习党史 - 重新规划版本", "days": [], "education_goals": [] } for batch_num in range(batches): start_day = batch_num * MoonshotAIService.BATCH_SIZE + 1 end_day = min((batch_num + 1) * MoonshotAIService.BATCH_SIZE, total_days) current_batch_days = end_day - start_day + 1 try: # 为当前批次筛选景点（严格排除已使用的景点） batch_attractions = [ att for att in all_attractions if (str(att.id) not in used_attractions['ids'] and att.name not in used_attractions['names']) ] # 确保每批有足够的红色景点 red_attractions = [ att for att in batch_attractions if any(tag in att.tags for tag in MoonshotAIService.RED_TOURISM_TAGS) ] # 计算需要的红色景点数（每天至少2个） needed_red_attractions = current_batch_days * 2 if len(red_attractions) < needed_red_attractions: # 如果红色景点不足，从所有景点中补充 extra_attractions = [ att for att in batch_attractions if att not in red_attractions ] batch_attractions = red_attractions + extra_attractions[ :needed_red_attractions - len(red_attractions)] else: batch_attractions = red_attractions[:needed_red_attractions] + [ att for att in batch_attractions if att not in red_attractions ] # 构建当前批次的提示词 - 添加景点去重提示 batch_prompt = f""" 你是一个专业的红色旅游路线规划AI助手。现在需要基于用户偏好重新规划第{start_day}天到第{end_day}天的红色旅游路线，共{current_batch_days}天。 重要要求： 1. 必须严格避免选择重复景点（已使用景点：{list(used_attractions['names'])[:10]}...） 2. 必须返回严格符合JSON格式的数据 3. 所有字符串值必须用双引号括起来 4. 确保所有JSON字段都有正确的闭合 5. 不要包含任何注释或额外文本 6. 每天必须安排1-5个景点 7. 每个景点必须与红色主题有关 8. 景点之间要考虑地理位置和交通时间 9. 确保每个JSON对象属性后面都有逗号分隔符，除了最后一个属性 10. 所有字符串值必须在一行内完成，不能换行 11. 字符串值中不能包含未转义的引号 特别注意： - 严格避免景点重复（已使用景点：{list(used_attractions['names'])[:5]}...） - 必须生成全新的景点安排，避免与原有行程重复 - "description"、"history_significance"字段必须详细描述(15字左右) - "history_significance"应详细描述该地点在党史中的具体作用和事件 - 所有字符串值必须用双引号正确闭合 - 确保没有缺失的逗号或引号 城市范围：{[city.name for city in cities]} 交通方式：{preference['transport']} 特殊要求：{preference.get('custom_requirements', '无')} 返回格式示例： {{ "days": [ {{ "day": {start_day}, "theme": "新主题", "attractions": [ {{ "id": "新景点ID", "name": "新景点名称", "address": "详细地址", "description": "这里应该提供详细的景点描述，包括建筑特色、主要展区、重要展品等。例如：井冈山革命博物馆始建于1958年，馆藏文物3万余件，珍贵文献资料和历史图片2万余份，保存毛泽东、朱德等领导人重上井冈山时的影视资料数百件。", "history_significance": "这里应该详细描述景点的历史意义。例如：井冈山是中国革命的摇篮，1927年10月，毛泽东率领秋收起义部队到达井冈山，创建了中国第一个农村革命根据地，开辟了农村包围城市、武装夺取政权的中国革命道路。", "open_hours": "08:00-17:00", "ticket_info": "门票信息", "is_red_tourism": true, "latitude": 35.1234, "longitude": 116.5678, "visit_time": "09:00-11:00", "duration": 120 }} ] }} ] }} """ # 调用AI接口 client = OpenAI( api_key=settings.MOONSHOT_API_KEY, base_url="https://api.moonshot.cn/v1" ) response = client.chat.completions.create( model="moonshot-v1-8k", messages=[ { "role": "system", "content": "你是专业的红色旅游规划专家，正在重新规划行程。必须严格按要求返回有效的JSON格式数据。特别注意：\n" "1. 严格避免景点重复（已使用景点：" + ", ".join( list(used_attractions['names'])[:5]) + "...)\n" "2. 必须生成全新的景点安排\n" "3. 提供详细的景点描述、历史意义和学习要点\n" "4. 所有字符串用双引号\n" "5. 确保所有引号正确闭合\n" "6. 不要有注释\n" "7. 确保JSON格式完整正确\n" "8. 对于红色教育基地，必须详细描述其历史背景和教育意义" }, {"role": "user", "content": batch_prompt} ], response_format={"type": "json_object"}, temperature=0.4, # 稍高的temperature以获得更多变化 max_tokens=6000 ) # 获取并预处理响应内容 response_content = response.choices[0].message.content.strip() logger.debug(f"重新规划原始AI响应: {response_content}") # 尝试解析JSON try: batch_data = json.loads(response_content) except json.JSONDecodeError as e: logger.warning(f"重新规划JSON解析失败，尝试修复... 错误: {str(e)}") try: fixed_content = MoonshotAIService._fix_json(response_content) batch_data = json.loads(fixed_content) logger.info("JSON修复成功") except Exception as fix_error: logger.error(f"JSON修复失败: {str(fix_error)}") raise ValueError(f"无法解析AI响应，请检查响应格式。错误：{str(e)}") # 验证数据结构 if not isinstance(batch_data, dict) or 'days' not in batch_data: raise ValueError("AI返回的数据结构无效，缺少'days'字段") # 处理AI响应并检查景点重复 processed_data = MoonshotAIService._process_red_tourism_response(batch_data, batch_attractions) # 检查并处理重复景点 duplicate_attractions = [] for day in processed_data.get('days', []): for attraction in day.get('attractions', []): if ('id' in attraction and attraction['id'] in used_attractions['ids']) or \ ('name' in attraction and attraction['name'] in used_attractions['names']): duplicate_attractions.append(attraction.get('name', '未知景点')) # 记录已使用的景点 for day in processed_data.get('days', []): for attraction in day.get('attractions', []): if 'id' in attraction: used_attractions['ids'].add(attraction['id']) if 'name' in attraction: used_attractions['names'].add(attraction['name']) # 合并到完整行程中 full_plan['days'].extend(processed_data['days']) if processed_data.get('education_goals'): full_plan['education_goals'].extend(processed_data['education_goals']) except Exception as e: logger.error(f"处理第{batch_num + 1}批重新规划行程时出错: {str(e)}", exc_info=True) if len(full_plan['days']) == 0: raise Exception(f"重新生成行程失败: {str(e)}") else: logger.warning(f"部分行程重新生成失败，已生成{len(full_plan['days'])}天") # 最终验证和去重检查 if len(full_plan['days']) < total_days: logger.warning(f"重新生成的行程不完整，只有{len(full_plan['days'])}天/{total_days}天") full_plan['warning'] = f"行程不完整，只生成{len(full_plan['days'])}天" # 最终检查所有景点是否重复 all_attractions_in_plan = [] for day in full_plan['days']: for attraction in day['attractions']: all_attractions_in_plan.append((attraction.get('id'), attraction.get('name'))) unique_attractions = set(all_attractions_in_plan) if len(all_attractions_in_plan) != len(unique_attractions): duplicate_count = len(all_attractions_in_plan) - len(unique_attractions) logger.warning(f"最终检查发现{duplicate_count}个重复景点") # 进行最终去重处理 seen = set() for day in full_plan['days']: day['attractions'] = [ att for att in day['attractions'] if (att.get('id'), att.get('name')) not in seen and not seen.add((att.get('id'), att.get('name'))) ] return full_plan @staticmethod def _process_red_tourism_response(plan_data, attractions): """处理AI响应数据，补充完整景点信息""" attraction_map = {str(att.id): att for att in attractions} for day in plan_data.get('days', []): for attraction in day.get('attractions', []): # 从数据库获取完整景点信息 db_attraction = attraction_map.get(str(attraction.get('id'))) if db_attraction: # 补充详细信息 attraction.update({ 'address': getattr(db_attraction, 'address', '地址信息待补充'), 'description': getattr(db_attraction, 'description', '红色教育基地，具有重要的历史教育意义'), 'history_significance': getattr(db_attraction, 'history_significance', '革命历史重要遗址'), 'open_hours': getattr(db_attraction, 'open_hours', '09:00-17:00'), 'ticket_info': getattr(db_attraction, 'ticket_info', '凭身份证免费参观'), 'image': db_attraction.image.url if hasattr(db_attraction, 'image') and db_attraction.image else '/static/images/default-red.jpg' }) else: # 为AI生成的景点提供默认值 attraction.update({ 'address': attraction.get('address', '地址信息待补充'), 'description': attraction.get('description', '红色教育基地，具有重要的历史教育意义'), 'history_significance': attraction.get('history_significance', '革命历史重要遗址'), 'open_hours': attraction.get('open_hours', '09:00-17:00'), 'ticket_info': attraction.get('ticket_info', '凭身份证免费参观'), 'image': attraction.get('image', '/static/images/default-red.jpg') }) return plan_data