新增URL&IPC功能（基础）

Author: lzy98276

app/common/IPC_URL/__init__.py

@@ -0,0 +1,47 @@
+"""
+URL和IPC混合体模块 - 跨平台通用实现
+支持Windows和Linux系统的URL协议注册和IPC通信
+
+主要功能：
+1. URL协议注册管理（跨平台）
+2. IPC进程间通信
+3. 命令行参数处理
+4. 多实例检测和管理
+
+快速开始：
+    from app.common.IPC_URL import URLIPCHandler
+    
+    # 创建处理器
+    handler = URLIPCHandler("MyApp", "myapp")
+    
+    # 注册协议
+    handler.register_url_protocol()
+    
+    # 启动IPC服务器
+    handler.start_ipc_server()
+    
+    # 注册消息处理器
+    def handle_url(payload):
+        print(f"收到URL: {payload.get('url', '')}")
+        return {"status": "success"}
+    
+    handler.register_message_handler('url', handle_url)
+
+详细文档请参考：
+- README.md: 完整的使用文档
+- QUICK_START.md: 快速开始指南
+- example_usage.py: 使用示例
+"""
+
+from .url_ipc_handler import URLIPCHandler
+from .protocol_manager import ProtocolManager
+from .url_command_handler import URLCommandHandler
+from .security_verifier import SecurityVerifier, SimplePasswordVerifier, DynamicPasswordVerifier, CompositeVerifier
+
+__version__ = "1.0.0"
+__author__ = "SecRandom Team"
+__description__ = "跨平台URL协议注册和IPC通信处理器"
+
+__all__ = ['URLIPCHandler', 'ProtocolManager', 'URLCommandHandler', 
+           'SecurityVerifier', 'SimplePasswordVerifier', 'DynamicPasswordVerifier', 'CompositeVerifier',
+           '__version__', '__author__', '__description__']

app/common/IPC_URL/protocol_manager.py

