# -*- coding: utf-8 -*- """ Chromium/Edge browser profile opener. Uses system Chrome/Edge executable with --user-data-dir (same profile as RPA shortcuts). Does NOT download Playwright Chromium. Does NOT check login status. Does NOT write secret to SQLite. """ import json import os import subprocess import sys from db.connection import get_conn, init_db from db.accounts_repo import get_account_by_id, get_platform_from_db from service.chrome_profile_prefs import ( ensure_session_restore, profile_has_restorable_session, ) from util.logging_config import get_skill_logger from util.platforms import PLATFORMS def _win_find_exe(candidates): for p in candidates: if p and os.path.isfile(p): return p return None def resolve_chromium_executable(): """ Windows 上返回 chrome.exe 或 msedge.exe 的绝对路径；未找到或非 Windows 返回 None。 """ if sys.platform != "win32": return None pf = os.environ.get("ProgramFiles", r"C:\Program Files") pfx86 = os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)") local = os.environ.get("LocalAppData", "") chrome = _win_find_exe( [ os.path.join(pf, "Google", "Chrome", "Application", "chrome.exe"), os.path.join(pfx86, "Google", "Chrome", "Application", "chrome.exe"), os.path.join(local, "Google", "Chrome", "Application", "chrome.exe") if local else "", ] ) if chrome: return chrome edge = _win_find_exe( [ os.path.join(pfx86, "Microsoft", "Edge", "Application", "msedge.exe"), os.path.join(pf, "Microsoft", "Edge", "Application", "msedge.exe"), os.path.join(local, "Microsoft", "Edge", "Application", "msedge.exe") if local else "", ] ) if edge: return edge return None def resolve_chromium_channel(): """ Playwright channel: 'chrome' | 'msedge' | None Windows 优先 Chrome，其次 Edge；其它系统默认 chrome（由 Playwright 解析 PATH）。 """ if sys.platform != "win32": return "chrome" exe = resolve_chromium_executable() if not exe: return None if os.path.basename(exe).lower() == "msedge.exe": return "msedge" return "chrome" # 给人核对/手工登录用：跳过 Chrome 首次欢迎页并最大化。不要带 Playwright 的 --enable-automation。 HUMAN_CHROME_FLAGS = ( "--start-maximized", "--no-first-run", "--no-default-browser-check", "--disable-search-engine-choice-screen", ) def human_chrome_launch_args(profile_dir, url=None): """chrome.exe 之后的参数（Popen 列表形式，路径不额外加引号）。""" profile_abs = os.path.abspath(profile_dir) args = [f"--user-data-dir={profile_abs}", *HUMAN_CHROME_FLAGS] cleaned = (url or "").strip() if cleaned: args.append(cleaned) return args def human_chrome_shortcut_arguments(profile_dir, url=None): """Windows .lnk Arguments 字符串（user-data-dir 带引号）。""" profile_abs = os.path.abspath(profile_dir) parts = [f'--user-data-dir="{profile_abs}"', *HUMAN_CHROME_FLAGS] cleaned = (url or "").strip() if cleaned: parts.append(cleaned) return " ".join(parts) def _print_browser_install_hint(): print("❌ 未检测到 Google Chrome 或 Microsoft Edge，请先安装：") print(" • Chrome: https://www.google.com/chrome/") print(" • Edge: https://www.microsoft.com/zh-cn/edge") def resolve_account_browser_url(account: dict, conn=None) -> str: """account.url 优先，其次 platforms 表 default_url，最后内置 PLATFORMS。""" url = (account.get("url") or "").strip() if url: return url pk = (account.get("platform_key") or "").strip() if not pk: return "" close_conn = False if conn is None: init_db() conn = get_conn() close_conn = True try: plat = get_platform_from_db(conn, pk) if plat and (plat.get("default_url") or "").strip(): return plat["default_url"].strip() finally: if close_conn: conn.close() return (PLATFORMS.get(pk, {}).get("default_url") or "").strip() def cmd_reveal_profile(account_id): """ 在系统文件管理器中打开该账号的 profile_dir。 不写数据库、不启动浏览器。 """ log = get_skill_logger() init_db() log.info("reveal_profile_start account_id=%r", account_id) target = get_account_by_id(account_id) if not target: log.warning("reveal_profile_error account_id=%r not_found", account_id) print("ERROR:ACCOUNT_NOT_FOUND") sys.exit(1) profile_dir = (target.get("profile_dir") or "").strip() if not profile_dir: log.warning("reveal_profile_error account_id=%s profile_dir_missing", account_id) print("ERROR:PROFILE_DIR_MISSING 账号中 profile_dir 为空。") sys.exit(1) profile_abs = os.path.abspath(profile_dir) os.makedirs(profile_abs, exist_ok=True) try: if sys.platform == "win32": # explorer 即使成功也常返回非 0；用 startfile 打开目录更稳妥 os.startfile(profile_abs) # type: ignore[attr-defined] elif sys.platform == "darwin": subprocess.run(["open", profile_abs], check=False) else: subprocess.run(["xdg-open", profile_abs], check=False) except OSError as exc: log.error("reveal_profile_error account_id=%s error=%s", account_id, str(exc)) print(f"ERROR:REVEAL_PROFILE_FAILED {exc}") sys.exit(1) log.info("reveal_profile_ok account_id=%s profile_dir=%s", account_id, profile_abs) print( json.dumps( { "type": "result", "schema_version": 1, "status": "success", "message": "", "success": True, "data": { "account_id": int(target["id"]) if target.get("id") is not None else account_id, "profile_dir": profile_abs, }, }, ensure_ascii=False, ) ) def _list_profile_browser_pids(profile_dir: str): """找出已占用本 Profile 的 chrome/msedge PID（与采集技能同逻辑）。""" import json as _json profile_abs = os.path.abspath(profile_dir) needle = os.path.normcase(profile_abs) if sys.platform != "win32": return [] ps = ( "Get-CimInstance Win32_Process -Filter " "\"name='chrome.exe' or name='msedge.exe'\" |" " Select-Object ProcessId,CommandLine | ConvertTo-Json -Compress" ) try: completed = subprocess.run( ["powershell", "-NoProfile", "-Command", ps], capture_output=True, text=True, timeout=20, check=False, ) except Exception: return [] raw = (completed.stdout or "").strip() if not raw: return [] try: payload = _json.loads(raw) except Exception: return [] rows = payload if isinstance(payload, list) else [payload] pids = [] for row in rows: if not isinstance(row, dict): continue cmdline = str(row.get("CommandLine") or "") if needle not in os.path.normcase(cmdline): continue try: pids.append(int(row.get("ProcessId"))) except (TypeError, ValueError): continue return sorted(set(pids)) def cmd_open(account_id): """ 打开该账号的持久化浏览器，仅用于肉眼核对 / 手工登录 / 抓 DOM。 与 reveal-profile 同模式：本进程直接拉起系统 Chrome/Edge 后立即返回。 最大化、跳过 Chrome 首次欢迎页；不写数据库、不判定是否已登录、不起 Playwright。 """ log = get_skill_logger() init_db() log.info("browser_open_start account_id=%r", account_id) target = get_account_by_id(account_id) if not target: log.warning("browser_open_error account_id=%r not_found", account_id) print("ERROR:ACCOUNT_NOT_FOUND") sys.exit(1) browser_exe = resolve_chromium_executable() if not browser_exe: channel = resolve_chromium_channel() if sys.platform == "win32" or not channel: _print_browser_install_hint() sys.exit(1) # 非 Windows：退回 PATH 上的常见命令名 browser_exe = "google-chrome" if channel == "chrome" else "microsoft-edge" profile_dir = (target.get("profile_dir") or "").strip() if not profile_dir: log.warning("browser_open_error account_id=%s profile_dir_missing", account_id) print("ERROR:PROFILE_DIR_MISSING 账号中 profile_dir 为空。") sys.exit(1) profile_abs = os.path.abspath(profile_dir) os.makedirs(profile_abs, exist_ok=True) existing = _list_profile_browser_pids(profile_abs) if existing: log.info( "browser_open_reuse account_id=%s profile_dir=%s pids=%s", account_id, profile_abs, existing, ) print( json.dumps( { "type": "result", "schema_version": 1, "status": "success", "message": "已复用打开中的浏览器窗口", "success": True, "data": { "account_id": int(target["id"]) if target.get("id") is not None else account_id, "account_label": target.get("account_label"), "browser": "Google Chrome", "browser_exe": browser_exe, "profile_dir": profile_abs, "url": resolve_account_browser_url(target), "pid": existing[0], "reused": True, "pids": existing, }, }, ensure_ascii=False, ) ) return # 独立 Profile 默认不会「继续上次会话」，Session Cookie 关窗即丢；打开前写入 Preferences prefs_result = ensure_session_restore(profile_abs) log.info("browser_open_prefs %s", prefs_result) conn = get_conn() try: url = resolve_account_browser_url(target, conn=conn) finally: conn.close() # 有可恢复会话时不带 URL；首次/空 Profile 仍打开平台首页便于登录。 # 须识别现代 Chrome 的 Default/Sessions/Session_*（仅认旧文件名会导致每次都开首页冲掉 Session Cookie）。 has_session = profile_has_restorable_session(profile_abs) launch_url = None if has_session else url cmd = [browser_exe, *human_chrome_launch_args(profile_abs, launch_url)] log.info( "browser_open_launch exe=%s profile_dir=%s launch_url=%s has_session=%s " "platform=%s restore_on_startup=%s", browser_exe, profile_abs, launch_url or "(restore)", has_session, target.get("platform_key"), prefs_result.get("restore_after"), ) try: # GUI 浏览器：不挂控制台黑框；不等待关闭（与 reveal 一致立刻返回） popen_kwargs: dict = { "stdin": subprocess.DEVNULL, "stdout": subprocess.DEVNULL, "stderr": subprocess.DEVNULL, "close_fds": True, } if os.path.isabs(browser_exe): popen_kwargs["cwd"] = os.path.dirname(browser_exe) if sys.platform == "win32": popen_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP else: popen_kwargs["start_new_session"] = True child = subprocess.Popen(cmd, **popen_kwargs) except OSError as exc: log.error("browser_open_error account_id=%s error=%s", account_id, str(exc)) print(f"ERROR:BROWSER_OPEN_FAILED {exc}") sys.exit(1) browser_name = ( "Microsoft Edge" if os.path.basename(browser_exe).lower() == "msedge.exe" else "Google Chrome" ) log.info( "browser_open_ok account_id=%s browser=%s pid=%s", account_id, browser_name, child.pid, ) print( json.dumps( { "type": "result", "schema_version": 1, "status": "success", "message": "", "success": True, "data": { "account_id": int(target["id"]) if target.get("id") is not None else account_id, "account_label": target.get("account_label"), "browser": browser_name, "browser_exe": browser_exe, "profile_dir": profile_abs, "url": url, "pid": child.pid, "session_restore_on_startup": prefs_result.get("restore_after"), }, }, ensure_ascii=False, ) )