import os import sys import glob import random import threading import shutil import platform import subprocess from typing import List, Dict, Optional import logging def setup_logging(): """配置日志""" logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger(__name__) class BrowserPathFinder: """跨平台浏览器路径查找器""" # 浏览器可执行文件名（按优先级排序） BROWSER_EXECUTABLES = { 'windows': [ 'chrome.exe', 'msedge.exe', 'chromium.exe', 'brave.exe', 'vivaldi.exe', 'opera.exe', ], 'darwin': [ # macOS 'Google Chrome', 'Chromium', 'Microsoft Edge', 'Brave Browser', 'Vivaldi', 'Opera', ], 'linux': [ 'google-chrome', 'google-chrome-stable', 'google-chrome-beta', 'google-chrome-dev', 'google-chrome-unstable', 'chromium', 'chromium-browser', 'microsoft-edge', 'microsoft-edge-stable', 'microsoft-edge-beta', 'microsoft-edge-dev', 'brave-browser', 'brave-browser-stable', 'brave', 'vivaldi', 'vivaldi-stable', 'opera', 'opera-stable', 'chrome', # 通用名称 ], } # 各平台常见安装路径 COMMON_PATHS = { 'windows': [ # Chrome os.path.expandvars(r'%ProgramFiles%\Google\Chrome\Application'), os.path.expandvars(r'%ProgramFiles(x86)%\Google\Chrome\Application'), os.path.expandvars(r'%LocalAppData%\Google\Chrome\Application'), # Edge os.path.expandvars(r'%ProgramFiles%\Microsoft\Edge\Application'), os.path.expandvars(r'%ProgramFiles(x86)%\Microsoft\Edge\Application'), os.path.expandvars(r'%LocalAppData%\Microsoft\Edge\Application'), # Brave os.path.expandvars(r'%ProgramFiles%\BraveSoftware\Brave-Browser\Application'), os.path.expandvars(r'%LocalAppData%\BraveSoftware\Brave-Browser\Application'), # Chromium os.path.expandvars(r'%LocalAppData%\Chromium\Application'), # Vivaldi os.path.expandvars(r'%LocalAppData%\Vivaldi\Application'), # Opera os.path.expandvars(r'%LocalAppData%\Programs\Opera'), # Playwright os.path.expandvars(r'%LocalAppData%\ms-playwright'), os.path.expandvars(r'%UserProfile%\.cache\ms-playwright'), # Puppeteer os.path.expandvars(r'%LocalAppData%\puppeteer'), os.path.expandvars(r'%UserProfile%\.cache\puppeteer'), ], 'darwin': [ # macOS # 标准应用目录 '/Applications', os.path.expanduser('~/Applications'), # Chrome '/Applications/Google Chrome.app/Contents/MacOS', os.path.expanduser('~/Applications/Google Chrome.app/Contents/MacOS'), # Chromium '/Applications/Chromium.app/Contents/MacOS', os.path.expanduser('~/Applications/Chromium.app/Contents/MacOS'), # Edge '/Applications/Microsoft Edge.app/Contents/MacOS', os.path.expanduser('~/Applications/Microsoft Edge.app/Contents/MacOS'), # Brave '/Applications/Brave Browser.app/Contents/MacOS', os.path.expanduser('~/Applications/Brave Browser.app/Contents/MacOS'), # Vivaldi '/Applications/Vivaldi.app/Contents/MacOS', # Opera '/Applications/Opera.app/Contents/MacOS', # Homebrew '/opt/homebrew/bin', '/usr/local/bin', '/opt/homebrew/Caskroom/google-chrome/latest/Google Chrome.app/Contents/MacOS', # Playwright os.path.expanduser('~/.cache/ms-playwright'), os.path.expanduser('~/Library/Caches/ms-playwright'), # Puppeteer os.path.expanduser('~/.cache/puppeteer'), ], 'linux': [ # 标准路径 '/usr/bin', '/usr/local/bin', '/bin', '/sbin', # Chrome '/opt/google/chrome', '/opt/google/chrome-stable', '/opt/google/chrome-beta', '/opt/google/chrome-unstable', # Chromium '/opt/chromium', '/opt/chromium-browser', '/usr/lib/chromium', '/usr/lib/chromium-browser', '/usr/lib64/chromium-browser', # Edge '/opt/microsoft/msedge', '/opt/microsoft/msedge-stable', '/opt/microsoft/msedge-beta', '/opt/microsoft/msedge-dev', # Brave '/opt/brave.com/brave', '/opt/brave.com/brave-browser', '/usr/lib/brave-browser', # Snap 安装路径 '/snap/bin', '/snap/chromium/current/usr/lib/chromium-browser', '/snap/chromium/current/usr/lib/chromium', '/var/lib/snapd/snap/bin', # Flatpak 路径 '/var/lib/flatpak/exports/bin', os.path.expanduser('~/.local/share/flatpak/exports/bin'), # AppImage 路径 os.path.expanduser('~/Applications'), os.path.expanduser('~/.local/bin'), # Playwright os.path.expanduser('~/.cache/ms-playwright'), '/root/.cache/ms-playwright', '/home/*/.cache/ms-playwright', # Puppeteer os.path.expanduser('~/.cache/puppeteer'), '/root/.cache/puppeteer', '/home/*/.cache/puppeteer', # Docker 常见路径 '/headless-shell', '/chrome', '/chromium', '/browser', '/app/chrome', '/app/chromium', '/usr/share/chromium', # NixOS '/run/current-system/sw/bin', os.path.expanduser('~/.nix-profile/bin'), ], } # Playwright/Puppeteer 浏览器的 glob 模式 GLOB_PATTERNS = [ # Playwright Chromium os.path.expanduser('~/.cache/ms-playwright/chromium-*/chrome-linux/chrome'), os.path.expanduser('~/.cache/ms-playwright/chromium-*/chrome-mac/Chromium.app/Contents/MacOS/Chromium'), os.path.expanduser('~/.cache/ms-playwright/chromium-*/chrome-win/chrome.exe'), '/root/.cache/ms-playwright/chromium-*/chrome-linux/chrome', '/home/*/.cache/ms-playwright/chromium-*/chrome-linux/chrome', # Playwright Chrome os.path.expanduser('~/.cache/ms-playwright/chrome-*/chrome-linux/chrome'), os.path.expanduser('~/.cache/ms-playwright/chrome-*/chrome-mac/Google Chrome for Testing.app/Contents/MacOS/Google Chrome for Testing'), '/root/.cache/ms-playwright/chrome-*/chrome-linux/chrome', # Puppeteer os.path.expanduser('~/.cache/puppeteer/chrome/*/chrome-linux64/chrome'), os.path.expanduser('~/.cache/puppeteer/chrome/*/chrome-mac-x64/Google Chrome for Testing.app/Contents/MacOS/Google Chrome for Testing'), os.path.expanduser('~/.cache/puppeteer/chrome/*/chrome-win64/chrome.exe'), '/root/.cache/puppeteer/chrome/*/chrome-linux64/chrome', '/home/*/.cache/puppeteer/chrome/*/chrome-linux64/chrome', # Windows Playwright os.path.expandvars(r'%LocalAppData%\ms-playwright\chromium-*\chrome-win\chrome.exe'), os.path.expandvars(r'%UserProfile%\.cache\ms-playwright\chromium-*\chrome-win\chrome.exe'), ] @classmethod def get_platform(cls) -> str: """获取当前平台""" system = platform.system().lower() if system == 'windows': return 'windows' elif system == 'darwin': return 'darwin' else: return 'linux' # Linux 和其他 Unix-like 系统 @classmethod def find_browser(cls) -> Optional[str]: """ 查找可用的浏览器路径 返回找到的第一个可执行浏览器路径，未找到返回None """ current_platform = cls.get_platform() # 1. 首先检查环境变量 for env_var in ['BROWSER_PATH', 'CHROME_PATH', 'CHROMIUM_PATH', 'CHROME_EXECUTABLE_PATH']: env_path = os.getenv(env_var) if env_path and cls._is_valid_executable(env_path): return env_path # 2. 使用 which/where 命令查找 executables = cls.BROWSER_EXECUTABLES.get(current_platform, cls.BROWSER_EXECUTABLES['linux']) for name in executables: path = shutil.which(name) if path and cls._is_valid_executable(path): return path # 3. 检查 glob 模式（Playwright/Puppeteer） for pattern in cls.GLOB_PATTERNS: matches = glob.glob(pattern) for match in sorted(matches, reverse=True): # 优先使用最新版本 if cls._is_valid_executable(match): return match # 4. 遍历常见路径 common_paths = cls.COMMON_PATHS.get(current_platform, cls.COMMON_PATHS['linux']) for base_path in common_paths: # 处理包含通配符的路径 if '*' in base_path: expanded_paths = glob.glob(base_path) else: expanded_paths = [base_path] for expanded_path in expanded_paths: if not os.path.isdir(expanded_path): continue for name in executables: full_path = os.path.join(expanded_path, name) if cls._is_valid_executable(full_path): return full_path # 5. macOS 特殊处理：检查 .app 包 if current_platform == 'darwin': found = cls._find_macos_app() if found: return found # 6. 递归搜索特定目录 search_roots = { 'windows': [ os.path.expandvars(r'%ProgramFiles%'), os.path.expandvars(r'%LocalAppData%'), ], 'darwin': [ '/Applications', os.path.expanduser('~/Applications'), ], 'linux': [ '/opt', '/usr/lib', '/snap', os.path.expanduser('~/.cache'), ], } for search_root in search_roots.get(current_platform, []): if os.path.isdir(search_root): found = cls._recursive_search(search_root, executables, max_depth=4) if found: return found # 7. Linux: 使用 find 命令（最后手段） if current_platform == 'linux': found = cls._find_with_command(executables) if found: return found return None @classmethod def _is_valid_executable(cls, path: str) -> bool: """检查路径是否是有效的可执行文件""" if not path: return False # 处理 Windows 路径 path = os.path.normpath(path) if not os.path.isfile(path): return False # Windows 不需要检查执行权限 if cls.get_platform() == 'windows': return path.lower().endswith('.exe') return os.access(path, os.X_OK) @classmethod def _find_macos_app(cls) -> Optional[str]: """macOS 特殊处理：查找 .app 包""" app_dirs = ['/Applications', os.path.expanduser('~/Applications')] app_names = [ ('Google Chrome.app', 'Contents/MacOS/Google Chrome'), ('Chromium.app', 'Contents/MacOS/Chromium'), ('Microsoft Edge.app', 'Contents/MacOS/Microsoft Edge'), ('Brave Browser.app', 'Contents/MacOS/Brave Browser'), ('Vivaldi.app', 'Contents/MacOS/Vivaldi'), ('Opera.app', 'Contents/MacOS/Opera'), ] for app_dir in app_dirs: for app_name, executable_path in app_names: full_path = os.path.join(app_dir, app_name, executable_path) if cls._is_valid_executable(full_path): return full_path return None @classmethod def _recursive_search(cls, root: str, executables: List[str], max_depth: int = 3, current_depth: int = 0) -> Optional[str]: """递归搜索目录""" if current_depth >= max_depth: return None # 跳过的目录 skip_dirs = {'.git', 'node_modules', '__pycache__', 'cache', 'tmp', 'temp', 'logs', 'log', '.npm', '.yarn', 'site-packages', 'dist-packages'} try: for entry in os.scandir(root): try: if entry.is_file(follow_symlinks=True): if entry.name in executables or entry.name.lower() in [e.lower() for e in executables]: if cls._is_valid_executable(entry.path): return entry.path elif entry.is_dir(follow_symlinks=False): if entry.name.lower() in skip_dirs: continue found = cls._recursive_search(entry.path, executables, max_depth, current_depth + 1) if found: return found except (PermissionError, OSError): continue except (PermissionError, OSError): pass return None @classmethod def _find_with_command(cls, executables: List[str]) -> Optional[str]: """使用系统命令查找浏览器（Linux）""" for name in executables[:5]: # 只检查前几个常见的 try: result = subprocess.run( ['find', '/', '-maxdepth', '6', '-name', name, '-type', 'f', '-executable', '-print', '-quit'], capture_output=True, text=True, timeout=30, stderr=subprocess.DEVNULL ) if result.returncode == 0 and result.stdout.strip(): path = result.stdout.strip().split('\n')[0] if cls._is_valid_executable(path): return path except (subprocess.TimeoutExpired, FileNotFoundError, Exception): continue return None @classmethod def get_browser_info(cls) -> Dict: """获取浏览器信息""" path = cls.find_browser() info = { 'found': path is not None, 'path': path, 'version': None, 'platform': cls.get_platform() } return info @classmethod def find_all_browsers(cls) -> List[Dict]: """查找所有可用的浏览器""" browsers = [] seen_paths = set() current_platform = cls.get_platform() executables = cls.BROWSER_EXECUTABLES.get(current_platform, cls.BROWSER_EXECUTABLES['linux']) # 查找所有可能的浏览器 for name in executables: path = shutil.which(name) if path and path not in seen_paths and cls._is_valid_executable(path): seen_paths.add(path) browsers.append({'name': name, 'path': path}) # 检查 glob 模式 for pattern in cls.GLOB_PATTERNS: matches = glob.glob(pattern) for match in matches: if match not in seen_paths and cls._is_valid_executable(match): seen_paths.add(match) browsers.append({'name': os.path.basename(match), 'path': match}) return browsers class Config: """配置管理器""" def __init__(self): self._lock = threading.Lock() self._browser_path = None self._user_agent = None self._max_workers = 1 self._headless = True self._email_configs = [] self._browser_fingerprint = {} self._load_from_env() self._detect_browser() def _detect_browser(self): """检测浏览器路径""" import os # 优先使用环境变量 env_path = os.getenv('BROWSER_PATH') or os.getenv('CHROME_PATH') if env_path and os.path.isfile(env_path): self._browser_path = env_path logger.info(f"[Config] 使用环境变量指定的浏览器: {env_path}") return # 自动检测 # 自动检测 logger.info(f"[Config] 正在自动检测浏览器路径 (平台: {BrowserPathFinder.get_platform()})...") browser_info = BrowserPathFinder.get_browser_info() if browser_info['found']: self._browser_path = browser_info['path'] if browser_info['found']: self._browser_path = browser_info['path'] logger.info(f"[Config] ✓ 检测到浏览器: {browser_info['path']}") if browser_info['version']: logger.info(f"[Config] ✓ 浏览器版本: {browser_info['version']}") else: logger.warning("[Config] ✗ 警告: 未检测到可用的浏览器！") logger.warning("[Config] 请确保已安装 Chrome/Chromium/Edge 或设置 BROWSER_PATH 环境变量") logger.warning("[Config] 支持的浏览器: Chrome, Chromium, Edge, Brave, Vivaldi, Opera") # 列出所有检查过的路径（调试用） all_browsers = BrowserPathFinder.find_all_browsers() if all_browsers: logger.info(f"[Config] 找到以下浏览器但可能不兼容:") for b in all_browsers: logger.info(f"[Config] - {b['name']}: {b['path']}") def get_browser_path(self) -> Optional[str]: """获取浏览器路径""" with self._lock: return self._browser_path def set_browser_path(self, path: str): """设置浏览器路径""" with self._lock: if os.path.isfile(path): self._browser_path = path else: raise ValueError(f"无效的浏览器路径: {path}") def _load_from_env(self): """从环境变量加载配置""" # User-Agent self._user_agent = os.getenv( 'USER_AGENT', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36' ) # 最大工作线程数 self._max_workers = int(os.getenv('MAX_WORKERS', '1')) # 无头模式 self._headless = os.getenv('HEADLESS', 'true').lower() == 'true' # 邮箱配置（支持多个，用分号分隔） self._email_configs = [] worker_domains = os.getenv('WORKER_DOMAINS', '').split(';') email_domains = os.getenv('EMAIL_DOMAINS', '').split(';') admin_passwords = os.getenv('ADMIN_PASSWORDS', '').split(';') # 确保三个列表长度一致 max_len = max(len(worker_domains), len(email_domains), len(admin_passwords)) for i in range(max_len): worker = worker_domains[i].strip() if i < len(worker_domains) else '' email = email_domains[i].strip() if i < len(email_domains) else '' password = admin_passwords[i].strip() if i < len(admin_passwords) else '' if worker and email and password: self._email_configs.append({ 'worker_domain': worker, 'email_domain': email, 'admin_password': password }) # 浏览器指纹配置 self._browser_fingerprint = { 'window_size': os.getenv('WINDOW_SIZE', '1920x1080'), 'timezone': os.getenv('TIMEZONE', 'Asia/Shanghai'), 'locale': os.getenv('LOCALE', 'zh-CN'), 'platform': os.getenv('PLATFORM', 'Win64'), 'color_depth': int(os.getenv('COLOR_DEPTH', '24')), 'device_memory': int(os.getenv('DEVICE_MEMORY', '8')), 'hardware_concurrency': int(os.getenv('HARDWARE_CONCURRENCY', '8')) } def get_user_agent(self) -> str: with self._lock: return self._user_agent def set_user_agent(self, ua: str): with self._lock: self._user_agent = ua def get_max_workers(self) -> int: with self._lock: return self._max_workers def set_max_workers(self, count: int): with self._lock: self._max_workers = max(1, min(count, 10)) def get_headless(self) -> bool: with self._lock: return self._headless def set_headless(self, headless: bool): with self._lock: self._headless = headless def get_email_configs(self) -> List[Dict]: with self._lock: return self._email_configs.copy() def get_email_configs_safe(self) -> List[Dict]: """获取邮箱配置（隐藏密码）""" with self._lock: return [ { 'worker_domain': c['worker_domain'], 'email_domain': c['email_domain'], 'admin_password': '***' } for c in self._email_configs ] def get_random_email_config(self) -> Optional[Dict]: with self._lock: if self._email_configs: return random.choice(self._email_configs) return None def add_email_config(self, worker_domain: str, email_domain: str, admin_password: str): with self._lock: self._email_configs.append({ 'worker_domain': worker_domain, 'email_domain': email_domain, 'admin_password': admin_password }) def update_email_config(self, index: int, worker_domain: str = None, email_domain: str = None, admin_password: str = None): with self._lock: if 0 <= index < len(self._email_configs): if worker_domain: self._email_configs[index]['worker_domain'] = worker_domain if email_domain: self._email_configs[index]['email_domain'] = email_domain if admin_password: self._email_configs[index]['admin_password'] = admin_password else: raise IndexError("配置索引不存在") def delete_email_config(self, index: int): with self._lock: if 0 <= index < len(self._email_configs): del self._email_configs[index] else: raise IndexError("配置索引不存在") def get_browser_fingerprint(self) -> Dict: with self._lock: return self._browser_fingerprint.copy() def set_browser_fingerprint(self, fingerprint: Dict): with self._lock: self._browser_fingerprint.update(fingerprint)