Delete protobuf_manager.py

protobuf_manager.py

@@ -1,317 +0,0 @@
-"""
-Protobuf管理模块
-提供统一的protobuf操作接口，直接使用已编译的pb2文件
-"""
-import base64
-from typing import List, Optional, Dict, Any
-from utils import Utils, logger
-
-# 直接导入已编译的protobuf模块
-try:
-    import warp_unified_pb2
-    logger.success("✅ 成功导入已编译的protobuf模块")
-except ImportError as e:
-    logger.error(f"❌ 无法导入protobuf模块: {e}")
-    logger.error("请确保 warp_unified_pb2.py 文件存在且可访问")
-    raise
-
-
-class ProtobufManager:
-    """Protobuf操作管理器"""
-    
-    @staticmethod
-    def create_chat_request(messages: List[Dict], model: str = "gemini-2.0-flash") -> Optional[bytes]:
-        """
-        创建聊天请求的protobuf数据
-        
-        Args:
-            messages: OpenAI格式的消息列表
-            model: 模型名称
-            
-        Returns:
-            序列化的protobuf数据，如果失败返回None
-        """
-        try:
-            from request_converter import RequestConverter
-            
-            # 处理消息历史
-            merged_groups, current_message = RequestConverter.process_message_history(messages)
-            logger.debug(f"🔄 处理消息历史: {len(merged_groups)} 组历史消息, 当前消息: {current_message[:50]}...")
-            
-            # 创建主请求
-            request = warp_unified_pb2.ChatRequest()
-            session_id = Utils.generate_request_id()
-
-            # 如果只有一条消息，只设置当前消息
-            if len(messages) == 1:
-                # 字段1: 会话信息
-                request.session.session_id = session_id
-                
-                # 字段2: 系统信息
-                request.system.environment.timestamp.time = Utils.get_current_timestamp()
-                
-                # 当前请求信息
-                request.system.request.content.message.text = current_message
-                logger.debug("📝 创建单消息请求")
-            else:
-                # 有历史消息时，设置第一个消息组作为历史数据
-                if merged_groups:
-                    first_content = merged_groups[0].get("content", "")
-                    request.session.history.first_message_id = session_id
-                    request.session.history.first_message_content = first_content
-
-                    # 添加系统消息 (固定格式)
-                    entry_a = request.session.history.entries.add()
-                    entry_a.message_id = Utils.generate_request_id()
-                    entry_a.system_msg.content_id = Utils.generate_request_id()
-                    entry_a.system_msg.metadata.data = "IgIQAQ=="
-
-                    # 处理剩余的消息组
-                    for message in merged_groups:
-                        if message.get("role", "") == "user":
-                            # 添加用户消息
-                            entry_b = request.session.history.entries.add()
-                            entry_b.message_id = Utils.generate_request_id()
-                            entry_b.user_msg.content = message.get("content", "")
-                        elif message.get("role", "") == "assistant":
-                            # 添加助手消息
-                            entry_c = request.session.history.entries.add()
-                            entry_c.message_id = Utils.generate_request_id()
-                            entry_c.assistant_msg.content = message.get("content", "")
-                    
-                    logger.debug(f"📚 创建多消息请求: {len(merged_groups)} 条历史消息")
-                
-                # 字段1: 会话信息
-                request.session.session_id = session_id
-                
-                # 字段2: 系统信息
-                request.system.environment.timestamp.time = Utils.get_current_timestamp()
-                
-                # 当前请求信息
-                request.system.request.content.message.text = current_message
-            
-            # 字段3: AI配置
-            request.ai_config.model.model_name = model
-            request.ai_config.model.variant = "o3"
-            request.ai_config.model.mode = "auto"
-            request.ai_config.priority_level_1 = 1
-            request.ai_config.priority_level_2 = 1
-            request.ai_config.priority_level_3 = 1
-            request.ai_config.priority_level_4 = 1
-            request.ai_config.config_data = bytes([0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x00, 0x02, 0x03, 0x01]).decode('utf-8')
-        
-            serialized = request.SerializeToString()
-            logger.success(f"✅ 聊天请求protobuf数据序列化成功，大小: {len(serialized)} 字节")
-            
-            return serialized
-            
-        except Exception as e:
-            logger.error(f"❌ 创建聊天请求protobuf数据时出错: {e}")
-            import traceback
-            traceback.print_exc()
-            return None
-
-    @staticmethod
-    def create_login_request(login_url: str) -> Optional[bytes]:
-        """
-        创建登录请求的protobuf数据
-        
-        Args:
-            login_url: 登录URL
-            
-        Returns:
-            序列化的protobuf数据，如果失败返回None
-        """
-        try:
-            # 创建登录请求
-            request = warp_unified_pb2.LoginRequest()
-
-            # 设置请求标志
-            request.request_flags.extend([9, 18, 20, 21, 31, 35])
-            request.request_type = 2
-            
-            # 设置登录URL
-            request.login_data.url = login_url
-            
-            # 设置配置标志
-            request.config_flags.extend([3, 2])
-            
-            # 设置语言
-            request.language = "zh-CN"
-            
-            # 设置请求设置
-            request.settings.setting_value = 6
-            
-            # 序列化
-            serialized = request.SerializeToString()
-            logger.debug(f"🔄 登录请求protobuf序列化成功，数据长度: {len(serialized)} 字节")
-            return serialized
-            
-        except Exception as e:
-            logger.error(f"❌ 创建登录请求protobuf数据时出错: {e}")
-            return None
-
-    @staticmethod
-    def parse_chat_response(base64_data: str) -> str:
-        """
-        解析聊天响应的protobuf数据
-        
-        Args:
-            base64_data: base64编码的响应数据
-            
-        Returns:
-            解析后的文本内容
-        """
-        try:
-            response = warp_unified_pb2.ChatResponse()
-            b64_data = base64.urlsafe_b64decode(base64_data)
-            response.ParseFromString(b64_data)
-            
-            # 尝试从主要响应中获取文本
-            if hasattr(response.metadata.content, 'primary') and response.metadata.content.primary.text.content.text:
-                text = response.metadata.content.primary.text.content.text
-                logger.debug(f"📥 解析主响应文本: {len(text)} 字符")
-                return text
-            
-            # 尝试从备用响应中获取文本
-            if hasattr(response.metadata.content, 'secondary') and response.metadata.content.secondary.alternative.final_content.text:
-                text = response.metadata.content.secondary.alternative.final_content.text
-                logger.debug(f"📥 解析备用响应文本: {len(text)} 字符")
-                return text
-                
-            logger.debug("⚠️ 响应中未找到文本内容")
-            return ""
-            
-        except Exception as e:
-            logger.error(f"❌ 解析聊天响应protobuf数据时出错: {e}")
-            return ""
-
-    @staticmethod
-    def parse_login_response(response_content: bytes) -> str:
-        """
-        解析登录响应的protobuf数据
-        
-        Args:
-            response_content: 响应的二进制内容
-            
-        Returns:
-            解析后的登录URL
-        """
-        try:
-            import urllib.parse
-            
-            parsed_data = warp_unified_pb2.LoginRequest()
-            parsed_data.ParseFromString(response_content)
-            
-            # 提取嵌套的URL
-            parsed_data_url = parsed_data.url_chain.container_2.container_3.container_4.container_5.container_6.url_container.final_url.login_url
-            logger.info(f"🔄 解析后的URL: {parsed_data_url}")
-            
-            decoded_url = urllib.parse.unquote(parsed_data_url)
-            logger.info(f"🔄 解码后的URL: {decoded_url}")
-            
-            return decoded_url
-            
-        except Exception as e:
-            logger.error(f"❌ 解析登录响应protobuf数据时出错: {e}")
-            return ""
-
-    @staticmethod
-    def get_protobuf_classes() -> Dict[str, Any]:
-        """
-        获取所有可用的protobuf类
-        
-        Returns:
-            包含所有protobuf类的字典
-        """
-        return {
-            'ChatRequest': warp_unified_pb2.ChatRequest,
-            'ChatResponse': warp_unified_pb2.ChatResponse,
-            'LoginRequest': warp_unified_pb2.LoginRequest,
-            'SessionInfo': warp_unified_pb2.SessionInfo,
-            'MessageHistory': warp_unified_pb2.MessageHistory,
-            'MessageEntry': warp_unified_pb2.MessageEntry,
-            'SystemInfo': warp_unified_pb2.SystemInfo,
-            'AIConfig': warp_unified_pb2.AIConfig,
-            'ModelConfig': warp_unified_pb2.ModelConfig,
-            'LoginData': warp_unified_pb2.LoginData,
-            'RequestSettings': warp_unified_pb2.RequestSettings,
-        }
-
-    @staticmethod
-    def create_empty_request(request_type: str) -> Optional[Any]:
-        """
-        创建空的protobuf请求对象
-        
-        Args:
-            request_type: 请求类型 ('chat' 或 'login')
-            
-        Returns:
-            对应的protobuf请求对象
-        """
-        try:
-            if request_type.lower() == 'chat':
-                return warp_unified_pb2.ChatRequest()
-            elif request_type.lower() == 'login':
-                return warp_unified_pb2.LoginRequest()
-            else:
-                logger.error(f"❌ 不支持的请求类型: {request_type}")
-                return None
-        except Exception as e:
-            logger.error(f"❌ 创建空请求对象时出错: {e}")
-            return None
-
-    @staticmethod
-    def validate_protobuf_module() -> bool:
-        """
-        验证protobuf模块是否正确加载
-        
-        Returns:
-            如果模块正确加载返回True，否则返回False
-        """
-        try:
-            # 测试创建基本对象
-            chat_request = warp_unified_pb2.ChatRequest()
-            login_request = warp_unified_pb2.LoginRequest()
-            
-            # 测试基本序列化
-            chat_request.session.session_id = "test"
-            serialized = chat_request.SerializeToString()
-            
-            if len(serialized) > 0:
-                logger.success("✅ Protobuf模块验证通过")
-                return True
-            else:
-                logger.error("❌ Protobuf模块验证失败：序列化结果为空")
-                return False
-                
-        except Exception as e:
-            logger.error(f"❌ Protobuf模块验证失败: {e}")
-            return False
-
-
-# 模块级别的便捷函数
-def create_chat_protobuf(messages: List[Dict], model: str = "gemini-2.0-flash") -> Optional[bytes]:
-    """便捷函数：创建聊天请求protobuf数据"""
-    return ProtobufManager.create_chat_request(messages, model)
-
-
-def create_login_protobuf(login_url: str) -> Optional[bytes]:
-    """便捷函数：创建登录请求protobuf数据"""
-    return ProtobufManager.create_login_request(login_url)
-
-
-def parse_chat_protobuf(base64_data: str) -> str:
-    """便捷函数：解析聊天响应protobuf数据"""
-    return ProtobufManager.parse_chat_response(base64_data)
-
-
-def parse_login_protobuf(response_content: bytes) -> str:
-    """便捷函数：解析登录响应protobuf数据"""
-    return ProtobufManager.parse_login_response(response_content)
-
-
-# 在模块加载时验证protobuf模块
-if __name__ != "__main__":
-    if not ProtobufManager.validate_protobuf_module():
-        logger.warning("⚠️ Protobuf模块验证失败，某些功能可能不可用")