409 lines
16 KiB
Python
409 lines
16 KiB
Python
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) |