304 lines
10 KiB
Python
304 lines
10 KiB
Python
# -*- 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
|
||
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 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)
|
||
|
||
# 独立 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()
|
||
|
||
# 与 profile_shortcut 同一套人用参数：独立 user-data-dir 保存登录态
|
||
cmd = [browser_exe, *human_chrome_launch_args(profile_abs, url)]
|
||
|
||
log.info(
|
||
"browser_open_launch exe=%s profile_dir=%s url=%s platform=%s restore_on_startup=%s",
|
||
browser_exe,
|
||
profile_abs,
|
||
url,
|
||
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,
|
||
)
|
||
)
|