import os import json import uuid import base64 import pickle import logging import traceback # 确保导入traceback模块 from pathlib import Path import dateutil.parser as parser from datetime import datetime, timedelta from bs4 import BeautifulSoup from django.utils import timezone from django.conf import settings # 添加Django设置导入 from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.mime.base import MIMEBase from email import encoders import warnings import mimetypes import requests import django.db.utils import time from django.db.models import Q import socket # 忽略oauth2client相关的所有警告 warnings.filterwarnings('ignore', message='file_cache is unavailable when using oauth2client >= 4.0.0 or google-auth') warnings.filterwarnings('ignore', category=DeprecationWarning, module='oauth2client') warnings.filterwarnings('ignore', message='.*google-auth.*') warnings.filterwarnings('ignore', message='.*oauth2client.*') from apiclient import discovery from httplib2 import Http from oauth2client import file, client, tools from .models import GmailCredential, KnowledgeBase, KnowledgeBaseDocument, User, GmailTalentMapping, ChatHistory, GmailAttachment, UserProfile # Gmail API调用超时设置（秒） GMAIL_REQUEST_TIMEOUT = getattr(settings, 'GMAIL_REQUEST_TIMEOUT', 30) logger = logging.getLogger(__name__) # Gmail服务单例管理器 class GmailServiceManager: _instances = {} # 以用户ID和Gmail凭证ID为键存储Gmail服务实例 @classmethod def get_instance(cls, user, gmail_credential_id=None): """ 获取用户的Gmail服务实例，如果不存在则创建 参数: user: 用户对象 gmail_credential_id: 指定Gmail凭证ID，如果为None，则使用默认凭证 """ user_id = str(user.id) # 生成实例键，组合用户ID和Gmail凭证ID if gmail_credential_id: instance_key = f"{user_id}:{gmail_credential_id}" else: instance_key = user_id if instance_key not in cls._instances: try: # 从数据库获取认证信息 if gmail_credential_id: # 获取指定ID的凭证 credential = GmailCredential.objects.filter( id=gmail_credential_id, user=user, is_active=True ).first() else: # 获取默认凭证，优先获取is_default=True的凭证 credential = GmailCredential.objects.filter( user=user, is_active=True, is_default=True ).first() # 如果没有找到默认凭证，则获取最近更新的一个凭证 if not credential: credential = GmailCredential.objects.filter( user=user, is_active=True ).order_by('-updated_at').first() if credential and credential.credentials: # 反序列化凭证 creds = pickle.loads(credential.credentials) # 检查凭证是否有效 if not creds.invalid: # 初始化服务 gmail_service = discovery.build('gmail', 'v1', http=creds.authorize(Http())) # 存储实例 cls._instances[instance_key] = { 'service': gmail_service, 'credentials': creds, 'timestamp': timezone.now(), 'user': user, 'gmail_credential': credential } logger.info(f"创建用户 {user.username} 的Gmail服务单例，Gmail账号: {credential.gmail_email or '未知'}，名称: {credential.name}") return cls._instances[instance_key] except Exception as e: logger.error(f"创建Gmail服务单例失败: {e}") else: # 检查实例是否过期(超过30分钟) instance = cls._instances[instance_key] time_diff = timezone.now() - instance['timestamp'] if time_diff.total_seconds() > 1800: # 30分钟过期 del cls._instances[instance_key] return cls.get_instance(user, gmail_credential_id) # 递归调用，重新创建 # 更新时间戳 cls._instances[instance_key]['timestamp'] = timezone.now() credential = instance.get('gmail_credential') gmail_email = credential.gmail_email if credential else '未知' credential_name = credential.name if credential else '默认' logger.info(f"复用用户 {user.username} 的Gmail服务单例，Gmail账号: {gmail_email}，名称: {credential_name}") return cls._instances[instance_key] return None @classmethod def get_credential_instance(cls, credential): """通过GmailCredential对象获取服务实例""" if not credential or not credential.user: return None return cls.get_instance(credential.user, str(credential.id)) @classmethod def update_instance(cls, user, credentials, service, gmail_credential=None): """更新用户的Gmail服务实例""" user_id = str(user.id) # 确定实例键 if gmail_credential: instance_key = f"{user_id}:{gmail_credential.id}" else: instance_key = user_id cls._instances[instance_key] = { 'service': service, 'credentials': credentials, 'timestamp': timezone.now(), 'user': user, 'gmail_credential': gmail_credential } @classmethod def clear_instance(cls, user, gmail_credential_id=None): """清除用户的Gmail服务实例""" user_id = str(user.id) # 清除特定凭证的实例 if gmail_credential_id: instance_key = f"{user_id}:{gmail_credential_id}" if instance_key in cls._instances: del cls._instances[instance_key] else: # 清除该用户的所有实例 keys_to_delete = [] for key in cls._instances.keys(): if key == user_id or key.startswith(f"{user_id}:"): keys_to_delete.append(key) for key in keys_to_delete: del cls._instances[key] @classmethod def get_all_user_instances(cls, user): """获取用户的所有Gmail服务实例""" user_id = str(user.id) user_instances = {} # 收集该用户所有的实例 for key, instance in cls._instances.items(): if key == user_id or key.startswith(f"{user_id}:"): credential = instance.get('gmail_credential') if credential: user_instances[str(credential.id)] = instance return user_instances @classmethod def clear_all_instances_by_email(cls, gmail_email): """ 清除与特定Gmail邮箱关联的所有服务实例 参数: gmail_email: Gmail邮箱地址 返回: int: 清除的实例数量 """ try: # 导入GmailCredential模型 from .models import GmailCredential # 查找与该邮箱关联的所有凭证 credentials = GmailCredential.objects.filter( gmail_email=gmail_email, is_active=True ).select_related('user') # 记录清除的实例数量 cleared_count = 0 # 清除每个凭证关联的实例 for credential in credentials: user = credential.user # 清除特定用户和凭证的实例 if cls.clear_instance(user, str(credential.id)): cleared_count += 1 # 同时清除用户的默认实例 user_id = str(user.id) if user_id in cls._instances: del cls._instances[user_id] cleared_count += 1 return cleared_count except Exception as e: import logging logger = logging.getLogger(__name__) logger.error(f"清除Gmail邮箱 {gmail_email} 的服务实例时出错: {str(e)}") return 0 class GmailIntegration: """Gmail集成类""" # Gmail API 权限范围 SCOPES = ['https://mail.google.com/'] def __init__(self, user, email=None, client_secret_json=None, use_proxy=True, proxy_url='http://127.0.0.1:7890', gmail_credential_id=None): self.user = user self.user_email = user.email if user else None self.email = email # 目标邮箱 self.client_secret = client_secret_json self.credentials = None self.gmail_service = None self.use_proxy = use_proxy self.proxy_url = proxy_url self.gmail_credential_id = gmail_credential_id self.gmail_credential = None # 设置代理 if self.use_proxy: logger.info(f"设置Gmail API代理: {proxy_url}") os.environ['HTTP_PROXY'] = proxy_url os.environ['HTTPS_PROXY'] = proxy_url # 设置Token文件存储路径 if user: # 使用用户ID创建唯一的token存储路径 token_file = f"gmail_token_{user.id}.json" if gmail_credential_id: token_file = f"gmail_token_{user.id}_{gmail_credential_id}.json" self.token_storage_path = os.path.join("gmail_tokens", token_file) # 确保目录存在 os.makedirs("gmail_tokens", exist_ok=True) logger.info(f"设置Token存储路径: {self.token_storage_path}") # 尝试从数据库加载凭证 try: if gmail_credential_id: # 加载指定ID的凭证 gmail_cred = GmailCredential.objects.filter( id=gmail_credential_id, user=user, is_active=True ).first() else: # 加载默认凭证 gmail_cred = GmailCredential.objects.filter( user=user, is_active=True, is_default=True ).first() # 如果没有默认凭证，加载最新的一个 if not gmail_cred: gmail_cred = GmailCredential.objects.filter( user=user, is_active=True ).order_by('-updated_at').first() if gmail_cred and gmail_cred.credentials: self.gmail_credential = gmail_cred logger.info(f"从数据库加载用户 {user.username} 的Gmail凭证 (ID: {gmail_cred.id}, Email: {gmail_cred.gmail_email or '未知'}, 名称: {gmail_cred.name})") # 使用新方法加载凭证 creds = self._load_credentials_from_storage(gmail_cred.credentials) if creds: self.credentials = creds # 初始化Gmail服务 self.gmail_service = discovery.build('gmail', 'v1', http=creds.authorize(Http())) logger.info("从数据库凭证初始化Gmail服务成功") else: logger.error("无法从数据库中加载有效凭证") # 标记需要重新认证 if hasattr(gmail_cred, 'needs_reauth'): gmail_cred.needs_reauth = True gmail_cred.save() logger.info(f"已将Gmail账号 {gmail_cred.gmail_email} 标记为需要重新认证") except Exception as e: logger.error(f"从数据库加载凭证失败: {str(e)}") # 继续使用文件方式 else: self.token_storage_path = "gmail_token.json" logger.warning("未提供用户，将使用默认Token存储路径") # 存储对象 self.storage = file.Storage(self.token_storage_path) def authenticate(self): """获取Gmail API凭证并认证""" try: # 优先尝试使用单例服务 if self.user: instance = GmailServiceManager.get_instance(self.user, self.gmail_credential_id) if instance: self.gmail_service = instance['service'] self.credentials = instance['credentials'] self.gmail_credential = instance.get('gmail_credential') credential_id = str(self.gmail_credential.id) if self.gmail_credential else "未知" gmail_email = self.gmail_credential.gmail_email if self.gmail_credential else "未知" credential_name = self.gmail_credential.name if self.gmail_credential else "默认" logger.info(f"使用现有的Gmail服务单例 (ID: {credential_id}, Email: {gmail_email}, 名称: {credential_name})") return True # 以下是原有的认证逻辑... # 写入client_secret.json if self.client_secret: client_secret_path = 'client_secret.json' with open(client_secret_path, 'w') as f: if isinstance(self.client_secret, str): try: # 确保是有效的JSON import json # 在本地作用域引入 json_data = json.loads(self.client_secret) # 强制设置redirect_uris为非浏览器模式，避免localhost连接拒绝问题 for key in ['web', 'installed']: if key in json_data and 'redirect_uris' in json_data[key]: json_data[key]['redirect_uris'] = ['urn:ietf:wg:oauth:2.0:oob'] logger.info("已强制设置redirect_uri为非浏览器模式") json.dump(json_data, f) except json.JSONDecodeError as e: logger.error(f"client_secret不是有效的JSON: {str(e)}") return False else: # 如果是字典，也进行相同处理 import json # 在本地作用域引入 client_secret_dict = dict(self.client_secret) for key in ['web', 'installed']: if key in client_secret_dict and 'redirect_uris' in client_secret_dict[key]: client_secret_dict[key]['redirect_uris'] = ['urn:ietf:wg:oauth:2.0:oob'] logger.info("已强制设置redirect_uri为非浏览器模式") json.dump(client_secret_dict, f) logger.info(f"已将client_secret写入临时文件: {client_secret_path}") # 使用与quickstart.py相同的流程 logger.info(f"创建或读取token存储: {self.token_storage_path}") # 确保token目录存在 token_dir = os.path.dirname(self.token_storage_path) if token_dir and not os.path.exists(token_dir): logger.info(f"创建token目录: {token_dir}") os.makedirs(token_dir) store = file.Storage(self.token_storage_path) creds = store.get() if not creds or creds.invalid: logger.info("没有有效的凭证，需要重新授权") if not self.client_secret: logger.error("没有提供client_secret_json且找不到有效凭证") return False # 强制使用非浏览器认证模式，避免localhost连接问题 logger.info("使用非浏览器认证模式") redirect_uri = 'urn:ietf:wg:oauth:2.0:oob' flow = client.flow_from_clientsecrets('client_secret.json', self.SCOPES) flow.redirect_uri = redirect_uri # 获取授权URL并抛出异常 auth_url = flow.step1_get_authorize_url() logger.info(f"获取授权URL: {auth_url[:50]}...") raise Exception(f"Please visit this URL to authorize: {auth_url}") # 如果有有效凭证，初始化服务 self.credentials = creds logger.info("使用现有凭证初始化Gmail服务") self.gmail_service = discovery.build('gmail', 'v1', http=creds.authorize(Http())) logger.info("Gmail服务初始化成功") # 获取Gmail账号信息 gmail_email = None try: # 调用Gmail API获取用户资料 profile = self.gmail_service.users().getProfile(userId='me').execute() gmail_email = profile.get('emailAddress') logger.info(f"获取到Gmail账号: {gmail_email}") except Exception as profile_error: logger.error(f"获取Gmail账号失败: {str(profile_error)}") # 即使获取Gmail账号失败，也要尝试获取消息以提取邮箱 if not gmail_email: try: # 尝试获取一条消息来提取邮箱 messages = self.gmail_service.users().messages().list(userId='me', maxResults=1).execute() if 'messages' in messages and len(messages['messages']) > 0: msg_id = messages['messages'][0]['id'] msg = self.gmail_service.users().messages().get(userId='me', id=msg_id).execute() # 从消息中查找与当前用户匹配的邮箱 if 'payload' in msg and 'headers' in msg['payload']: for header in msg['payload']['headers']: if header['name'] in ['From', 'To', 'Cc', 'Bcc']: if '<' in header['value'] and '>' in header['value']: email = header['value'].split('<')[1].split('>')[0] # 如果找到一个邮箱就使用它 if not gmail_email: gmail_email = email logger.info(f"从消息中提取到Gmail账号: {gmail_email}") except Exception as msg_error: logger.error(f"尝试从消息中提取Gmail账号失败: {str(msg_error)}") # 保存凭证到数据库 if self.user: from django.utils import timezone logger.info("保存凭证到数据库") # 将凭证对象序列化 - 改为JSON序列化 try: # 首先尝试使用JSON序列化 credentials_data = creds.to_json() except Exception as json_error: logger.error(f"JSON序列化凭证失败: {str(json_error)}") # 备选使用pickle序列化 credentials_data = pickle.dumps(creds) logger.info("使用pickle序列化凭证") # 如果提供了具体的gmail_credential_id，更新对应的记录 if self.gmail_credential_id: # 更新指定ID的凭证 try: gmail_credential = GmailCredential.objects.get( id=self.gmail_credential_id, user=self.user ) # 更新凭证信息 gmail_credential.credentials = credentials_data gmail_credential.gmail_email = gmail_email gmail_credential.updated_at = timezone.now() gmail_credential.is_active = True gmail_credential.save() self.gmail_credential = gmail_credential logger.info(f"已更新ID为 {self.gmail_credential_id} 的Gmail凭证，Gmail账号: {gmail_email}") except GmailCredential.DoesNotExist: logger.error(f"未找到ID为 {self.gmail_credential_id} 的Gmail凭证，将创建新凭证") # 如果指定的凭证不存在，创建新凭证 self.gmail_credential_id = None # 如果没有具体的gmail_credential_id，或者指定的不存在，创建或更新默认凭证 if not self.gmail_credential_id: # 检查是否已存在相同gmail_email的凭证 existing_credential = None if gmail_email: existing_credential = GmailCredential.objects.filter( user=self.user, gmail_email=gmail_email, is_active=True ).first() if existing_credential: # 更新现有凭证 existing_credential.credentials = credentials_data existing_credential.token_path = self.token_storage_path existing_credential.updated_at = timezone.now() existing_credential.save() self.gmail_credential = existing_credential logger.info(f"已更新Gmail账号 {gmail_email} 的现有凭证") else: # 获取Gmail账户数量 gmail_count = GmailCredential.objects.filter(user=self.user).count() # 创建新凭证 name = f"Gmail账号 {gmail_count + 1}" if gmail_email: name = gmail_email else: # 确保gmail_email有默认值，避免null错误 gmail_email = "未知邮箱" # 将凭证转换为JSON字符串 if isinstance(credentials_data, dict): # 确保json模块在本地作用域可访问 import json credentials_data = json.dumps(credentials_data) gmail_credential = GmailCredential.objects.create( user=self.user, credentials=credentials_data, token_path=self.token_storage_path, gmail_email=gmail_email, name=name, is_default=(gmail_count == 0), # 第一个账号设为默认 updated_at=timezone.now(), is_active=True ) self.gmail_credential = gmail_credential logger.info(f"已创建新的Gmail凭证，Gmail账号: {gmail_email}, 名称: {name}") # 认证成功后更新单例 if self.user and self.gmail_service and self.credentials: GmailServiceManager.update_instance( self.user, self.credentials, self.gmail_service, self.gmail_credential ) return True except Exception as e: # 保留授权URL异常，视图层会处理 if "Please visit this URL" in str(e): raise e logger.error(f"Gmail认证失败: {str(e)}") import traceback logger.error(traceback.format_exc()) return False finally: # 清理client_secret.json文件 if self.client_secret and os.path.exists('client_secret.json'): logger.info("删除临时client_secret文件") os.unlink('client_secret.json') def create_talent_knowledge_base(self, talent_email): """ 创建或获取与talent_email关联的知识库 Args: talent_email (str): 达人Gmail邮箱地址 Returns: tuple: (knowledge_base, created) - 知识库对象和是否新创建的标志 """ try: # 优先查找现有的关联关系 mapping = GmailTalentMapping.objects.filter( user=self.user, talent_email=talent_email, is_active=True ).first() if mapping and mapping.knowledge_base: logger.info(f"找到现有的Gmail-达人映射: {talent_email} -> {mapping.knowledge_base.name}") return mapping.knowledge_base, False # 查找与该达人邮箱关联的知识库 # 根据达人邮箱生成一个唯一的标识名称 kb_name = f"Gmail-{talent_email.split('@')[0]}" # 检查该名称的知识库是否已存在 existing_kb = KnowledgeBase.objects.filter( name=kb_name, user_id=self.user.id ).first() if existing_kb: logger.info(f"找到现有知识库: {kb_name}") # 创建映射关系 GmailTalentMapping.objects.update_or_create( user=self.user, talent_email=talent_email, defaults={ 'knowledge_base': existing_kb, 'is_active': True } ) return existing_kb, False # 没有找到现有知识库，直接使用基本方法创建 logger.info(f"使用基本方法创建知识库: {kb_name}") return self._create_knowledge_base_basic(talent_email) except Exception as e: logger.error(f"创建或获取Gmail-达人知识库失败: {str(e)}") import traceback logger.error(traceback.format_exc()) # 创建失败时，尝试使用基本方法创建 logger.info("尝试使用基本方法创建知识库") return self._create_knowledge_base_basic(talent_email) def _create_knowledge_base_basic(self, talent_email): """创建基础知识库，不处理映射关系""" try: # 根据达人邮箱生成一个唯一的标识名称 kb_name = f"Gmail-{talent_email.split('@')[0]}" # 检查该名称的知识库是否已存在 existing_kb = KnowledgeBase.objects.filter( name=kb_name, user_id=self.user.id ).first() if existing_kb: logger.info(f"找到现有知识库: {kb_name}") # 检查external_id是否存在，如果不存在则创建 if not existing_kb.external_id: logger.info(f"知识库 {kb_name} 缺少external_id，尝试创建外部知识库") try: from .views import KnowledgeBaseViewSet kb_viewset = KnowledgeBaseViewSet() external_id = kb_viewset._create_external_dataset(existing_kb) if external_id: existing_kb.external_id = external_id existing_kb.save() logger.info(f"成功为现有知识库创建外部知识库: {external_id}") else: logger.error("创建外部知识库失败：未返回external_id") return {"conversation_id": None, "success": False, "error": "创建外部知识库失败"} except Exception as e: logger.error(f"为现有知识库创建外部知识库失败: {str(e)}") return {"conversation_id": None, "success": False, "error": f"创建外部知识库失败: {str(e)}"} # 创建映射关系 GmailTalentMapping.objects.update_or_create( user=self.user, talent_email=talent_email, defaults={ 'knowledge_base': existing_kb, 'is_active': True } ) return existing_kb, False # 创建新知识库 try: knowledge_base = KnowledgeBase.objects.create( name=kb_name, desc=f"与{talent_email}的Gmail邮件交流记录", type="private", user_id=self.user.id, documents=[] ) except django.db.utils.IntegrityError as e: # 处理名称重复的情况 logger.warning(f"知识库名称'{kb_name}'已存在，尝试获取或创建带随机后缀的名称") # 先尝试查找已存在的知识库（不限制用户） existing_kb = KnowledgeBase.objects.filter(name=kb_name).first() if existing_kb and str(existing_kb.user_id) == str(self.user.id): # 如果存在且属于当前用户，直接使用 logger.info(f"找到属于当前用户的知识库: {kb_name}") # 创建映射关系 GmailTalentMapping.objects.update_or_create( user=self.user, talent_email=talent_email, defaults={ 'knowledge_base': existing_kb, 'is_active': True } ) return existing_kb, False else: # 如果不存在或不属于当前用户，创建带随机后缀的新知识库 import random import string suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=5)) new_kb_name = f"{kb_name}-{suffix}" logger.info(f"创建带随机后缀的知识库: {new_kb_name}") knowledge_base = KnowledgeBase.objects.create( name=new_kb_name, desc=f"与{talent_email}的Gmail邮件交流记录", type="private", user_id=self.user.id, documents=[] ) # 创建外部知识库 try: from .views import KnowledgeBaseViewSet kb_viewset = KnowledgeBaseViewSet() external_id = kb_viewset._create_external_dataset(knowledge_base) if external_id: knowledge_base.external_id = external_id knowledge_base.save() logger.info(f"成功创建外部知识库: {external_id}") except Exception as e: logger.error(f"创建外部知识库失败: {str(e)}") # 继续执行，不影响基本功能 # 创建映射关系 GmailTalentMapping.objects.create( user=self.user, talent_email=talent_email, knowledge_base=knowledge_base, is_active=True ) logger.info(f"成功创建新知识库: {knowledge_base.name}, ID: {knowledge_base.id}") return knowledge_base, True except Exception as e: logger.error(f"创建或获取Gmail-达人知识库基本方法失败: {str(e)}") import traceback logger.error(traceback.format_exc()) # 尝试直接获取已存在的知识库作为最后手段 try: kb_name = f"Gmail-{talent_email.split('@')[0]}" existing_kb = KnowledgeBase.objects.filter(name__startswith=kb_name).first() if existing_kb: logger.info(f"在错误处理中找到可用的知识库: {existing_kb.name}") # 创建映射关系 GmailTalentMapping.objects.update_or_create( user=self.user, talent_email=talent_email, defaults={ 'knowledge_base': existing_kb, 'is_active': True } ) return existing_kb, False except Exception as inner_e: logger.error(f"错误处理中尝试获取知识库失败: {str(inner_e)}") # 如果所有尝试都失败，抛出异常 raise def get_conversations(self, talent_gmail): """获取与特定用户的所有邮件对话""" try: if not self.gmail_service: logger.error("Gmail服务未初始化") return [] # 使用crushwds@gmail.com作为固定邮箱 email1 = "crushwds@gmail.com" # 固定设置为quickstart.py中的邮箱 email2 = talent_gmail # 使用参数传入的目标邮箱 logger.info(f"执行Gmail查询: {email1} 与 {email2}") # 构建搜索查询 query = f"from:({email1} OR {email2}) to:({email1} OR {email2})" # 获取所有匹配的邮件 response = self.gmail_service.users().messages().list(userId='me', q=query).execute() messages = [] if 'messages' in response: messages.extend(response['messages']) message_count = len(messages) logger.info(f"找到 {message_count} 封邮件") # 分页获取所有邮件 while 'nextPageToken' in response: page_token = response['nextPageToken'] response = self.gmail_service.users().messages().list( userId='me', q=query, pageToken=page_token ).execute() if 'messages' in response: new_messages = response['messages'] messages.extend(new_messages) new_count = len(new_messages) message_count += new_count logger.info(f"加载额外 {new_count} 封邮件，当前总数: {message_count}") else: logger.warning(f"未找到匹配邮件") # 处理每封邮件 conversations = [] if messages: logger.info(f"开始获取 {len(messages)} 封邮件详情...") for i, msg in enumerate(messages): try: message = self.gmail_service.users().messages().get(userId='me', id=msg['id']).execute() email_data = self._extract_email_content(message) if email_data: conversations.append(email_data) if (i+1) % 5 == 0 or i+1 == len(messages): # 每5封邮件或最后一封邮件时记录进度 logger.info(f"已处理 {i+1}/{len(messages)} 封邮件") else: logger.warning(f"邮件 {i+1}/{len(messages)} 内容提取失败") except Exception as e: logger.error(f"获取邮件 {msg['id']} 详情失败: {str(e)}") # 按时间排序 conversations.sort(key=lambda x: x['date']) logger.info(f"总共找到并解析了 {len(conversations)} 封邮件") return conversations except Exception as e: logger.error(f"获取Gmail对话失败: {str(e)}") import traceback logger.error(traceback.format_exc()) return [] def _extract_email_content(self, message): """提取邮件内容，完全按照quickstart.py的get_email_content函数实现""" try: message_id = message['id'] # 获取邮件ID payload = message['payload'] headers = payload['headers'] # 获取邮件基本信息 email_data = { 'id': message_id, # 保存邮件ID 'subject': '', 'from': '', 'date': '', 'body': '', 'attachments': [] # 新增附件列表 } # 提取头部信息 for header in headers: if header['name'] == 'Subject': email_data['subject'] = header['value'] elif header['name'] == 'From': email_data['from'] = header['value'] elif header['name'] == 'Date': date_value = header['value'] logger.info(f"原始邮件日期字符串: '{date_value}'") try: # 打印原始日期字符串信息 import dateutil.parser as date_parser import pytz from django.utils import timezone # 先尝试解析日期 date = date_parser.parse(date_value) logger.info(f"解析后的日期对象: {date}, 是否有时区: {date.tzinfo is not None}") # 处理时区问题 if date.tzinfo is not None: # 已有时区的日期，转换为系统时区 if hasattr(settings, 'TIME_ZONE'): system_tz = pytz.timezone(settings.TIME_ZONE) date = date.astimezone(system_tz) logger.info(f"转换到系统时区后: {date}") # 如果需要naive datetime，删除时区信息 if hasattr(settings, 'USE_TZ') and not settings.USE_TZ: date = date.replace(tzinfo=None) logger.info(f"移除时区信息后: {date}") else: # 无时区的日期，如果系统使用时区，添加时区信息 if hasattr(settings, 'USE_TZ') and settings.USE_TZ: try: date = timezone.make_aware(date) logger.info(f"添加时区信息后: {date}") except Exception as tz_error: logger.warning(f"添加时区信息失败: {str(tz_error)}") # 格式化为字符串 email_data['date'] = date.strftime('%Y-%m-%d %H:%M:%S') logger.info(f"最终日期字符串: {email_data['date']}") except Exception as e: logger.error(f"解析日期失败: {str(e)}, 原始值: '{date_value}'") # 保留原始值 email_data['date'] = date_value # 定义一个递归函数来处理所有部分和附件 def process_parts(parts): for part in parts: # 检查是否是附件 if 'filename' in part and part['filename']: attachment = { 'filename': part['filename'], 'mimeType': part['mimeType'], 'size': part['body'].get('size', 0) } # 如果有附件内容数据，可以获取附件ID if 'attachmentId' in part['body']: attachment['attachmentId'] = part['body']['attachmentId'] email_data['attachments'].append(attachment) # 处理文本内容 if part['mimeType'] == 'text/plain' and not email_data['body']: data = part['body'].get('data', '') if data: try: text = base64.urlsafe_b64decode(data).decode('utf-8') email_data['body'] = text except Exception as e: logger.error(f"解码邮件内容失败: {str(e)}") # 递归处理多部分内容 if 'parts' in part: process_parts(part['parts']) # 处理邮件正文和附件 if 'parts' in payload: process_parts(payload['parts']) elif 'body' in payload and 'data' in payload['body']: # 没有parts，直接处理body data = payload['body'].get('data', '') if data: try: text = base64.urlsafe_b64decode(data).decode('utf-8') email_data['body'] = text except Exception as e: logger.error(f"解码邮件内容失败: {str(e)}") return email_data except Exception as e: logger.error(f"处理邮件内容时出错: {str(e)}") import traceback logger.error(traceback.format_exc()) return None def download_attachment(self, message_id, attachment_id, filename): """下载邮件附件""" try: attachment = self.gmail_service.users().messages().attachments().get( userId='me', messageId=message_id, id=attachment_id ).execute() data = attachment['data'] file_data = base64.urlsafe_b64decode(data) # 创建附件目录 attachments_dir = 'gmail_attachments' if not os.path.exists(attachments_dir): os.makedirs(attachments_dir) # 保存附件 filepath = os.path.join(attachments_dir, f"{message_id}_{filename}") with open(filepath, 'wb') as f: f.write(file_data) return filepath except Exception as e: logger.error(f"下载附件失败: {str(e)}") import traceback logger.error(traceback.format_exc()) return None def save_conversations_to_knowledge_base(self, conversations, knowledge_base): """ 将Gmail对话保存到知识库 Args: conversations: Gmail邮件列表 knowledge_base: 保存到的知识库对象 """ try: # 导入所需模型 from .models import GmailAttachment, ChatHistory, KnowledgeBaseDocument, GmailTalentMapping # 检查入参 if not conversations or not knowledge_base: logger.error("参数不完整: conversations或knowledge_base为空") return {"conversation_id": None, "success": False, "error": "参数不完整"} if not conversations: logger.warning("没有邮件对话可保存") return {"conversation_id": None, "success": False, "error": "没有邮件对话可保存"} # 确保knowledge_base有external_id if not knowledge_base.external_id: logger.warning(f"知识库 {knowledge_base.name} 缺少external_id，尝试创建外部知识库") try: from .views import KnowledgeBaseViewSet kb_viewset = KnowledgeBaseViewSet() external_id = kb_viewset._create_external_dataset(knowledge_base) if external_id: knowledge_base.external_id = external_id knowledge_base.save() logger.info(f"成功为知识库创建外部知识库: {external_id}") else: logger.error("创建外部知识库失败：未返回external_id") return {"conversation_id": None, "success": False, "error": "创建外部知识库失败"} except Exception as e: logger.error(f"为知识库创建外部知识库失败: {str(e)}") return {"conversation_id": None, "success": False, "error": f"创建外部知识库失败: {str(e)}"} # 查找现有的对话ID - 优先使用talent_email查找 conversation_id = None # 1. 首先尝试通过talent_email查找现有映射 if self.email: existing_mapping = GmailTalentMapping.objects.filter( user=self.user, talent_email=self.email, knowledge_base=knowledge_base, is_active=True ).first() if existing_mapping and existing_mapping.conversation_id: conversation_id = existing_mapping.conversation_id logger.info(f"通过达人邮箱 {self.email} 找到现有对话ID: {conversation_id}") # 2. 如果上面没找到，尝试通过知识库查找任何相关的对话 if not conversation_id: existing_conversation = ChatHistory.objects.filter( knowledge_base=knowledge_base, user=self.user, is_deleted=False ).values('conversation_id').distinct().first() if existing_conversation: conversation_id = existing_conversation['conversation_id'] logger.info(f"通过知识库 {knowledge_base.name} 找到现有对话ID: {conversation_id}") # 如果有talent_email，更新或创建映射关系 if self.email: GmailTalentMapping.objects.update_or_create( user=self.user, talent_email=self.email, defaults={ 'knowledge_base': knowledge_base, 'conversation_id': conversation_id, 'is_active': True } ) logger.info(f"更新Gmail达人映射: {self.email} -> 对话ID {conversation_id}") # 3. 如果仍然没找到，则创建新的对话ID if not conversation_id: # 生成唯一的对话ID conversation_id = str(uuid.uuid4()) logger.info(f"创建新的对话ID: {conversation_id}") # 创建或更新Gmail和达人的映射关系 if self.email: talent_mapping, created = GmailTalentMapping.objects.update_or_create( user=self.user, talent_email=self.email, defaults={ 'knowledge_base': knowledge_base, 'conversation_id': conversation_id, 'is_active': True } ) action = "创建" if created else "更新" logger.info(f"已{action}Gmail达人映射: {self.email} -> {knowledge_base.name}") # 构建知识库文档 logger.info(f"开始构建知识库文档，共{len(conversations)}个邮件对话") document = { "title": f"与{self.email or '联系人'}的Gmail邮件对话", "url": "", "source_type": "gmail", "paragraphs": [] } # 对话按时间顺序排序 conversations.sort(key=lambda x: x.get('date', '')) # 将对话添加到文档 previous_message_dict = {} # 用于存储邮件时间与消息ID的映射，以便找到正确的parent_id # 首先查找现有的聊天记录 existing_messages = ChatHistory.objects.filter( conversation_id=conversation_id, is_deleted=False ).order_by('created_at') # 如果已有消息，跳过重复导入 existing_gmail_ids = set() if existing_messages.exists(): for msg in existing_messages: if msg.metadata and 'gmail_message_id' in msg.metadata: existing_gmail_ids.add(msg.metadata['gmail_message_id']) logger.info(f"发现已存在 {len(existing_gmail_ids)} 条消息记录，将跳过重复导入") for message in conversations: try: # 跳过已导入的消息 if message.get('id') in existing_gmail_ids: logger.info(f"跳过已导入的消息: {message.get('id')}") continue # 提取邮件内容 sender = message.get('from', '未知发件人') date_str = message.get('date', '未知时间') subject = message.get('subject', '无主题') body = message.get('body', '').strip() logger.info(f"处理邮件: {date_str}, {sender}, {subject[:20]}...") # 判断角色 # 从发件人地址中提取邮箱部分 sender_email = '' if '<' in sender and '>' in sender: # 格式如 "姓名 " sender_email = sender.split('<')[1].split('>')[0] else: # 格式可能直接是邮箱 sender_email = sender # 修改角色判断逻辑：根据Gmail凭证邮箱和达人邮箱判断 # 获取当前Gmail凭证邮箱 gmail_credential_email = None if self.gmail_credential and self.gmail_credential.gmail_email: gmail_credential_email = self.gmail_credential.gmail_email.lower() # 判断是否是凭证邮箱发出的邮件 is_credential_email = gmail_credential_email and gmail_credential_email == sender_email.lower() # 判断是否是系统用户邮箱 is_user_email = self.user.email.lower() == sender_email.lower() # 判断是否是目标达人邮箱 is_talent_email = False if self.email: is_talent_email = self.email.lower() == sender_email.lower() # 如果是Gmail凭证邮箱或用户邮箱，设为user角色；如果是达人邮箱，设为assistant角色 if is_credential_email or is_user_email: role = 'user' elif is_talent_email: role = 'assistant' else: # 尝试检查是否有与发件人邮箱相匹配的人才映射 from .models import GmailTalentMapping talent_mapping = GmailTalentMapping.objects.filter( user=self.user, talent_email=sender_email, is_active=True ).first() # 如果有映射关系则视为达人邮箱，设为assistant角色 role = 'assistant' if talent_mapping else 'user' logger.info(f"设置消息角色: {role}，发件人: {sender_email}，用户Gmail: {gmail_credential_email}，用户邮箱: {self.user.email}，目标达人: {self.email}，是否有达人映射: {bool(talent_mapping) if 'talent_mapping' in locals() else False}") # 将邮件添加到文档 paragraph = { "id": f"msg_{len(document['paragraphs']) + 1}", "title": f"{date_str} - {sender} - {subject}", "content": body, "meta": { "sender": sender, "date": date_str, "subject": subject, "has_attachments": len(message.get('attachments', [])) > 0, "message_id": message.get('id', '') } } document['paragraphs'].append(paragraph) # 解析邮件日期为datetime对象 try: # 先检查是否是时间戳格式 if isinstance(date_str, str) and date_str.isdigit(): # 如果是时间戳，直接转换 date_obj = datetime.fromtimestamp(int(date_str)) else: # 尝试标准格式解析 try: date_obj = datetime.strptime(date_str, '%Y-%m-%d %H:%M:%S') except ValueError: # 如果标准格式解析失败，使用dateutil更灵活的解析 import dateutil.parser as date_parser date_obj = date_parser.parse(date_str) # 如果解析出的日期有时区信息，转换为不带时区的日期 if date_obj.tzinfo is not None: date_obj = date_obj.replace(tzinfo=None) except (ValueError, TypeError) as e: # 如果解析失败，使用当前时间 logger.warning(f"无法解析邮件日期: {date_str}，错误: {str(e)}，使用当前时间") date_obj = datetime.now() # 查找parent_id（查找日期早于当前邮件且最接近的消息） parent_id = None closest_date = None for d, mid in previous_message_dict.items(): if d < date_obj and (closest_date is None or d > closest_date): closest_date = d parent_id = mid # 保存消息到聊天历史，使用邮件实际日期 from django.utils import timezone # 确保时区处理正确 try: # 检查date_obj是否已经是aware if timezone.is_aware(date_obj): aware_date = date_obj logger.info(f"日期已经包含时区信息: {aware_date}") else: # 将naive转换为aware aware_date = timezone.make_aware(date_obj) logger.info(f"日期添加时区信息后: {aware_date}") except Exception as tz_error: logger.warning(f"时区转换失败: {str(tz_error)}，使用当前时间") aware_date = timezone.now() # 创建消息记录 chat_message = ChatHistory.objects.create( user=self.user, knowledge_base=knowledge_base, conversation_id=conversation_id, parent_id=parent_id, role=role, content=f"{subject}\n\n{body}", metadata={ 'gmail_message_id': message.get('id', ''), 'from': sender, 'date': date_str, 'subject': subject, 'dataset_id_list': [str(knowledge_base.id)], 'dataset_names': [knowledge_base.name] }, created_at=aware_date # 设置正确的创建时间 ) # 更新previous_message_dict previous_message_dict[date_obj] = str(chat_message.id) # 处理附件 attachments = message.get('attachments', []) if attachments: for attachment in attachments: if 'attachmentId' in attachment and 'filename' in attachment: try: # 下载附件 filepath = self.download_attachment( message_id=message['id'], attachment_id=attachment['attachmentId'], filename=attachment['filename'] ) if filepath: # 记录附件信息 GmailAttachment.objects.create( chat_message=chat_message, gmail_message_id=message['id'], filename=attachment['filename'], filepath=filepath, mimetype=attachment.get('mimeType', ''), filesize=attachment.get('size', 0) ) logger.info(f"已保存附件: {attachment['filename']}") except Exception as e: logger.error(f"保存附件失败: {str(e)}") import traceback logger.error(traceback.format_exc()) except Exception as e: logger.error(f"处理邮件时出错: {str(e)}") import traceback logger.error(traceback.format_exc()) # 将文档添加到知识库 logger.info(f"将Gmail对话添加到知识库: {knowledge_base.name}") # 准备文档数据结构 doc_data = { "name": f"Gmail对话与{self.email or '联系人'}.txt", "paragraphs": [] } # 将所有对话转换为段落 for paragraph in document['paragraphs']: doc_data["paragraphs"].append({ "title": paragraph['title'], "content": paragraph['content'], "is_active": True, "problem_list": [] }) # 检查paragraphs是否为空 if not doc_data["paragraphs"]: logger.warning("没有段落内容可上传，添加默认段落") doc_data["paragraphs"].append({ "title": "初始化邮件对话", "content": f"与{self.email or '联系人'}的邮件对话准备就绪，等待新的邮件内容。", "is_active": True, "problem_list": [] }) # 调用知识库的文档上传API from .views import KnowledgeBaseViewSet kb_viewset = KnowledgeBaseViewSet() # 添加随机文档ID doc_data["id"] = str(uuid.uuid4()) logger.info(f"生成随机文档ID: {doc_data['id']}") # 上传文档 upload_response = kb_viewset._call_upload_api(knowledge_base.external_id, doc_data) if upload_response and upload_response.get('code') == 200 and upload_response.get('data'): # 上传成功，保存记录到数据库 document_id = upload_response['data']['id'] # 创建文档记录 from .models import KnowledgeBaseDocument doc_record = KnowledgeBaseDocument.objects.create( knowledge_base=knowledge_base, document_id=document_id, document_name=doc_data["name"], external_id=document_id ) logger.info(f"Gmail对话文档上传成功，ID: {document_id}") # 上传所有附件 self._upload_attachments_to_knowledge_base(knowledge_base, conversations) else: # 上传失败，记录错误信息 error_msg = upload_response.get('message', '未知错误') if upload_response else '上传API调用失败' logger.error(f"Gmail对话文档上传失败: {error_msg}") # 更新知识库文档计数 knowledge_base.document_count = KnowledgeBaseDocument.objects.filter( knowledge_base=knowledge_base, status='active' ).count() knowledge_base.save() logger.info(f"Gmail对话已保存到知识库: {knowledge_base.name}") return { "conversation_id": conversation_id, "success": True, "message_count": len(conversations), "knowledge_base_id": str(knowledge_base.id) } except Exception as e: logger.error(f"保存Gmail对话到知识库失败: {str(e)}") import traceback logger.error(traceback.format_exc()) return {"conversation_id": None, "success": False, "error": str(e)} def send_email(self, to_email, subject, body, conversation_id=None, attachments=None): """发送邮件并保存到聊天记录""" try: # 构建邮件内容 if attachments and len(attachments) > 0: message = self._create_message_with_attachment(to_email, subject, body, attachments) else: message = self._create_message(to_email, subject, body) # 发送邮件 sent_message = self.gmail_service.users().messages().send( userId='me', body=message ).execute() message_id = sent_message['id'] logger.info(f"邮件发送成功, ID: {message_id}") # 获取邮件详情，包括服务器时间戳 try: message_detail = self.gmail_service.users().messages().get( userId='me', id=message_id ).execute() # 从消息详情中提取时间戳 internal_date = message_detail.get('internalDate') if internal_date: try: # 转换毫秒时间戳为datetime email_date = datetime.fromtimestamp(int(internal_date)/1000) logger.info(f"从时间戳解析的日期: {email_date}") # 处理时区 from django.utils import timezone if hasattr(settings, 'USE_TZ') and settings.USE_TZ and not timezone.is_aware(email_date): email_date = timezone.make_aware(email_date) logger.info(f"添加时区信息后: {email_date}") date_str = email_date.strftime('%Y-%m-%d %H:%M:%S') logger.info(f"最终格式化日期: {date_str}") except Exception as tz_error: logger.warning(f"处理邮件时间戳失败: {str(tz_error)}，使用当前时间") from django.utils import timezone email_date = timezone.now() date_str = email_date.strftime('%Y-%m-%d %H:%M:%S') else: logger.warning("邮件没有时间戳，使用当前时间") from django.utils import timezone email_date = timezone.now() date_str = email_date.strftime('%Y-%m-%d %H:%M:%S') except Exception as e: logger.warning(f"获取邮件时间戳失败: {str(e)}, 使用当前时间") email_date = datetime.now() date_str = email_date.strftime('%Y-%m-%d %H:%M:%S') # 如果有conversation_id，保存到聊天记录 if conversation_id: # 查找现有的聊天记录 from django.db.models import Q existing_messages = ChatHistory.objects.filter( conversation_id=conversation_id, is_deleted=False ).order_by('created_at') if not existing_messages.exists(): logger.warning(f"找不到对话ID: {conversation_id}，无法保存消息") return message_id # 查找关联的知识库 first_message = existing_messages.first() if first_message.knowledge_base: knowledge_base = first_message.knowledge_base else: # 如果第一条消息没有知识库，尝试从metadata获取 if first_message.metadata and 'dataset_id_list' in first_message.metadata: kb_id = first_message.metadata.get('dataset_id_list', [])[0] knowledge_base = KnowledgeBase.objects.get(id=kb_id) else: logger.error(f"找不到关联的知识库，无法保存消息") return message_id # 查找parent_id：时间早于当前邮件且最接近的消息，避免时区问题 parent_id = None try: # 使用简单查询，不依赖时区 previous_messages = ChatHistory.objects.filter( conversation_id=conversation_id, is_deleted=False ).order_by('-id')[:1] # 使用ID降序排序而非时间 if previous_messages: parent_id = str(previous_messages[0].id) except Exception as e: logger.error(f"查找父消息失败: {str(e)}") logger.error(traceback.format_exc()) # 构建metadata metadata = { 'gmail_message_id': message_id, 'from': self.user.email, 'to': to_email, 'date': date_str, 'subject': subject, 'dataset_id_list': [str(knowledge_base.id)], 'dataset_names': [knowledge_base.name] } # 如果现有消息有dataset_id_list，保留它 if first_message.metadata and 'dataset_id_list' in first_message.metadata: metadata['dataset_id_list'] = first_message.metadata['dataset_id_list'] # 尝试获取对应的dataset_names if 'dataset_names' in first_message.metadata: metadata['dataset_names'] = first_message.metadata['dataset_names'] # 创建聊天记录，使用当前时间而不是邮件的实际时间 chat_message = ChatHistory.objects.create( user=self.user, knowledge_base=knowledge_base, conversation_id=conversation_id, parent_id=parent_id, role='user', # 用户发送的消息，角色固定为'user' content=f"[{subject}] {body}", metadata=metadata # 不设置created_at，使用数据库默认时间 ) # 如果有附件，保存附件信息 if attachments and len(attachments) > 0: for att_path in attachments: file_name = os.path.basename(att_path) file_size = os.path.getsize(att_path) # 获取MIME类型 mime_type, _ = mimetypes.guess_type(att_path) if not mime_type: mime_type = 'application/octet-stream' # 保存附件信息到数据库 gmail_attachment = GmailAttachment.objects.create( chat_message=chat_message, gmail_message_id=message_id, filename=file_name, filepath=att_path, mimetype=mime_type, filesize=file_size ) # 更新知识库文档 self._append_to_knowledge_base_document(knowledge_base, subject, body, to_email) return message_id except Exception as e: logger.error(f"发送邮件失败: {str(e)}") logger.error(traceback.format_exc()) raise def _create_message(self, to, subject, body): """创建邮件消息""" message = MIMEText(body) message['to'] = to message['subject'] = subject # 编码为JSON安全的字符串 raw = base64.urlsafe_b64encode(message.as_bytes()).decode() return {'raw': raw} def _create_message_with_attachment(self, to, subject, body, attachment_files): """创建带附件的邮件消息""" message = MIMEMultipart() message['to'] = to message['subject'] = subject # 添加邮件正文 msg = MIMEText(body) message.attach(msg) # 添加附件 for file in attachment_files: try: with open(file, 'rb') as f: part = MIMEBase('application', 'octet-stream') part.set_payload(f.read()) # 编码附件内容 encoders.encode_base64(part) # 添加头部信息 filename = os.path.basename(file) part.add_header( 'Content-Disposition', f'attachment; filename="{filename}"' ) message.attach(part) except Exception as e: logger.error(f"添加附件 {file} 失败: {str(e)}") # 编码为JSON安全的字符串 raw = base64.urlsafe_b64encode(message.as_bytes()).decode() return {'raw': raw} def _append_to_knowledge_base_document(self, knowledge_base, subject, body, recipient): """将新发送的邮件内容追加到知识库文档中""" try: # 准备文档数据结构 doc_data = { "name": f"Gmail回复给{recipient}_{datetime.now().strftime('%Y%m%d%H%M%S')}.txt", "paragraphs": [ { "title": f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - {subject}", "content": f"发件人: {self.user.email}\n收件人: {recipient}\n\n{body}", "is_active": True, "problem_list": [] } ] } # 调用知识库的文档上传API from .views import KnowledgeBaseViewSet kb_viewset = KnowledgeBaseViewSet() # 验证知识库是否有external_id if not knowledge_base.external_id: logger.error(f"知识库没有external_id，无法上传文档: {knowledge_base.name}") return None # 上传文档 upload_response = kb_viewset._call_upload_api(knowledge_base.external_id, doc_data) if upload_response and upload_response.get('code') == 200 and upload_response.get('data'): # 上传成功，保存记录到数据库 document_id = upload_response['data']['id'] # 创建文档记录 from .models import KnowledgeBaseDocument doc_record = KnowledgeBaseDocument.objects.create( knowledge_base=knowledge_base, document_id=document_id, document_name=doc_data["name"], external_id=document_id ) logger.info(f"Gmail回复文档上传成功，ID: {document_id}") # 更新知识库文档计数 knowledge_base.document_count = KnowledgeBaseDocument.objects.filter( knowledge_base=knowledge_base, status='active' ).count() knowledge_base.save() return upload_response else: # 上传失败，记录错误信息 error_msg = upload_response.get('message', '未知错误') if upload_response else '上传API调用失败' logger.error(f"Gmail回复文档上传失败: {error_msg}") return None except Exception as e: logger.error(f"追加邮件内容到知识库失败: {str(e)}") import traceback logger.error(traceback.format_exc()) return None def setup_watch(self, topic_name=None): """设置Gmail监听""" try: # 如果没有指定topic，使用配置的主题名称 if not topic_name: # 从settings获取项目配置的主题名称 topic_name = getattr(settings, 'GMAIL_TOPIC_NAME', 'gmail-watch-topic') # 直接使用settings中配置的项目ID，不再从client_secret.json获取 project_id = getattr(settings, 'GOOGLE_CLOUD_PROJECT', 'knowledge-454905') logger.info(f"使用settings中配置的项目ID: {project_id}") # 确保Gmail服务已经初始化 if not hasattr(self, 'gmail_service') or not self.gmail_service: logger.warning("Gmail服务未初始化，尝试重新认证") if not self.authenticate(): logger.error("Gmail服务初始化失败") return None # 注意：不再使用用户邮箱作为判断依据，Gmail API总是使用'me'作为userId # Gmail认证是通过OAuth流程，与系统用户邮箱无关 logger.info(f"系统用户: {self.user.email}，Gmail API使用OAuth认证的邮箱 (userId='me')") # 构建完整的webhook URL webhook_url = getattr(settings, 'GMAIL_WEBHOOK_URL', None) if not webhook_url: # 使用默认值，确保这是一个完全限定的URL domain = getattr(settings, 'ALLOWED_HOSTS', ['localhost'])[0] protocol = 'https' if getattr(settings, 'SECURE_SSL_REDIRECT', False) else 'http' if domain == 'localhost' or domain.startswith('127.0.0.1'): # 本地开发环境需要公网可访问的URL，可以用ngrok等工具暴露本地服务 webhook_url = f"{protocol}://{domain}:8000/api/user/gmail/webhook/" logger.warning("使用本地开发环境URL作为webhook回调，Google可能无法访问。建议使用ngrok等工具创建公网地址。") else: webhook_url = f"{protocol}://{domain}/api/user/gmail/webhook/" logger.info(f"使用Gmail webhook URL: {webhook_url}") # 如果想在Google Cloud控制台中手动配置webhook，打印明确的说明 logger.info("如需手动配置Gmail推送通知，请确保在Google Cloud控制台的Pub/Sub主题配置中设置以下URL:") logger.info(f"推送端点: {webhook_url}") logger.info("权限: 为Gmail API服务账号(gmail-api-push@system.gserviceaccount.com)和您的Gmail账号授予发布权限") # 请求监听 request = { 'labelIds': ['INBOX', 'SENT', 'IMPORTANT', 'UNREAD'], # 监听更多标签 'topicName': f"projects/{project_id}/topics/{topic_name}", 'labelFilterAction': 'include' } logger.info(f"设置Gmail监听: {request}") # 设置超时，避免长时间阻塞 original_timeout = socket.getdefaulttimeout() socket.setdefaulttimeout(GMAIL_REQUEST_TIMEOUT) try: # 执行watch请求 response = self.gmail_service.users().watch(userId='me', body=request).execute() except Exception as e: logger.error(f"执行watch请求失败: {str(e)}") return None finally: # 恢复原始超时设置 socket.setdefaulttimeout(original_timeout) # 获取historyId，用于后续同步 history_id = response.get('historyId') expiration = response.get('expiration') logger.info(f"Gmail监听设置成功: historyId={history_id}, expiration={expiration}") # 保存监听信息到数据库 if self.user: from .models import GmailCredential credential = GmailCredential.objects.filter(user=self.user, is_active=True).first() if credential: # 转换时间戳为datetime expiration_time = None if expiration: try: # 将毫秒时间戳转换为datetime naive_time = datetime.fromtimestamp(int(expiration)/1000) logger.info(f"从时间戳解析的日期: {naive_time}") # 如果系统使用时区，添加时区信息 if hasattr(settings, 'USE_TZ') and settings.USE_TZ: try: expiration_time = timezone.make_aware(naive_time) logger.info(f"添加时区信息后: {expiration_time}") except Exception as tz_error: logger.warning(f"添加时区信息失败: {str(tz_error)}") expiration_time = naive_time else: expiration_time = naive_time except Exception as conv_error: logger.error(f"转换时间戳失败: {str(conv_error)}") expiration_time = None credential.last_history_id = history_id credential.watch_expiration = expiration_time credential.save() logger.info(f"更新Gmail监听信息: {self.user.username}, history_id: {history_id}") return { 'historyId': history_id, 'expiration': expiration } except Exception as e: logger.error(f"设置Gmail监听失败: {str(e)}") logger.error(traceback.format_exc()) return None def get_history(self, start_history_id): """获取历史变更""" try: logger.info(f"获取历史记录，起始ID: {start_history_id}") # 确保Gmail服务已经初始化 if not hasattr(self, 'gmail_service') or not self.gmail_service: logger.warning("Gmail服务未初始化，尝试重新认证") if not self.authenticate(): logger.error("Gmail服务初始化失败") return [] # 设置超时，避免长时间阻塞 original_timeout = socket.getdefaulttimeout() socket.setdefaulttimeout(GMAIL_REQUEST_TIMEOUT) try: # 执行history请求 response = self.gmail_service.users().history().list( userId='me', startHistoryId=start_history_id ).execute() except Exception as e: logger.error(f"执行history请求失败: {str(e)}") return [] finally: # 恢复原始超时设置 socket.setdefaulttimeout(original_timeout) logger.info(f"历史记录响应: {response}") history_list = [] if 'history' in response: history_list.extend(response['history']) logger.info(f"找到 {len(response['history'])} 个历史记录") # 获取所有页，但每次请求都设置超时 page_count = 0 max_pages = 5 # 限制最大页数，避免无限循环 while 'nextPageToken' in response and page_count < max_pages: page_token = response['nextPageToken'] page_count += 1 # 设置超时 socket.setdefaulttimeout(GMAIL_REQUEST_TIMEOUT) try: response = self.gmail_service.users().history().list( userId='me', startHistoryId=start_history_id, pageToken=page_token ).execute() except Exception as e: logger.error(f"获取历史记录分页失败: {str(e)}") break finally: # 恢复原始超时设置 socket.setdefaulttimeout(original_timeout) if 'history' in response: history_list.extend(response['history']) logger.info(f"加载额外 {len(response['history'])} 个历史记录 (页 {page_count})") if page_count >= max_pages and 'nextPageToken' in response: logger.warning(f"达到最大页数限制 ({max_pages})，可能有更多历史记录未获取") else: logger.info(f"没有新的历史记录，最新historyId: {response.get('historyId', 'N/A')}") # 提取新消息ID new_message_ids = set() for history in history_list: logger.info(f"处理历史记录: {history}") if 'messagesAdded' in history: for message in history['messagesAdded']: message_id = message['message']['id'] new_message_ids.add(message_id) logger.info(f"新增消息ID: {message_id}") if 'labelsAdded' in history: for label in history['labelsAdded']: message_id = label['message']['id'] if 'INBOX' in label.get('labelIds', []): new_message_ids.add(message_id) logger.info(f"标签变更的消息ID: {message_id}, 添加了INBOX标签") if new_message_ids: logger.info(f"总共找到 {len(new_message_ids)} 个新消息") else: logger.info("没有新增消息") return list(new_message_ids) except Exception as e: logger.error(f"获取Gmail历史记录失败: {str(e)}") logger.error(traceback.format_exc()) return [] def process_notification(self, notification_data): """处理Gmail推送通知""" try: # 提取通知数据 logger.info(f"处理Gmail通知: {notification_data}") # 处理Google Pub/Sub消息格式 if isinstance(notification_data, dict) and 'message' in notification_data and 'data' in notification_data['message']: try: import base64 import json logger.info("检测到Google Pub/Sub消息格式") # Base64解码data字段 encoded_data = notification_data['message']['data'] decoded_data = base64.b64decode(encoded_data).decode('utf-8') logger.info(f"解码后的数据: {decoded_data}") # 解析JSON获取email和historyId json_data = json.loads(decoded_data) email = json_data.get('emailAddress') history_id = json_data.get('historyId') logger.info(f"从Pub/Sub消息中提取: email={email}, historyId={history_id}") except Exception as decode_error: logger.error(f"解析Pub/Sub消息失败: {str(decode_error)}") logger.error(traceback.format_exc()) return False else: # 原始格式处理 email = notification_data.get('emailAddress') history_id = notification_data.get('historyId') if not email or not history_id: logger.error("Gmail通知数据不完整: 找不到emailAddress或historyId") return False # 查找关联用户 from .models import User, GmailCredential user = User.objects.filter(email=email).first() credential = None # 如果找不到用户，尝试使用gmail_email字段查找 if not user: logger.info(f"找不到email={email}的用户，尝试使用gmail_email查找") credential = GmailCredential.objects.filter(gmail_email=email, is_active=True).first() if credential: user = credential.user logger.info(f"通过gmail_email找到用户: {user.email}") if not user: logger.error(f"找不到与 {email} 关联的用户") return False # 确保有相应的Gmail凭证 if not credential: credential = GmailCredential.objects.filter(user=user, is_active=True).first() if not credential: logger.error(f"用户 {user.email} 没有活跃的Gmail凭证") return False # 检查凭证是否需要重新授权 if credential.needs_reauth: logger.warning(f"Gmail凭证 {credential.id} 需要重新授权，无法处理通知") # 记录当前通知到队列或数据库，供用户重新授权后再处理 try: from .models import GmailNotificationQueue # 存储通知到队列 GmailNotificationQueue.objects.create( user=user, gmail_credential=credential, email=email, history_id=history_id, notification_data=json.dumps(notification_data) ) logger.info(f"通知已保存到队列，等待用户重新授权后处理") except Exception as queue_error: logger.error(f"保存通知到队列失败: {str(queue_error)}") return False # 更新历史ID if credential and history_id: try: # 仅当新的历史ID大于当前值时更新 if not credential.last_history_id or int(history_id) > int(credential.last_history_id): credential.last_history_id = history_id credential.save() logger.info(f"更新历史ID: {history_id}") else: logger.info(f"收到的历史ID ({history_id}) 不大于当前值 ({credential.last_history_id})，不更新") except Exception as update_error: logger.error(f"更新历史ID失败: {str(update_error)}") # 初始化Gmail集成 gmail_integration = GmailIntegration(user, gmail_credential_id=credential.id if credential else None) # 记录详细的处理信息 logger.info(f"Gmail通知处理: 用户={user.email}, Gmail邮箱={email}, 历史ID={history_id}, 凭证ID={credential.id}") # 设置超时 original_timeout = socket.getdefaulttimeout() socket.setdefaulttimeout(GMAIL_REQUEST_TIMEOUT) try: # 认证Gmail服务 if not gmail_integration.authenticate(): logger.error(f"Gmail认证失败: {email}") # 尝试刷新令牌 refresh_result = gmail_integration.refresh_token() if not refresh_result: logger.error("刷新令牌失败，需要用户重新授权") return False logger.info("令牌刷新成功，继续处理通知") # 获取历史变更 message_ids = gmail_integration.get_history(history_id) # 如果没有找到新消息，尝试获取最近的消息 if not message_ids: logger.info("没有找到历史变更，尝试获取最近的邮件") # 查询最近的5封邮件 try: recent_messages = gmail_integration.gmail_service.users().messages().list( userId='me', maxResults=5 ).execute() if 'messages' in recent_messages: message_ids = [msg['id'] for msg in recent_messages['messages']] logger.info(f"获取到 {len(message_ids)} 封最近的邮件: {message_ids}") else: logger.info("没有找到最近的邮件") except Exception as recent_error: logger.error(f"获取最近邮件失败: {str(recent_error)}") except Exception as e: error_msg = str(e) logger.error(f"处理Gmail通知时发生错误: {error_msg}") # 检查是否是令牌过期 if "invalid_grant" in error_msg.lower() or "401" in error_msg: logger.warning("检测到OAuth令牌问题，尝试刷新令牌") # 尝试刷新令牌 refresh_result = gmail_integration.refresh_token() if not refresh_result: logger.error("刷新令牌失败，需要用户重新授权") return False # 再次尝试获取历史记录 try: message_ids = gmail_integration.get_history(history_id) # 如果仍然没有找到新消息，尝试获取最近的消息 if not message_ids: logger.info("刷新令牌后仍未找到历史变更，尝试获取最近邮件") recent_messages = gmail_integration.gmail_service.users().messages().list( userId='me', maxResults=5 ).execute() if 'messages' in recent_messages: message_ids = [msg['id'] for msg in recent_messages['messages']] logger.info(f"获取到 {len(message_ids)} 封最近的邮件: {message_ids}") except Exception as retry_error: logger.error(f"刷新令牌后再次尝试失败: {str(retry_error)}") return False else: return False finally: # 恢复原始超时设置 socket.setdefaulttimeout(original_timeout) if message_ids: logger.info(f"找到 {len(message_ids)} 个需要处理的消息") # 限制处理的消息数量，防止过多消息导致系统负载过高 max_messages = 10 if len(message_ids) > max_messages: logger.warning(f"消息数量 ({len(message_ids)}) 超过限制 ({max_messages})，将只处理前 {max_messages} 条") message_ids = message_ids[:max_messages] # 处理消息 success_count = 0 for message_id in message_ids: try: # 设置超时 socket.setdefaulttimeout(GMAIL_REQUEST_TIMEOUT) if self._process_new_message(gmail_integration, message_id): success_count += 1 except Exception as msg_error: error_msg = str(msg_error) logger.error(f"处理消息 {message_id} 失败: {error_msg}") # 检查是否是令牌过期 if "invalid_grant" in error_msg.lower() or "401" in error_msg: logger.warning("处理消息时检测到OAuth令牌问题，标记需要重新授权") gmail_integration._mark_credential_needs_reauth() break logger.error(traceback.format_exc()) finally: # 恢复原始超时设置 socket.setdefaulttimeout(original_timeout) logger.info(f"成功处理 {success_count}/{len(message_ids)} 个消息") return success_count > 0 # 如果没有找到新消息但仍需确认处理，验证GooglePubSub连接并保持通知持续 try: # 简单验证连接性测试 logger.info(f"未找到需要处理的消息，执行服务连接测试") profile = gmail_integration.gmail_service.users().getProfile(userId='me').execute() if profile and 'emailAddress' in profile: logger.info(f"Gmail服务连接正常: {profile['emailAddress']}") # 检查并更新监听状态 if credential: # 检查监听是否过期 needs_watch_renew = False if credential.watch_expiration: from django.utils import timezone # 如果监听将在3天内过期，提前更新 three_days_later = timezone.now() + timezone.timedelta(days=3) if credential.watch_expiration < three_days_later: needs_watch_renew = True logger.info(f"监听将在3天内过期，需要更新: {credential.watch_expiration}") else: needs_watch_renew = True logger.info("没有监听过期时间记录，需要更新") if needs_watch_renew: try: # 更新监听 watch_result = gmail_integration.setup_watch() logger.info(f"更新监听成功: {watch_result}") return True except Exception as watch_error: logger.error(f"更新监听失败: {str(watch_error)}") return False return True except Exception as verify_error: logger.error(f"Gmail服务连接测试失败: {str(verify_error)}") return False # 如果没有找到新消息，记录日志并返回成功 logger.info("没有找到新消息，处理完成") return True except Exception as e: logger.error(f"处理Gmail通知失败: {str(e)}") logger.error(traceback.format_exc()) return False def _process_new_message(self, gmail_integration, message_id): """处理新收到的邮件""" try: # 导入所需模型 from .models import GmailTalentMapping, GmailAttachment, KnowledgeBase, ChatHistory, UserProfile # 添加详细日志 logger.info(f"处理新消息ID: {message_id}") # 获取消息内容 message = gmail_integration.gmail_service.users().messages().get(userId='me', id=message_id).execute() # 从邮件中提取相关信息 email_content = gmail_integration._extract_email_content(message) if not email_content: logger.error(f"无法提取消息 {message_id} 的内容") return False # 记录邮件详情 sender = email_content.get('from', '') recipient = email_content.get('to', '') subject = email_content.get('subject', '') body = email_content.get('body', '') logger.info(f"提取的邮件信息: 发件人={sender}, 收件人={recipient}, 主题={subject}") # 提取达人邮箱 - 可能是发件人或收件人 talent_email = None is_talent_sending = False # 标记是否是达人发送的邮件 # 检查收件人是否是当前用户 user_email = None if self.user and hasattr(self.user, 'email'): user_email = self.user.email # 如果发件人不是用户，则认为发件人是达人 if user_email and sender and user_email.lower() not in sender.lower(): talent_email = sender.lower() is_talent_sending = True logger.info(f"检测到达人(发件人)邮箱: {talent_email}") # 如果收件人不是用户，则认为收件人是达人 elif user_email and recipient and user_email.lower() not in recipient.lower(): talent_email = recipient.lower() is_talent_sending = False logger.info(f"检测到达人(收件人)邮箱: {talent_email}") # 从邮箱中提取纯地址（去除名称部分） if talent_email: if '<' in talent_email and '>' in talent_email: talent_email = talent_email.split('<')[1].split('>')[0] # 转换为小写，提高匹配准确性 talent_email = talent_email.lower() # 如果找不到明确的达人邮箱，尝试从映射中查找 if not talent_email: # 尝试从主题或正文中找线索 # 实现取决于具体需求 pass logger.info(f"最终确定的达人邮箱: {talent_email}") # 如果找到了达人邮箱，进行知识库处理 if talent_email: # 创建或获取知识库 knowledge_base, created = self.create_talent_knowledge_base(talent_email) if not knowledge_base: logger.error(f"无法为达人 {talent_email} 创建知识库") return False logger.info(f"使用知识库: {knowledge_base.name} (ID: {knowledge_base.id}), 新创建: {created}") # 获取映射关系 talent_mapping, mapping_created = GmailTalentMapping.objects.get_or_create( user=self.user, talent_email=talent_email, defaults={ 'knowledge_base': knowledge_base, 'conversation_id': f"gmail_{talent_email.replace('@', '_').replace('.', '_')}", 'is_active': True } ) # 获取对话ID conversation_id = talent_mapping.conversation_id # 创建聊天记录 - 确定正确的角色 # 修正角色判断逻辑：达人始终是assistant，用户始终是user # 如果是达人发送的邮件，角色应该是assistant；如果是用户发送的邮件，角色应该是user chat_role = 'assistant' if is_talent_sending else 'user' # 添加更详细的角色日志 logger.info(f"设置聊天角色: 角色={chat_role}, 是否达人发送={is_talent_sending}, 发件人={sender}, 达人邮箱={talent_email}") # 创建或更新聊天记录 chat_entry = ChatHistory.objects.create( user=self.user, knowledge_base=knowledge_base, conversation_id=conversation_id, role=chat_role, content=f"{subject}\n\n{body}", parent_id=message_id ) logger.info(f"已创建聊天记录: ID={chat_entry.id}, 角色={chat_role}, 对话ID={conversation_id}") # 查找适合的parent_id：获取对话中最后一条消息 try: # 获取该对话中非自己的最新消息作为正确的父消息 last_message = ChatHistory.objects.filter( conversation_id=conversation_id, is_deleted=False ).exclude( id=chat_entry.id # 排除刚刚创建的消息 ).order_by('-created_at').first() if last_message: # 更新parent_id为对话中的上一条消息ID chat_entry.parent_id = str(last_message.id) chat_entry.save(update_fields=['parent_id']) logger.info(f"更新消息 {chat_entry.id} 的parent_id为 {last_message.id}") except Exception as parent_error: logger.error(f"更新父消息ID失败: {str(parent_error)}") # 处理附件 if 'attachments' in email_content and email_content['attachments']: # 下载并处理附件 for attachment in email_content['attachments']: try: logger.info(f"处理附件: {attachment.get('filename')}") # 下载附件 filepath = self.download_attachment( message_id=message_id, attachment_id=attachment.get('attachmentId'), filename=attachment.get('filename') ) if filepath: # 创建附件记录 GmailAttachment.objects.create( chat_message=chat_entry, gmail_message_id=message_id, filename=attachment.get('filename'), filepath=filepath, mimetype=attachment.get('mimeType', ''), filesize=attachment.get('size', 0) ) logger.info(f"已保存附件: {filepath}") except Exception as attachment_error: logger.error(f"处理附件 {attachment.get('filename')} 失败: {str(attachment_error)}") # 检查是否需要自动回复 try: # 只有达人发送的邮件才考虑自动回复 if is_talent_sending and chat_role == 'user': # 检查用户是否启用了自动回复 profile = UserProfile.objects.filter(user=self.user).first() if profile and profile.auto_recommend_reply: logger.info(f"用户 {self.user.email} 已启用自动回复，生成回复建议") # 获取该对话的历史记录 conversation_history = ChatHistory.objects.filter( conversation_id=conversation_id ).order_by('created_at') # 生成回复 recommended_reply = self._get_recommended_reply_from_deepseek(conversation_history) if recommended_reply: logger.info(f"生成了推荐回复，长度: {len(recommended_reply)}") # 保存推荐回复 recommended_reply_entry = ChatHistory.objects.create( user=self.user, knowledge_base=knowledge_base, conversation_id=conversation_id, role='user', content=recommended_reply, parent_id=chat_entry.id ) logger.info(f"已保存推荐回复到历史记录, 角色=user, ID={recommended_reply_entry.id}, 父消息ID={chat_entry.id}") except Exception as auto_reply_error: logger.error(f"生成自动回复失败: {str(auto_reply_error)}") # 记录详细的处理结果日志 logger.info(f"成功处理达人邮件: ID={message_id}, 发件人={sender}, 收件人={recipient}, 主题={subject}") logger.info(f"保存消息信息: 角色={chat_role}, 对话ID={conversation_id}, 是否达人发送={is_talent_sending}") return True else: logger.warning(f"无法确定达人邮箱，跳过知识库处理") return False except Exception as e: logger.error(f"处理新消息 {message_id} 失败: {str(e)}") logger.error(traceback.format_exc()) return False def _get_recommended_reply_from_deepseek(self, conversation_history): """调用DeepSeek API生成回复建议""" try: # 使用有效的API密钥 api_key = "" # 尝试从环境变量获取 import os from dotenv import load_dotenv load_dotenv() env_api_key = os.environ.get('DEEPSEEK_API_KEY') if env_api_key: api_key = env_api_key # 从Django设置中获取密钥 from django.conf import settings if hasattr(settings, 'DEEPSEEK_API_KEY') and settings.DEEPSEEK_API_KEY: api_key = settings.DEEPSEEK_API_KEY # 如果仍然没有有效的API密钥，使用默认值 if not api_key: api_key = "sk-xqbujijjqqmlmlvkhvxeogqjtzslnhdtqxqgiyuhwpoqcjvf" logger.warning("使用默认API密钥，请在环境变量或settings.py中设置DEEPSEEK_API_KEY") url = "https://api.siliconflow.cn/v1/chat/completions" # 获取用户总目标 from .models import UserGoal, ConversationSummary user_goal = UserGoal.objects.filter(user=self.user, is_active=True).first() goal_content = user_goal.content if user_goal else None # 尝试获取对话总结 talent_email = None conversation_id = None # 从对话历史中尝试提取达人邮箱 for message in conversation_history: if message.get('role') == 'user' and 'metadata' in message and 'from_email' in message['metadata']: talent_email = message['metadata']['from_email'] break # 从对话历史中尝试提取对话ID for message in conversation_history: if 'metadata' in message and 'conversation_id' in message['metadata']: conversation_id = message['metadata']['conversation_id'] break # 获取对话总结 conversation_summary = None if talent_email and conversation_id: summary_obj = ConversationSummary.objects.filter( user=self.user, talent_email=talent_email, conversation_id=conversation_id, is_active=True ).first() if summary_obj: conversation_summary = summary_obj.summary # 构建系统提示，结合用户总目标和对话总结 system_content = "你是一位专业的电商客服和达人助手。你的任务是针对用户最近的消息生成一个有帮助、礼貌且详细的回复。" if goal_content: system_content += f"\n\n【用户总目标】\n{goal_content}\n\n" if conversation_summary: system_content += f"【对话总结】\n{conversation_summary}\n\n" system_content += "请针对达人最后一条消息，结合用户总目标和对话总结生成一个有针对性的专业回复。回复必须对达人当前问题直接响应，同时与用户的总体合作目标保持一致。回复应该有至少100个字符，必须提供有实质内容的回复。" system_message = { "role": "system", "content": system_content } messages = [system_message] # 限制对话历史长度，只保留最近的5条消息，避免超出token限制 recent_messages = conversation_history[-5:] if len(conversation_history) > 5 else conversation_history messages.extend(recent_messages) # 确保最后一条消息是用户消息，如果不是，添加一个提示 if not recent_messages or recent_messages[-1]['role'] != 'user': # 添加一个系统消息作为用户的最后一条消息 messages.append({ "role": "user", "content": "请针对达人最后一条消息生成专业的回复，考虑我们之前的对话历史和我设定的总目标。" }) # 完全按照文档提供的参数格式构建请求 payload = { "model": "deepseek-ai/DeepSeek-V3", "messages": messages, "stream": False, "max_tokens": 1024, # 增加token上限 "temperature": 0.7, # 提高多样性 "top_p": 0.9, "top_k": 50, "frequency_penalty": 0.5, "presence_penalty": 0.2, # 添加新参数 "n": 1, "stop": [], "response_format": { "type": "text" } } headers = { "Content-Type": "application/json", "Authorization": f"Bearer {api_key}" } logger.info(f"开始调用DeepSeek API生成推荐回复") response = requests.post(url, json=payload, headers=headers) if response.status_code != 200: logger.error(f"DeepSeek API调用失败: {response.status_code}, {response.text}") return None result = response.json() logger.debug(f"DeepSeek API返回: {result}") # 提取回复内容 if 'choices' in result and len(result['choices']) > 0: reply = result['choices'][0]['message']['content'] # 如果返回的内容为空，直接返回None if not reply or reply.strip() == '': logger.warning("DeepSeek API返回的回复内容为空") return None return reply logger.warning(f"DeepSeek API返回格式异常: {result}") return None except Exception as e: logger.error(f"调用DeepSeek API失败: {str(e)}") logger.error(traceback.format_exc()) return None def get_attachment_by_conversation(self, conversation_id): """获取特定对话的所有附件""" try: # 查找对话记录 records = ChatHistory.objects.filter( conversation_id=conversation_id, user=self.user, is_deleted=False ) if not records: logger.warning(f"找不到对话记录: {conversation_id}") return [] # 使用GmailAttachment数据模型查询附件 attachments = [] records_ids = [record.id for record in records] gmail_attachments = GmailAttachment.objects.filter( chat_message_id__in=records_ids ).select_related('chat_message') for attachment in gmail_attachments: chat_message = attachment.chat_message attachments.append({ 'id': str(attachment.id), 'filename': attachment.filename, 'filepath': attachment.filepath, 'mimetype': attachment.mimetype, 'filesize': attachment.filesize, 'message_id': str(chat_message.id), 'gmail_message_id': attachment.gmail_message_id, 'role': chat_message.role, 'created_at': attachment.created_at.strftime('%Y-%m-%d %H:%M:%S'), 'content_preview': chat_message.content[:100] + '...' if len(chat_message.content) > 100 else chat_message.content }) return attachments except Exception as e: logger.error(f"获取对话附件失败: {str(e)}") return [] def handle_auth_code(self, auth_code): """处理OAuth2回调授权码""" try: flow = self.get_oauth_flow() flow.fetch_token(code=auth_code) self.credentials = flow.credentials # 初始化Gmail服务 self.gmail_service = discovery.build('gmail', 'v1', http=self.credentials.authorize(Http())) # 获取Gmail账号 gmail_email = None try: # 调用API获取用户资料 profile = self.gmail_service.users().getProfile(userId='me').execute() gmail_email = profile.get('emailAddress') logger.info(f"获取到Gmail账号: {gmail_email}") except Exception as e: logger.error(f"获取Gmail账号失败: {str(e)}") # 保存凭证到数据库 if self.user: from django.utils import timezone # 将凭证对象序列化 credentials_data = pickle.dumps(self.credentials) # 如果提供了具体的gmail_credential_id，更新对应的记录 if self.gmail_credential_id: try: gmail_credential = GmailCredential.objects.get( id=self.gmail_credential_id, user=self.user ) # 更新凭证信息 gmail_credential.credentials = credentials_data gmail_credential.gmail_email = gmail_email gmail_credential.token_path = self.token_storage_path gmail_credential.updated_at = timezone.now() gmail_credential.is_active = True gmail_credential.save() self.gmail_credential = gmail_credential logger.info(f"已更新ID为 {self.gmail_credential_id} 的Gmail凭证，Gmail账号: {gmail_email}") except GmailCredential.DoesNotExist: logger.error(f"未找到ID为 {self.gmail_credential_id} 的Gmail凭证，将创建新凭证") self.gmail_credential_id = None # 如果没有具体的gmail_credential_id，或者指定的不存在，创建或更新默认凭证 if not self.gmail_credential_id: # 检查是否已存在相同gmail_email的凭证 existing_credential = None if gmail_email: existing_credential = GmailCredential.objects.filter( user=self.user, gmail_email=gmail_email, is_active=True ).first() if existing_credential: # 更新现有凭证 existing_credential.credentials = credentials_data existing_credential.token_path = self.token_storage_path existing_credential.updated_at = timezone.now() existing_credential.save() self.gmail_credential = existing_credential logger.info(f"已更新Gmail账号 {gmail_email} 的现有凭证") else: # 获取Gmail账户数量 gmail_count = GmailCredential.objects.filter(user=self.user).count() # 创建新凭证 name = f"Gmail账号 {gmail_count + 1}" if gmail_email: name = gmail_email else: # 确保gmail_email有默认值，避免null错误 gmail_email = "未知邮箱" # 将凭证转换为JSON字符串 if isinstance(credentials_data, dict): # 确保json模块在本地作用域可访问 import json credentials_data = json.dumps(credentials_data) gmail_credential = GmailCredential.objects.create( user=self.user, credentials=credentials_data, token_path=self.token_storage_path, gmail_email=gmail_email, name=name, is_default=(gmail_count == 0), # 第一个账号设为默认 updated_at=timezone.now(), is_active=True ) self.gmail_credential = gmail_credential logger.info(f"已创建新的Gmail凭证，Gmail账号: {gmail_email}, 名称: {name}") # 更新单例 GmailServiceManager.update_instance( self.user, self.credentials, self.gmail_service, self.gmail_credential ) return { 'status': 'success', 'gmail_email': gmail_email } except Exception as e: logger.error(f"处理授权码失败: {str(e)}") import traceback logger.error(traceback.format_exc()) return { 'status': 'error', 'message': str(e) } def _upload_attachments_to_knowledge_base(self, knowledge_base, conversations): """上传所有邮件附件到知识库""" try: # 收集所有需要上传的附件 attachments_to_upload = [] # 从conversations中提取所有附件信息 for message in conversations: if 'attachments' in message and message['attachments']: message_id = message.get('id', '') sender = message.get('from', '未知发件人') subject = message.get('subject', '无主题') date_str = message.get('date', '未知时间') for attachment in message['attachments']: if 'attachmentId' in attachment and 'filename' in attachment: # 下载附件（如果尚未下载） filepath = None # 检查数据库中是否已有记录 existing_attachment = GmailAttachment.objects.filter( gmail_message_id=message_id, filename=attachment['filename'] ).first() if existing_attachment and os.path.exists(existing_attachment.filepath): filepath = existing_attachment.filepath else: # 下载附件 filepath = self.download_attachment( message_id=message_id, attachment_id=attachment['attachmentId'], filename=attachment['filename'] ) if filepath: attachments_to_upload.append({ 'filepath': filepath, 'filename': attachment['filename'], 'message_id': message_id, 'sender': sender, 'subject': subject, 'date': date_str }) # 上传收集到的所有附件 if attachments_to_upload: logger.info(f"开始上传 {len(attachments_to_upload)} 个附件到知识库") from .views import KnowledgeBaseViewSet import django.core.files.uploadedfile as uploadedfile # 导入FileUploadParser from rest_framework.parsers import FileUploadParser # 创建视图集实例 kb_viewset = KnowledgeBaseViewSet() # 批量上传附件 for i in range(0, len(attachments_to_upload), 10): # 每批最多10个文件 batch = attachments_to_upload[i:i+10] files = [] for att in batch: filepath = att['filepath'] filename = att['filename'] # 确认文件存在 if not os.path.exists(filepath): logger.warning(f"附件文件不存在: {filepath}") continue # 读取文件并创建UploadedFile对象 with open(filepath, 'rb') as f: file_content = f.read() files.append(uploadedfile.SimpleUploadedFile( name=filename, content=file_content, content_type=mimetypes.guess_type(filepath)[0] or 'application/octet-stream' )) # 如果这批有文件，调用_call_split_api_multiple上传 if files: # 直接调用文档分割API split_response = kb_viewset._call_split_api_multiple(files) if not split_response or split_response.get('code') != 200: logger.error(f"附件批量分割失败: {split_response.get('message', '未知错误') if split_response else '请求失败'}") continue # 处理分割后的文档 documents_data = split_response.get('data', []) # 对每个文档调用上传API for doc in documents_data: doc_name = doc.get('name', 'Gmail附件') doc_content = doc.get('content', []) # 准备文档数据 upload_doc_data = { "name": doc_name, "paragraphs": [] } # 将所有段落添加到文档中 for paragraph in doc_content: upload_doc_data["paragraphs"].append({ "content": paragraph.get('content', ''), "title": paragraph.get('title', ''), "is_active": True, "problem_list": [] }) # 调用文档上传API upload_response = kb_viewset._call_upload_api(knowledge_base.external_id, upload_doc_data) if upload_response and upload_response.get('code') == 200 and upload_response.get('data'): # 上传成功，保存记录到数据库 document_id = upload_response['data']['id'] # 创建文档记录 from .models import KnowledgeBaseDocument doc_record = KnowledgeBaseDocument.objects.create( knowledge_base=knowledge_base, document_id=document_id, document_name=doc_name, external_id=document_id ) logger.info(f"Gmail附件文档上传成功，ID: {document_id}, 文件名: {doc_name}") # 更新知识库文档计数 knowledge_base.document_count = KnowledgeBaseDocument.objects.filter( knowledge_base=knowledge_base, status='active' ).count() knowledge_base.save() logger.info(f"完成附件上传，共 {len(attachments_to_upload)} 个文件") except Exception as e: logger.error(f"上传附件到知识库失败: {str(e)}") import traceback logger.error(traceback.format_exc()) def _upload_message_attachments_to_knowledge_base(self, knowledge_base, attachment_records): """上传单个邮件的附件到知识库""" try: # 检查是否有附件需要上传 if not attachment_records: return logger.info(f"开始上传 {len(attachment_records)} 个附件到知识库") from .views import KnowledgeBaseViewSet import django.core.files.uploadedfile as uploadedfile # 创建视图集实例 kb_viewset = KnowledgeBaseViewSet() # 准备文件列表 files = [] for att in attachment_records: filepath = att.get('filepath') filename = att.get('filename') # 确认文件存在 if not os.path.exists(filepath): logger.warning(f"附件文件不存在: {filepath}") continue # 读取文件并创建UploadedFile对象 with open(filepath, 'rb') as f: file_content = f.read() files.append(uploadedfile.SimpleUploadedFile( name=filename, content=file_content, content_type=mimetypes.guess_type(filepath)[0] or 'application/octet-stream' )) # 如果有文件，调用_call_split_api_multiple上传 if files: # 直接调用文档分割API split_response = kb_viewset._call_split_api_multiple(files) if not split_response or split_response.get('code') != 200: logger.error(f"附件分割失败: {split_response.get('message', '未知错误') if split_response else '请求失败'}") return # 处理分割后的文档 documents_data = split_response.get('data', []) # 对每个文档调用上传API for doc in documents_data: doc_name = doc.get('name', 'Gmail附件') doc_content = doc.get('content', []) # 准备文档数据 upload_doc_data = { "name": doc_name, "paragraphs": [] } # 将所有段落添加到文档中 for paragraph in doc_content: upload_doc_data["paragraphs"].append({ "content": paragraph.get('content', ''), "title": paragraph.get('title', ''), "is_active": True, "problem_list": [] }) # 调用文档上传API upload_response = kb_viewset._call_upload_api(knowledge_base.external_id, upload_doc_data) if upload_response and upload_response.get('code') == 200 and upload_response.get('data'): # 上传成功，保存记录到数据库 document_id = upload_response['data']['id'] # 创建文档记录 from .models import KnowledgeBaseDocument doc_record = KnowledgeBaseDocument.objects.create( knowledge_base=knowledge_base, document_id=document_id, document_name=doc_name, external_id=document_id ) logger.info(f"Gmail附件文档上传成功，ID: {document_id}, 文件名: {doc_name}") # 更新知识库文档计数 knowledge_base.document_count = KnowledgeBaseDocument.objects.filter( knowledge_base=knowledge_base, status='active' ).count() knowledge_base.save() logger.info(f"完成附件上传，共 {len(files)} 个文件") except Exception as e: logger.error(f"上传附件到知识库失败: {str(e)}") import traceback logger.error(traceback.format_exc()) def get_recent_emails(self, from_email=None, max_results=10): """获取最近的邮件""" try: service = self.gmail_service if not service: logger.error("Gmail服务未初始化") return [] query = "" if from_email: query = f"from:{from_email}" # 获取收件箱中的邮件列表 results = service.users().messages().list( userId='me', maxResults=max_results, q=query ).execute() messages = results.get('messages', []) emails = [] for message in messages: msg = service.users().messages().get(userId='me', id=message['id']).execute() email_data = self._extract_email_content(msg) emails.append(email_data) return emails except Exception as e: logger.error(f"获取最近邮件失败: {str(e)}") logger.error(traceback.format_exc()) return [] def manage_user_goal(self, goal_content=None): """ 创建或更新用户总目标 Args: goal_content (str): 用户总目标内容，如果为None则返回当前总目标 Returns: dict: 包含操作结果和总目标信息 """ from .models import UserGoal try: # 如果未提供内容，则返回当前总目标 if goal_content is None: user_goal = UserGoal.objects.filter(user=self.user, is_active=True).first() if user_goal: return { 'status': 'success', 'action': 'retrieve', 'goal': { 'id': str(user_goal.id), 'content': user_goal.content, 'created_at': user_goal.created_at.strftime('%Y-%m-%d %H:%M:%S'), 'updated_at': user_goal.updated_at.strftime('%Y-%m-%d %H:%M:%S') } } else: return { 'status': 'success', 'action': 'retrieve', 'goal': None } # 查找当前活跃的用户总目标 user_goal = UserGoal.objects.filter(user=self.user, is_active=True).first() # 如果已存在，则更新 if user_goal: user_goal.content = goal_content user_goal.save() action = 'update' else: # 否则创建新的总目标 user_goal = UserGoal.objects.create( user=self.user, content=goal_content ) action = 'create' return { 'status': 'success', 'action': action, 'goal': { 'id': str(user_goal.id), 'content': user_goal.content, 'created_at': user_goal.created_at.strftime('%Y-%m-%d %H:%M:%S'), 'updated_at': user_goal.updated_at.strftime('%Y-%m-%d %H:%M:%S') } } except Exception as e: logger.error(f"管理用户总目标失败: {str(e)}") logger.error(traceback.format_exc()) return { 'status': 'error', 'message': f"管理用户总目标失败: {str(e)}" } def generate_conversation_summary(self, talent_email): """ 为用户与特定达人的所有对话生成总结 Args: talent_email (str): 达人的邮箱地址 Returns: dict: 包含操作结果和总结信息 """ from .models import ConversationSummary try: # 获取与该达人的所有对话 conversations = self.get_conversations(talent_email) if not conversations: return { 'status': 'error', 'message': f"未找到与{talent_email}的对话记录" } # 准备对话历史记录 conversation_history = [] for conversation in conversations: if 'subject' in conversation and conversation['subject']: conversation_history.append({ 'role': 'system', 'content': f"邮件主题: {conversation['subject']}" }) for message in conversation.get('messages', []): role = 'user' if message.get('from_email') == talent_email else 'assistant' content = message.get('body', '') if content: conversation_history.append({ 'role': role, 'content': content }) # 调用DeepSeek API生成总结 summary = self._generate_summary_from_deepseek(conversation_history) if not summary: return { 'status': 'error', 'message': "生成对话总结失败" } # 保存或更新总结 conversation_id = conversations[0].get('id') if conversations else None if not conversation_id: return { 'status': 'error', 'message': "无法确定对话ID" } # 查找现有总结 conversation_summary = ConversationSummary.objects.filter( user=self.user, talent_email=talent_email, conversation_id=conversation_id, is_active=True ).first() # 如果已存在，则更新 if conversation_summary: conversation_summary.summary = summary conversation_summary.save() action = 'update' else: # 否则创建新的总结 conversation_summary = ConversationSummary.objects.create( user=self.user, talent_email=talent_email, conversation_id=conversation_id, summary=summary ) action = 'create' return { 'status': 'success', 'action': action, 'summary': { 'id': str(conversation_summary.id), 'talent_email': talent_email, 'conversation_id': conversation_id, 'summary': summary, 'created_at': conversation_summary.created_at.strftime('%Y-%m-%d %H:%M:%S'), 'updated_at': conversation_summary.updated_at.strftime('%Y-%m-%d %H:%M:%S') } } except Exception as e: logger.error(f"生成对话总结失败: {str(e)}") logger.error(traceback.format_exc()) return { 'status': 'error', 'message': f"生成对话总结失败: {str(e)}" } def _generate_summary_from_deepseek(self, conversation_history): """调用DeepSeek API生成对话总结""" try: # 使用有效的API密钥 api_key = "" # 尝试从环境变量获取 import os from dotenv import load_dotenv load_dotenv() env_api_key = os.environ.get('DEEPSEEK_API_KEY') if env_api_key: api_key = env_api_key # 从Django设置中获取密钥 from django.conf import settings if hasattr(settings, 'DEEPSEEK_API_KEY') and settings.DEEPSEEK_API_KEY: api_key = settings.DEEPSEEK_API_KEY # 如果仍然没有有效的API密钥，使用默认值 if not api_key: api_key = "sk-xqbujijjqqmlmlvkhvxeogqjtzslnhdtqxqgiyuhwpoqcjvf" logger.warning("使用默认API密钥，请在环境变量或settings.py中设置DEEPSEEK_API_KEY") url = "https://api.siliconflow.cn/v1/chat/completions" # 系统消息指定生成总结的任务 system_message = { "role": "system", "content": "你是一位专业的电商客服和达人助手。你的任务是分析用户与达人之间的所有对话历史，并生成一份简明扼要的总结。总结应包括：1. 主要讨论的产品或服务；2. 达人的主要关注点和需求；3. 已经达成的共识或协议；4. 未解决的问题或后续需要跟进的事项。总结应该客观、全面、结构清晰。" } messages = [system_message] # 添加对话历史，但限制消息数量避免超出token限制 # 如果对话历史太长，可能需要进一步处理或分割 if len(conversation_history) > 20: # 选取关键消息：第一条、最后几条以及中间的一些重要消息 selected_messages = ( conversation_history[:2] + # 前两条 conversation_history[len(conversation_history)//2-2:len(conversation_history)//2+2] + # 中间四条 conversation_history[-6:] # 最后六条 ) messages.extend(selected_messages) else: messages.extend(conversation_history) # 添加用户指令 messages.append({ "role": "user", "content": "请根据以上对话历史生成一份全面的总结。" }) # 构建API请求 payload = { "model": "deepseek-ai/DeepSeek-V3", "messages": messages, "stream": False, "max_tokens": 1500, # 增加token上限以容纳完整总结 "temperature": 0.3, # 降低随机性，使总结更加确定性 "top_p": 0.9, "top_k": 50, "frequency_penalty": 0.5, "presence_penalty": 0.2, "n": 1, "stop": [], "response_format": { "type": "text" } } headers = { "Content-Type": "application/json", "Authorization": f"Bearer {api_key}" } logger.info(f"开始调用DeepSeek API生成对话总结") response = requests.post(url, json=payload, headers=headers) if response.status_code != 200: logger.error(f"DeepSeek API调用失败: {response.status_code}, {response.text}") return None result = response.json() logger.debug(f"DeepSeek API返回: {result}") # 提取回复内容 if 'choices' in result and len(result['choices']) > 0: summary = result['choices'][0]['message']['content'] # 如果返回的内容为空，直接返回None if not summary or summary.strip() == '': logger.warning("DeepSeek API返回的总结内容为空") return None return summary logger.warning(f"DeepSeek API返回格式异常: {result}") return None except Exception as e: logger.error(f"调用DeepSeek API生成总结失败: {str(e)}") logger.error(traceback.format_exc()) return None def get_oauth_flow(self): """创建OAuth2流程处理器""" try: # 确保client_secret_json已提供 if not self.client_secret: logger.error("未提供client_secret_json，无法创建OAuth流程") raise ValueError("未提供client_secret_json") # 创建临时文件存储client_secret client_secret_path = 'client_secret.json' with open(client_secret_path, 'w') as f: if isinstance(self.client_secret, str): try: # 确保是有效的JSON import json # 在本地作用域引入 json_data = json.loads(self.client_secret) # 强制设置redirect_uris为非浏览器模式，避免localhost连接拒绝问题 for key in ['web', 'installed']: if key in json_data and 'redirect_uris' in json_data[key]: json_data[key]['redirect_uris'] = ['urn:ietf:wg:oauth:2.0:oob'] logger.info("已强制设置redirect_uri为非浏览器模式") json.dump(json_data, f) except json.JSONDecodeError as e: logger.error(f"client_secret不是有效的JSON: {str(e)}") raise ValueError(f"client_secret不是有效的JSON: {str(e)}") else: # 如果是字典，也进行相同处理 import json # 在本地作用域引入 client_secret_dict = dict(self.client_secret) for key in ['web', 'installed']: if key in client_secret_dict and 'redirect_uris' in client_secret_dict[key]: client_secret_dict[key]['redirect_uris'] = ['urn:ietf:wg:oauth:2.0:oob'] logger.info("已强制设置redirect_uri为非浏览器模式") json.dump(client_secret_dict, f) # 从client_secret创建flow flow = client.flow_from_clientsecrets( client_secret_path, self.SCOPES, redirect_uri='urn:ietf:wg:oauth:2.0:oob' ) return flow except Exception as e: logger.error(f"创建OAuth流程失败: {str(e)}") logger.error(traceback.format_exc()) raise e finally: # 删除临时文件 if os.path.exists(client_secret_path): try: os.unlink(client_secret_path) except Exception as del_e: logger.error(f"删除临时文件失败: {str(del_e)}") def check_and_renew_watch(self): """检查并更新Gmail监听状态 如果监听即将过期或已经过期，则自动更新监听 """ try: from .models import GmailCredential # 获取用户的Gmail凭证 credential = GmailCredential.objects.filter( user=self.user, is_active=True, gmail_credential_id=self.gmail_credential_id if self.gmail_credential_id else None ).first() if not credential: logger.error(f"找不到用户 {self.user.email} 的Gmail凭证") return False # 检查监听是否需要更新 needs_renewal = False # 如果没有watch_expiration，需要设置监听 if not credential.watch_expiration: logger.info(f"Gmail凭证 {credential.gmail_email} 没有监听过期时间，需要设置监听") needs_renewal = True else: # 如果监听将在24小时内过期，更新监听 now = timezone.now() if hasattr(settings, 'USE_TZ') and settings.USE_TZ else datetime.now() time_until_expiration = credential.watch_expiration - now hours_until_expiration = time_until_expiration.total_seconds() / 3600 if hours_until_expiration < 24: logger.info(f"Gmail监听将在 {hours_until_expiration:.2f} 小时后过期，需要更新") needs_renewal = True if needs_renewal: logger.info(f"为Gmail凭证 {credential.gmail_email} 更新监听") watch_result = self.setup_watch() if watch_result and 'historyId' in watch_result: logger.info(f"Gmail监听更新成功: historyId={watch_result['historyId']}") return True else: logger.error("Gmail监听更新失败") return False else: logger.info(f"Gmail监听状态良好，无需更新。过期时间: {credential.watch_expiration}") return True except Exception as e: logger.error(f"检查和更新Gmail监听失败: {str(e)}") logger.error(traceback.format_exc()) return False def verify_connectivity(self): """验证与Gmail API的连接性 测试Gmail API连接，并返回连接状态 Returns: dict: 包含连接测试结果的信息 """ try: if not hasattr(self, 'gmail_service') or not self.gmail_service: logger.warning("Gmail服务未初始化，尝试认证") if not self.authenticate(): return { 'status': 'error', 'message': 'Gmail认证失败', 'is_connected': False } # 尝试获取用户的个人资料，这是一个轻量级的API调用 profile = self.gmail_service.users().getProfile(userId='me').execute() # 检查是否有必要的监听信息 from .models import GmailCredential credential = GmailCredential.objects.filter( user=self.user, is_active=True, gmail_credential_id=self.gmail_credential_id if self.gmail_credential_id else None ).first() watch_info = {} if credential: watch_info = { 'has_watch': credential.watch_expiration is not None, 'watch_expiration': credential.watch_expiration.strftime('%Y-%m-%d %H:%M:%S') if credential.watch_expiration else None, 'last_history_id': credential.last_history_id } return { 'status': 'success', 'message': 'Gmail连接正常', 'is_connected': True, 'profile': { 'email': profile.get('emailAddress'), 'messages_total': profile.get('messagesTotal'), 'threads_total': profile.get('threadsTotal') }, 'watch_info': watch_info } except Exception as e: logger.error(f"Gmail连接测试失败: {str(e)}") logger.error(traceback.format_exc()) error_message = str(e) suggestions = [] if "invalid_grant" in error_message.lower(): suggestions.append("OAuth令牌已过期，需要重新授权") elif "401" in error_message: suggestions.append("认证失败，请重新登录Gmail账号") elif "EOF occurred in violation of protocol" in error_message: suggestions.append("SSL连接问题，可能是网络或代理配置问题") elif "connect timeout" in error_message.lower() or "connection timeout" in error_message.lower(): suggestions.append("连接超时，请检查网络连接和代理设置") return { 'status': 'error', 'message': f'Gmail连接失败: {error_message}', 'is_connected': False, 'suggestions': suggestions } @classmethod def batch_renew_watches(cls): """ 批量检查和更新所有用户的Gmail监听状态 此方法可以由定时任务调用，确保所有用户的Gmail监听都保持活跃状态 Returns: dict: 包含批处理结果的信息 """ try: from .models import GmailCredential, User import time # 获取所有活跃的Gmail凭证 now = timezone.now() if hasattr(settings, 'USE_TZ') and settings.USE_TZ else datetime.now() # 计算24小时后的时间 expiration_threshold = now + timedelta(hours=24) # 获取即将过期或没有监听的凭证 credentials_to_update = GmailCredential.objects.filter( is_active=True ).filter( Q(watch_expiration__lt=expiration_threshold) | Q(watch_expiration__isnull=True) ) logger.info(f"找到 {credentials_to_update.count()} 个需要更新监听的Gmail凭证") success_count = 0 failure_count = 0 # 依次处理每个凭证 for credential in credentials_to_update: try: user = credential.user if not user or not user.is_active: logger.warning(f"跳过已停用用户的凭证: {credential.id}") continue logger.info(f"处理用户 {user.email} 的Gmail凭证 {credential.gmail_email}") # 创建Gmail集成实例 gmail = cls(user, gmail_credential_id=credential.id) # 认证 if not gmail.authenticate(): logger.error(f"用户 {user.email} 的Gmail认证失败") failure_count += 1 continue # 更新监听 result = gmail.setup_watch() if result and 'historyId' in result: logger.info(f"成功更新用户 {user.email} 的Gmail监听") success_count += 1 else: logger.error(f"更新用户 {user.email} 的Gmail监听失败") failure_count += 1 # 休眠一小段时间，避免请求过于频繁 time.sleep(1) except Exception as e: failure_count += 1 logger.error(f"处理凭证 {credential.id} 时出错: {str(e)}") logger.error(traceback.format_exc()) return { 'status': 'success', 'message': f'批量更新Gmail监听完成', 'total_processed': credentials_to_update.count(), 'success_count': success_count, 'failure_count': failure_count } except Exception as e: logger.error(f"批量更新Gmail监听失败: {str(e)}") logger.error(traceback.format_exc()) return { 'status': 'error', 'message': f'批量更新Gmail监听失败: {str(e)}' } def refresh_token(self): """ 尝试刷新OAuth令牌，如果失败则标记凭证需要重新授权 Returns: bool: 刷新是否成功 """ try: logger.info(f"尝试刷新用户 {self.user.email} 的OAuth令牌") # 清除当前服务实例 GmailServiceManager.clear_instance(self.user, self.gmail_credential_id) # 尝试重新认证 if not self.authenticate(): logger.error("重新认证失败") self._mark_credential_needs_reauth() return False # 验证新的令牌是否有效 try: # 设置超时 original_timeout = socket.getdefaulttimeout() socket.setdefaulttimeout(GMAIL_REQUEST_TIMEOUT) try: # 尝试一个简单的API调用来验证 profile = self.gmail_service.users().getProfile(userId='me').execute() logger.info(f"令牌刷新成功，已验证: {profile.get('emailAddress')}") return True except Exception as e: error_msg = str(e) if "invalid_grant" in error_msg.lower() or "401" in error_msg: logger.error(f"令牌仍然无效: {error_msg}") self._mark_credential_needs_reauth() return False else: # 其他错误，但令牌可能是有效的 logger.warning(f"令牌可能有效，但API调用失败: {error_msg}") return True finally: # 恢复原始超时设置 socket.setdefaulttimeout(original_timeout) except Exception as e: logger.error(f"验证新令牌时出错: {str(e)}") return False except Exception as e: logger.error(f"刷新令牌失败: {str(e)}") logger.error(traceback.format_exc()) self._mark_credential_needs_reauth() return False def _mark_credential_needs_reauth(self): """标记凭证需要重新授权""" try: from .models import GmailCredential # 查找当前凭证 credential = GmailCredential.objects.filter( user=self.user, id=self.gmail_credential_id if self.gmail_credential_id else None, is_active=True ).first() if credential: # 标记需要重新授权 credential.needs_reauth = True credential.save() logger.info(f"已标记Gmail凭证 {credential.id} 需要重新授权") # 从缓存中移除服务实例 GmailServiceManager.clear_instance(self.user, self.gmail_credential_id) except Exception as e: logger.error(f"标记凭证需要重新授权失败: {str(e)}") @classmethod def process_queued_notifications(cls, user=None): """ 处理队列中的通知，可以在用户重新授权后调用 Args: user: 可选的用户参数，如果提供，只处理该用户的通知 Returns: dict: 处理结果统计 """ try: from .models import GmailNotificationQueue, GmailCredential import json # 构建查询 query = Q(processed=False) if user: query &= Q(user=user) # 获取未处理的通知 queued_notifications = GmailNotificationQueue.objects.filter(query) logger.info(f"找到 {queued_notifications.count()} 条未处理的Gmail通知") processed_count = 0 success_count = 0 # 按用户ID和Gmail凭证ID分组处理 from django.db.models import Count user_creds = queued_notifications.values('user_id', 'gmail_credential_id').annotate( count=Count('id') ).order_by('user_id', 'gmail_credential_id') for user_cred in user_creds: user_id = user_cred['user_id'] cred_id = user_cred['gmail_credential_id'] # 获取凭证信息 credential = GmailCredential.objects.filter(id=cred_id, is_active=True).first() if not credential or credential.needs_reauth: logger.warning(f"凭证 {cred_id} 需要重新授权，跳过相关通知") continue # 获取用户对象 from .models import User user_obj = User.objects.get(id=user_id) # 初始化Gmail集成 gmail_integration = cls(user_obj, gmail_credential_id=cred_id) # 获取用户的队列通知 user_notifications = queued_notifications.filter( user_id=user_id, gmail_credential_id=cred_id ).order_by('created_at') # 按创建时间排序 for notification in user_notifications: try: # 解析通知数据 try: notification_data = json.loads(notification.notification_data) except: notification_data = { 'emailAddress': notification.email, 'historyId': notification.history_id } # 处理通知 result = gmail_integration.process_notification(notification_data) # 更新处理状态 notification.processed = True notification.success = result notification.processed_at = timezone.now() notification.save() processed_count += 1 if result: success_count += 1 except Exception as e: logger.error(f"处理队列通知 {notification.id} 失败: {str(e)}") # 标记处理失败 notification.processed = True notification.success = False notification.error_message = str(e)[:255] # 截断错误消息 notification.processed_at = timezone.now() notification.save() processed_count += 1 return { 'status': 'success', 'total_processed': processed_count, 'success_count': success_count, 'remaining': queued_notifications.count() - processed_count } except Exception as e: logger.error(f"处理队列通知失败: {str(e)}") logger.error(traceback.format_exc()) return { 'status': 'error', 'message': str(e) } def _load_credentials_from_storage(self, credential_data): """从存储中加载凭证，处理可能的格式不匹配问题""" logger.info("尝试加载凭证数据") if not credential_data: logger.error("凭证数据为空") return None # 方法1: 尝试直接JSON解析 try: import json if isinstance(credential_data, str): cred_json = credential_data else: cred_json = credential_data.decode('utf-8') # 验证是否为有效JSON json.loads(cred_json) # 使用OAuth2Credentials从JSON创建凭证 logger.info("成功从JSON创建凭证") return client.OAuth2Credentials.from_json(cred_json) except Exception as json_error: logger.error(f"JSON解析凭证失败: {str(json_error)}") # 方法2: 尝试pickle解析 try: if isinstance(credential_data, str): # 如果是字符串，尝试编码为二进制 pickle_data = credential_data.encode('latin1') else: pickle_data = credential_data logger.info("尝试使用pickle解析凭证") return pickle.loads(pickle_data) except Exception as pickle_error: logger.error(f"Pickle解析凭证失败: {str(pickle_error)}") # 所有方法都失败 logger.error("所有凭证解析方法都失败") return None