import logging import json import base64 import traceback from celery import shared_task from django.utils import timezone from .models import GmailCredential, GmailConversation, UserGoal, ProcessedPushNotification, UnmatchedEmail from .services.gmail_service import GmailService from .services.goal_service import get_conversation_summary, get_last_message, generate_recommended_reply from apps.chat.models import ChatHistory logger = logging.getLogger(__name__) @shared_task( bind=True, max_retries=3, default_retry_delay=60, # 重试延迟60秒 acks_late=True, # 任务完成后再确认 time_limit=600, # 任务超时限制600秒 soft_time_limit=300 # 软超时限制300秒 ) def process_push_notification(self, message_data, message_id, subscription): """ 处理Gmail推送通知的Celery任务 Args: message_data: Base64编码的消息数据 message_id: 消息ID subscription: 订阅信息 Returns: bool: 处理是否成功 """ try: logger.info(f"开始处理推送通知: {message_id}") print(f"[Gmail Webhook Task] 处理推送通知: {message_id}") # 检查是否已处理过该消息 if ProcessedPushNotification.objects.filter(message_id=message_id).exists(): logger.info(f"通知 {message_id} 已处理过，跳过") print(f"[Gmail Webhook Task] 通知 {message_id} 已处理过，跳过") return True # 尝试创建一个处理记录，标记为开始处理 notification_record = ProcessedPushNotification.objects.create( message_id=message_id, email_address="processing", history_id="unknown", is_successful=False, metadata={ "subscription": subscription, "started_at": timezone.now().isoformat() } ) # Base64解码消息数据 try: decoded_data = json.loads(base64.b64decode(message_data).decode('utf-8')) print(f"[Gmail Webhook Task] 解码后的数据: {decoded_data}") # 获取Gmail通知关键信息 email_address = decoded_data.get('emailAddress') history_id = decoded_data.get('historyId') if not email_address: logger.error("[Gmail Webhook Task] 错误: 推送数据缺少邮箱地址") # 更新处理记录 notification_record.metadata.update({ "error": "推送数据缺少邮箱地址", "decoded_data": decoded_data }) notification_record.save() return False # 更新处理记录 notification_record.email_address = email_address notification_record.history_id = history_id or "unknown" notification_record.metadata.update(decoded_data) notification_record.save() # 查找对应的Gmail凭证 credential = GmailCredential.objects.filter(email=email_address, is_valid=True).first() if not credential: logger.warning(f"[Gmail Webhook Task] 警告: 未找到对应的Gmail凭证: {email_address}") notification_record.metadata.update({ "error": f"未找到对应的Gmail凭证: {email_address}" }) notification_record.save() return False user = credential.user logger.info(f"[Gmail Webhook Task] 找到有效凭证: 用户ID {user.id}, 邮箱 {email_address}") # 处理新邮件 processed_emails = GmailService.process_new_emails(user, credential, history_id) logger.info(f"[Gmail Webhook Task] 新邮件处理完成，共 {len(processed_emails)} 封") # 处理达人回复状态更新 if processed_emails: update_creator_conversation_status(user, processed_emails) # 处理自动回复 if processed_emails: try: process_auto_replies.delay(user.id, credential.id, processed_emails) logger.info(f"[Gmail Webhook Task] 已提交自动回复任务处理 {len(processed_emails)} 封邮件") except Exception as reply_task_error: logger.error(f"[Gmail Webhook Task] 提交自动回复任务失败: {str(reply_task_error)}") notification_record.metadata.update({ "auto_reply_error": str(reply_task_error), "processed_emails": processed_emails }) notification_record.save() # 更新凭证的历史ID if history_id: credential.last_history_id = history_id credential.save() logger.info(f"[Gmail Webhook Task] 已更新凭证历史ID: {history_id}") # 更新通知记录为成功 notification_record.is_successful = True notification_record.metadata.update({ "completed_at": timezone.now().isoformat(), "processed_emails_count": len(processed_emails) }) notification_record.save() return True except json.JSONDecodeError as e: logger.error(f"[Gmail Webhook Task] JSON解码失败: {str(e)}") notification_record.metadata.update({ "error": f"JSON解码失败: {str(e)}", "raw_data": message_data[:100] if message_data else "None" # 只保存前100个字符，避免数据过大 }) notification_record.save() # 抛出异常以便Celery可能进行重试 raise self.retry(exc=e, countdown=30) except Exception as decode_error: logger.error(f"[Gmail Webhook Task] 解析推送数据失败: {str(decode_error)}") notification_record.metadata.update({ "error": f"解析推送数据失败: {str(decode_error)}", "traceback": traceback.format_exc() }) notification_record.save() # 抛出异常以便Celery可能进行重试 raise self.retry(exc=decode_error, countdown=30) except Exception as e: logger.error(f"[Gmail Webhook Task] 处理推送通知失败: {str(e)}") logger.error(traceback.format_exc()) # 尝试更新通知记录 try: notification_record = ProcessedPushNotification.objects.filter(message_id=message_id).first() if notification_record: notification_record.metadata.update({ "error": str(e), "traceback": traceback.format_exc(), "failed_at": timezone.now().isoformat() }) notification_record.save() except Exception as record_error: logger.error(f"[Gmail Webhook Task] 更新通知记录失败: {str(record_error)}") # 尝试重试任务 raise self.retry(exc=e, countdown=60) def update_creator_conversation_status(user, email_ids): """ 根据新处理的邮件更新达人对话跟踪状态 Args: user: 用户对象 email_ids: 处理的邮件ID列表 Returns: bool: 处理是否成功 """ try: # 尝试导入CreatorConversationTracker模型 try: from apps.feishu.models import CreatorConversationTracker except ImportError: logger.warning("CreatorConversationTracker模型不存在，跳过更新达人对话状态") return False # 遍历每个邮件ID for email_id in email_ids: # 通过邮件ID查找对应的聊天记录 chat_msg = ChatHistory.objects.filter( metadata__gmail_message_id=email_id ).order_by('-created_at').first() if not chat_msg: logger.info(f"邮件 {email_id} 未找到对应的聊天记录，跳过状态更新") continue # 只处理达人的消息 if chat_msg.role != 'assistant': continue conversation_id = chat_msg.conversation_id # 查找关联的对话 conversation = GmailConversation.objects.filter( conversation_id=conversation_id ).first() if not conversation: logger.info(f"未找到对话 {conversation_id}，跳过状态更新") continue # 查找对应的跟踪记录 tracker = CreatorConversationTracker.objects.filter( conversation_id=conversation_id ).first() if not tracker: logger.info(f"未找到对话 {conversation_id} 的跟踪记录，跳过状态更新") continue # 更新回复状态 tracker.has_replied = True # 检查是否达成目标 goal_achieved = check_goal_achievement(conversation_id) if goal_achieved: tracker.goal_achieved = True # 保存更新 tracker.updated_at = timezone.now() tracker.save() logger.info(f"更新了对话 {conversation_id} 的跟踪状态: has_replied=True, goal_achieved={goal_achieved}") return True except Exception as e: logger.error(f"更新达人对话跟踪状态失败: {str(e)}") import traceback logger.error(traceback.format_exc()) return False def check_goal_achievement(conversation_id): """ 检查对话是否达成目标（找到价格信息） Args: conversation_id: 对话ID Returns: bool: 是否达成目标 """ try: # 获取对话中的所有消息 messages = ChatHistory.objects.filter(conversation_id=conversation_id).order_by('created_at') # 简单判断：检查是否有回复中包含价格信息 keywords = ['rate', 'price', 'cost', '$', 'usd', 'charge', 'fee'] for message in messages: # 只检查达人的回复 if message.role == 'assistant': content_lower = message.content.lower() if any(keyword in content_lower for keyword in keywords): logger.info(f"对话 {conversation_id} 已达成目标，找到价格信息") return True return False except Exception as e: logger.error(f"检查目标达成失败: {str(e)}") return False @shared_task( bind=True, max_retries=2, default_retry_delay=60, time_limit=300, soft_time_limit=240 ) def process_auto_replies(self, user_id, credential_id, email_ids): """ 处理自动回复的任务 Args: user_id: 用户ID credential_id: Gmail凭证ID email_ids: 需要处理的邮件ID列表 Returns: bool: 处理是否成功 """ try: from django.contrib.auth import get_user_model User = get_user_model() user = User.objects.get(id=user_id) credential = GmailCredential.objects.get(id=credential_id) logger.info(f"[Auto Reply Task] 开始处理{len(email_ids)}封邮件的自动回复") # 遍历每个处理过的邮件ID for email_id in email_ids: try: # 通过邮件ID查找对应的聊天记录 chat_msg = ChatHistory.objects.filter( metadata__gmail_message_id=email_id ).order_by('-created_at').first() if not chat_msg: logger.info(f"[Auto Reply Task] 邮件 {email_id} 未找到对应的聊天记录，跳过自动回复") continue conversation_id = chat_msg.conversation_id # 检查对话是否是活跃的 conversation = GmailConversation.objects.filter( conversation_id=conversation_id, is_active=True ).first() if not conversation: logger.info(f"[Auto Reply Task] 对话 {conversation_id} 不是活跃的，跳过自动回复") continue logger.info(f"[Auto Reply Task] 发现活跃对话 {conversation_id}，准备自动回复") # 检查邮件角色，只有达人发送的邮件才自动回复 if chat_msg.role != 'assistant': logger.info(f"[Auto Reply Task] 邮件 {email_id} 不是达人发送的，跳过自动回复") continue # 查找对话的目标 goal = UserGoal.objects.filter( conversation=conversation, is_active=True ).first() if not goal: logger.info(f"[Auto Reply Task] 对话 {conversation_id} 没有活跃目标，跳过自动回复") continue # 获取对话摘要 conversation_summary = get_conversation_summary(conversation_id) if not conversation_summary: conversation_summary = "无对话摘要" # 获取最后一条达人消息 last_message = get_last_message(conversation_id) if not last_message: logger.info(f"[Auto Reply Task] 对话 {conversation_id} 没有达人消息，跳过自动回复") continue # 生成推荐回复 reply_content, error = generate_recommended_reply( user=user, goal_description=goal.description, conversation_summary=conversation_summary, last_message=last_message ) if error: logger.error(f"[Auto Reply Task] 生成推荐回复失败: {error}") continue if not reply_content: logger.warning(f"[Auto Reply Task] 生成的推荐回复为空") continue # 构建回复的主题 subject = chat_msg.metadata.get('subject', '') reply_subject = f"回复: {subject}" if not subject.startswith('回复:') else subject # 准备发送自动回复 logger.info(f"[Auto Reply Task] 准备发送自动回复: 从{conversation.user_email}到{conversation.influencer_email}") success, reply_message_id = GmailService.send_email( user=user, user_email=conversation.user_email, to_email=conversation.influencer_email, subject=reply_subject, body=reply_content ) if success: logger.info(f"[Auto Reply Task] 已成功发送自动回复: {reply_message_id}") # 更新目标状态 goal.last_activity_time = timezone.now() if goal.status == 'pending': goal.status = 'in_progress' goal.save(update_fields=['last_activity_time', 'status', 'updated_at']) else: logger.error(f"[Auto Reply Task] 发送自动回复失败: {reply_message_id}") except Exception as reply_error: logger.error(f"[Auto Reply Task] 处理自动回复过程中出错: {str(reply_error)}") import traceback logger.error(traceback.format_exc()) return True except Exception as e: logger.error(f"[Auto Reply Task] 处理自动回复失败: {str(e)}") import traceback logger.error(traceback.format_exc()) raise self.retry(exc=e, countdown=30)