@@ -0,0 +1,297 @@
+"""
+协议管理器 - 跨平台URL协议注册管理
+支持Windows和Linux系统的自定义协议注册
+"""
+import os
+import sys
+import winreg
+import subprocess
+from pathlib import Path
+from typing import Optional
+from loguru import logger
+
+
+
+class ProtocolManager:
+    """协议管理器 - 处理跨平台的URL协议注册"""
+    
+    def __init__(self, app_name: str, protocol_name: str):
+        """
+        初始化协议管理器
+        
+        Args:
+            app_name: 应用程序名称
+            protocol_name: 自定义协议名称（不含://）
+        """
+        self.app_name = app_name
+        self.protocol_name = protocol_name
+        self.is_windows = sys.platform.startswith('win')
+        self.is_linux = sys.platform.startswith('linux')
+        
+    def register_protocol(self) -> bool:
+        """
+        注册自定义协议
+        
+        Returns:
+            注册成功返回True，失败返回False
+        """
+        if self.is_windows:
+            return self._register_windows_protocol()
+        elif self.is_linux:
+            return self._register_linux_protocol()
+        else:
+            logger.error(f"不支持的操作系统: {sys.platform}")
+            return False
+            
+    def unregister_protocol(self) -> bool:
+        """
+        注销自定义协议
+        
+        Returns:
+            注销成功返回True，失败返回False
+        """
+        if self.is_windows:
+            return self._unregister_windows_protocol()
+        elif self.is_linux:
+            return self._unregister_linux_protocol()
+        else:
+            logger.error(f"不支持的操作系统: {sys.platform}")
+            return False
+            
+    def is_protocol_registered(self) -> bool:
+        """
+        检查协议是否已注册
+        
+        Returns:
+            已注册返回True，未注册返回False
+        """
+        if self.is_windows:
+            return self._is_windows_protocol_registered()
+        elif self.is_linux:
+            return self._is_linux_protocol_registered()
+        else:
+            return False
+            
+    def _register_windows_protocol(self) -> bool:
+        """Windows系统注册协议"""
+        try:
+            # 获取当前可执行文件路径
+            exe_path = self._get_executable_path()
+            
+            # 首先尝试注册到HKEY_CLASSES_ROOT
+            try:
+                # 注册协议到HKEY_CLASSES_ROOT
+                with winreg.CreateKey(winreg.HKEY_CLASSES_ROOT, self.protocol_name) as key:
+                    winreg.SetValueEx(key, "", 0, winreg.REG_SZ, f"URL:{self.app_name} Protocol")
+                    winreg.SetValueEx(key, "URL Protocol", 0, winreg.REG_SZ, "")
+                    
+                # 注册命令
+                command_key_path = f"{self.protocol_name}\\shell\\open\\command"
+                with winreg.CreateKey(winreg.HKEY_CLASSES_ROOT, command_key_path) as key:
+                    winreg.SetValueEx(key, "", 0, winreg.REG_SZ, f'"{exe_path}" --url "%1"')
+                    
+                return True
+                
+            except (OSError, PermissionError) as e:
+                # 如果HKEY_CLASSES_ROOT失败，尝试注册到HKEY_CURRENT_USER
+                if e.errno == 5 or "Access is denied" in str(e):  # WinError 5
+                    logger.error(f"管理员权限不足，尝试注册到当前用户: {e}")
+                    return self._register_windows_protocol_current_user(exe_path)
+                else:
+                    raise
+                    
+        except Exception as e:
+            logger.error(f"Windows协议注册失败: {e}")
+            return False
+            
+    def _register_windows_protocol_current_user(self, exe_path: str) -> bool:
+        """注册到当前用户的Windows协议（无需管理员权限）"""
+        try:
+            # 注册协议到HKEY_CURRENT_USER\Software\Classes
+            key_path = f"Software\\Classes\\{self.protocol_name}"
+            with winreg.CreateKey(winreg.HKEY_CURRENT_USER, key_path) as key:
+                winreg.SetValueEx(key, "", 0, winreg.REG_SZ, f"URL:{self.app_name} Protocol")
+                winreg.SetValueEx(key, "URL Protocol", 0, winreg.REG_SZ, "")
+                
+            # 注册命令
+            command_key_path = f"Software\\Classes\\{self.protocol_name}\\shell\\open\\command"
+            with winreg.CreateKey(winreg.HKEY_CURRENT_USER, command_key_path) as key:
+                winreg.SetValueEx(key, "", 0, winreg.REG_SZ, f'"{exe_path}" --url "%1"')
+                
+            return True
+            
+        except Exception as e:
+            logger.error(f"Windows当前用户协议注册失败: {e}")
+            return False
+            
+    def _unregister_windows_protocol(self) -> bool:
+        """Windows系统注销协议"""
+        try:
+            # 首先尝试删除系统级注册（HKEY_CLASSES_ROOT）
+            try:
+                winreg.DeleteKey(winreg.HKEY_CLASSES_ROOT, f"{self.protocol_name}\\shell\\open\\command")
+                winreg.DeleteKey(winreg.HKEY_CLASSES_ROOT, f"{self.protocol_name}\\shell\\open")
+                winreg.DeleteKey(winreg.HKEY_CLASSES_ROOT, f"{self.protocol_name}\\shell")
+                winreg.DeleteKey(winreg.HKEY_CLASSES_ROOT, self.protocol_name)
+                return True
+            except (OSError, PermissionError) as e:
+                if e.errno == 5 or "Access is denied" in str(e):  # WinError 5
+                    # 如果系统级删除失败，尝试删除当前用户注册
+                    return self._unregister_windows_protocol_current_user()
+                else:
+                    raise
+                    
+        except Exception as e:
+            logger.error(f"Windows协议注销失败: {e}")
+            return False
+            
+    def _unregister_windows_protocol_current_user(self) -> bool:
+        """注销当前用户的Windows协议"""
+        try:
+            # 删除当前用户注册表项
+            key_path = f"Software\\Classes\\{self.protocol_name}\\shell\\open\\command"
+            winreg.DeleteKey(winreg.HKEY_CURRENT_USER, key_path)
+            winreg.DeleteKey(winreg.HKEY_CURRENT_USER, f"Software\\Classes\\{self.protocol_name}\\shell\\open")
+            winreg.DeleteKey(winreg.HKEY_CURRENT_USER, f"Software\\Classes\\{self.protocol_name}\\shell")
+            winreg.DeleteKey(winreg.HKEY_CURRENT_USER, f"Software\\Classes\\{self.protocol_name}")
+            return True
+            
+        except Exception as e:
+            logger.error(f"Windows当前用户协议注销失败: {e}")
+            return False
+            
+    def _is_windows_protocol_registered(self) -> bool:
+        """检查Windows协议是否已注册"""
+        try:
+            # 首先检查HKEY_CLASSES_ROOT（系统级）
+            with winreg.OpenKey(winreg.HKEY_CLASSES_ROOT, self.protocol_name) as key:
+                return True
+        except (OSError, FileNotFoundError):
+            try:
+                # 如果没找到，检查HKEY_CURRENT_USER（用户级）
+                key_path = f"Software\\Classes\\{self.protocol_name}"
+                with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path) as key:
+                    return True
+            except (OSError, FileNotFoundError):
+                return False
+            
+    def _register_linux_protocol(self) -> bool:
+        """Linux系统注册协议"""
+        try:
+            # 获取当前可执行文件路径
+            exe_path = self._get_executable_path()
+            
+            # 创建.desktop文件
+            desktop_file_path = self._get_linux_desktop_file_path()
+            desktop_content = self._generate_linux_desktop_file(exe_path)
+            
+            # 写入.desktop文件
+            with open(desktop_file_path, 'w', encoding='utf-8') as f:
+                f.write(desktop_content)
+                
+            # 设置可执行权限
+            os.chmod(desktop_file_path, 0o755)
+            
+            # 注册协议处理程序
+            self._register_linux_mime_handler()
+            
+            # 更新桌面数据库
+            self._update_linux_desktop_database()
+            
+            return True
+            
+        except Exception as e:
+            logger.error(f"Linux协议注册失败: {e}")
+            return False
+            
+    def _unregister_linux_protocol(self) -> bool:
+        """Linux系统注销协议"""
+        try:
+            desktop_file_path = self._get_linux_desktop_file_path()
+            
+            if os.path.exists(desktop_file_path):
+                os.remove(desktop_file_path)
+                
+            # 更新桌面数据库
+            self._update_linux_desktop_database()
+            
+            return True
+            
+        except Exception as e:
+            logger.error(f"Linux协议注销失败: {e}")
+            return False
+            
+    def _is_linux_protocol_registered(self) -> bool:
+        """检查Linux协议是否已注册"""
+        desktop_file_path = self._get_linux_desktop_file_path()
+        return os.path.exists(desktop_file_path)
+        
+    def _get_linux_desktop_file_path(self) -> str:
+        """获取Linux桌面文件路径"""
+        applications_dir = Path.home() / '.local' / 'share' / 'applications'
+        applications_dir.mkdir(parents=True, exist_ok=True)
+        return str(applications_dir / f"{self.app_name.lower()}-url-handler.desktop")
+        
+    def _generate_linux_desktop_file(self, exe_path: str) -> str:
+        """生成Linux桌面文件内容"""
+        return f"""[Desktop Entry]
+Name={self.app_name} URL Handler
+Comment=Handle {self.protocol_name}:// URLs for {self.app_name}
+Exec={exe_path} --url %u
+Icon={self.app_name.lower()}
+Terminal=false
+Type=Application
+Categories=Utility;
+MimeType=x-scheme-handler/{self.protocol_name};
+"""
+        
+    def _register_linux_mime_handler(self):
+        """注册Linux MIME处理程序"""
+        try:
+            # 使用xdg-mime注册协议处理程序
+            subprocess.run([
+                'xdg-mime', 'default',
+                f'{self.app_name.lower()}-url-handler.desktop',
+                f'x-scheme-handler/{self.protocol_name}'
+            ], check=True, capture_output=True)
+        except (subprocess.CalledProcessError, FileNotFoundError):
+            # 如果xdg-mime不可用，尝试直接创建mimeapps.list
+            self._create_linux_mimeapps_list()
+            
+    def _create_linux_mimeapps_list(self):
+        """创建Linux mimeapps.list文件"""
+        config_dir = Path.home() / '.config'
+        config_dir.mkdir(parents=True, exist_ok=True)
+        
+        mimeapps_file = config_dir / 'mimeapps.list'
+        content = f"""[Default Applications]
+x-scheme-handler/{self.protocol_name}={self.app_name.lower()}-url-handler.desktop
+
+[Added Associations]
+x-scheme-handler/{self.protocol_name}={self.app_name.lower()}-url-handler.desktop
+"""
+        
+        with open(mimeapps_file, 'a', encoding='utf-8') as f:
+            f.write(content)
+            
+    def _update_linux_desktop_database(self):
+        """更新Linux桌面数据库"""
+        try:
+            applications_dir = Path.home() / '.local' / 'share' / 'applications'
+            subprocess.run([
+                'update-desktop-database', str(applications_dir)
+            ], check=True, capture_output=True)
+        except (subprocess.CalledProcessError, FileNotFoundError):
+            # 如果update-desktop-database不可用，忽略错误
+            pass
+            
+    def _get_executable_path(self) -> str:
+        """获取当前可执行文件路径"""
+        if getattr(sys, 'frozen', False):
+            # PyInstaller打包后的可执行文件
+            exe_path = sys.executable
+        else:
+            # 普通Python脚本
+            exe_path = sys.argv[0]
+            
+        return os.path.abspath(exe_path)