373 lines
12 KiB
Python
373 lines
12 KiB
Python
"""Chrome launcher helpers for per-account CDP sessions."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from urllib.parse import urlsplit
|
||
|
||
from . import appconfig
|
||
from . import config as account_config
|
||
|
||
|
||
CDP_HOST = "127.0.0.1"
|
||
CHROME_APP_PATHS_KEY = r"SOFTWARE\Microsoft\Windows\CurrentVersion\App Paths\chrome.exe"
|
||
|
||
|
||
class ChromeLaunchError(RuntimeError):
|
||
"""Raised when Chrome launch configuration is invalid."""
|
||
|
||
|
||
def _normalize_chrome_path(value) -> str:
|
||
"""Return a filesystem-ready Chrome path, or an empty string."""
|
||
|
||
text = str(value or "").strip()
|
||
if len(text) >= 2 and text[0] == text[-1] and text[0] in {"\"", "'"}:
|
||
text = text[1:-1].strip()
|
||
if not text:
|
||
return ""
|
||
text = os.path.expandvars(os.path.expanduser(text))
|
||
return os.path.abspath(text)
|
||
|
||
|
||
def _load_winreg():
|
||
"""Return winreg only on Windows so non-Windows callers stay harmless."""
|
||
|
||
if os.name != "nt":
|
||
return None
|
||
try:
|
||
import winreg
|
||
except ImportError:
|
||
return None
|
||
return winreg
|
||
|
||
|
||
def _registry_view_flags(winreg):
|
||
flags = [0]
|
||
for name in ("KEY_WOW64_64KEY", "KEY_WOW64_32KEY"):
|
||
value = getattr(winreg, name, 0)
|
||
if value and value not in flags:
|
||
flags.append(value)
|
||
return flags
|
||
|
||
|
||
def _close_registry_key(key):
|
||
close = getattr(key, "Close", None) or getattr(key, "close", None)
|
||
if callable(close):
|
||
try:
|
||
close()
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def _registry_chrome_candidates():
|
||
"""Yield Chrome App Paths values without exposing registry failures."""
|
||
|
||
winreg = _load_winreg()
|
||
if winreg is None:
|
||
return
|
||
|
||
key_read = getattr(winreg, "KEY_READ", 0)
|
||
hives = (
|
||
getattr(winreg, "HKEY_CURRENT_USER", None),
|
||
getattr(winreg, "HKEY_LOCAL_MACHINE", None),
|
||
)
|
||
seen = set()
|
||
for hive in hives:
|
||
if hive is None:
|
||
continue
|
||
for view_flag in _registry_view_flags(winreg):
|
||
marker = (hive, view_flag)
|
||
if marker in seen:
|
||
continue
|
||
seen.add(marker)
|
||
key = None
|
||
try:
|
||
key = winreg.OpenKey(
|
||
hive,
|
||
CHROME_APP_PATHS_KEY,
|
||
0,
|
||
key_read | view_flag,
|
||
)
|
||
value, _value_type = winreg.QueryValueEx(key, "")
|
||
except (AttributeError, OSError):
|
||
continue
|
||
finally:
|
||
if key is not None:
|
||
_close_registry_key(key)
|
||
if value:
|
||
yield value
|
||
|
||
|
||
def _standard_chrome_candidates():
|
||
"""Yield the supported Windows Chrome install locations."""
|
||
|
||
suffix = ("Google", "Chrome", "Application", "chrome.exe")
|
||
for variable in ("ProgramFiles", "ProgramFiles(x86)", "LOCALAPPDATA"):
|
||
base = os.environ.get(variable, "").strip()
|
||
if base:
|
||
yield os.path.join(base, *suffix)
|
||
|
||
|
||
def is_configured_chrome_path_valid(value) -> bool:
|
||
"""Return whether a configured Chrome executable can be launched safely."""
|
||
|
||
normalized = _normalize_chrome_path(value)
|
||
if normalized and os.path.isfile(normalized):
|
||
return True
|
||
text = str(value or "").strip().strip("\"")
|
||
return text.lower() == "chrome.exe" and bool(shutil.which(text))
|
||
|
||
|
||
def detect_chrome_path() -> str:
|
||
"""Find an installed Google Chrome executable without probing other browsers."""
|
||
|
||
for candidate in _registry_chrome_candidates():
|
||
normalized = _normalize_chrome_path(candidate)
|
||
if normalized and os.path.isfile(normalized):
|
||
return normalized
|
||
for candidate in _standard_chrome_candidates():
|
||
normalized = _normalize_chrome_path(candidate)
|
||
if normalized and os.path.isfile(normalized):
|
||
return normalized
|
||
path_candidate = shutil.which("chrome.exe")
|
||
normalized = _normalize_chrome_path(path_candidate)
|
||
if normalized and os.path.isfile(normalized):
|
||
return normalized
|
||
return ""
|
||
|
||
|
||
def ensure_configured_chrome_path(config=None, config_path=None) -> dict:
|
||
"""Persist an auto-detected path only when the configured path is unusable."""
|
||
|
||
path = config_path or (config or {}).get("config_path") or appconfig.CONFIG_PATH
|
||
current = appconfig.load_config(path) if config is None else config
|
||
configured_path = appconfig.chrome_path(current)
|
||
if is_configured_chrome_path_valid(configured_path):
|
||
return {
|
||
"changed": False,
|
||
"config": current,
|
||
"path": configured_path,
|
||
"message": "",
|
||
}
|
||
|
||
detected_path = detect_chrome_path()
|
||
if not detected_path:
|
||
return {
|
||
"changed": False,
|
||
"config": current,
|
||
"path": configured_path,
|
||
"message": "",
|
||
}
|
||
|
||
updated = dict(current)
|
||
updated["chrome_path"] = detected_path
|
||
saved = appconfig.save_config(updated, path=path)
|
||
return {
|
||
"changed": True,
|
||
"config": saved,
|
||
"path": detected_path,
|
||
"message": f"已自动定位 Chrome：{detected_path}",
|
||
}
|
||
|
||
|
||
def _account_value(account, field, default=None):
|
||
if isinstance(account, dict):
|
||
return account.get(field, default)
|
||
return getattr(account, field, default)
|
||
|
||
|
||
def _required_account_value(account, field):
|
||
value = _account_value(account, field)
|
||
if value is None or str(value).strip() == "":
|
||
raise ChromeLaunchError(f"账号缺少字段: {field}")
|
||
return value
|
||
|
||
|
||
def _debug_port(account) -> int:
|
||
value = int(_required_account_value(account, "debug_port"))
|
||
if value <= 0 or value > 65535:
|
||
raise ChromeLaunchError("debug_port 必须在 1-65535 范围内")
|
||
return value
|
||
|
||
|
||
def _user_data_dir(account, config=None) -> str:
|
||
existing = _account_value(account, "user_data_dir")
|
||
if existing:
|
||
path = os.path.abspath(str(existing))
|
||
os.makedirs(path, exist_ok=True)
|
||
return path
|
||
|
||
slug = _account_value(account, "slug")
|
||
if not slug:
|
||
slug = account_config.make_slug(_required_account_value(account, "alias"))
|
||
return account_config.ensure_user_data_dir(slug, config=config)
|
||
|
||
|
||
def _validated_initial_url(value):
|
||
text = str(value or "").strip()
|
||
if not text:
|
||
return None
|
||
try:
|
||
parsed = urlsplit(text)
|
||
except ValueError as exc:
|
||
raise ChromeLaunchError("Chrome 初始页面必须是有效的蝦皮 http/https 地址") from exc
|
||
hostname = str(parsed.hostname or "").lower()
|
||
is_shopee_host = hostname.startswith("shopee.") or ".shopee." in hostname
|
||
if (
|
||
parsed.scheme.lower() not in {"http", "https"}
|
||
or not is_shopee_host
|
||
or parsed.username
|
||
or parsed.password
|
||
):
|
||
raise ChromeLaunchError("Chrome 初始页面必须是有效的蝦皮 http/https 地址")
|
||
return text
|
||
|
||
|
||
def build_launch_args(account, config=None, initial_url=None) -> list:
|
||
"""Build Chrome command arguments for one account."""
|
||
|
||
chrome_path = appconfig.chrome_path(config)
|
||
if not chrome_path:
|
||
raise ChromeLaunchError("chrome_path 不能为空")
|
||
port = _debug_port(account)
|
||
user_data_dir = _user_data_dir(account, config=config)
|
||
args = [
|
||
chrome_path,
|
||
f"--remote-debugging-port={port}",
|
||
"--remote-allow-origins=*",
|
||
f"--user-data-dir={user_data_dir}",
|
||
]
|
||
startup_url = _validated_initial_url(initial_url)
|
||
if startup_url:
|
||
args.extend(("--no-first-run", "--no-default-browser-check"))
|
||
args.append(startup_url)
|
||
return args
|
||
|
||
|
||
def _safe_shortcut_name(value) -> str:
|
||
text = str(value or "").strip()
|
||
if not text:
|
||
raise ChromeLaunchError("快捷方式名称不能为空")
|
||
text = re.sub(r'[<>:"/\\|?*\x00-\x1f]+', "_", text)
|
||
text = re.sub(r"\s+", " ", text).strip(" .")
|
||
if not text:
|
||
raise ChromeLaunchError("快捷方式名称不能为空")
|
||
return text
|
||
|
||
|
||
def _default_shortcut_name(account) -> str:
|
||
alias = _account_value(account, "alias")
|
||
account_name = _account_value(account, "account_name")
|
||
name = alias or account_name or _required_account_value(account, "slug")
|
||
return _safe_shortcut_name(f"cmshopee-{name}")
|
||
|
||
|
||
def _desktop_dir() -> str:
|
||
return os.path.join(os.path.expanduser("~"), "Desktop")
|
||
|
||
|
||
def _ps_single_quote(value) -> str:
|
||
return "'" + str(value).replace("'", "''") + "'"
|
||
|
||
|
||
def create_shortcut(account, shortcut_path=None, desktop_dir=None, name=None, config=None) -> str:
|
||
"""Create a Windows .lnk shortcut for one account's Chrome session."""
|
||
|
||
args = build_launch_args(account, config=config)
|
||
chrome_path = os.path.abspath(args[0])
|
||
arguments = subprocess.list2cmdline(args[1:])
|
||
if shortcut_path is None:
|
||
root = os.path.abspath(desktop_dir or _desktop_dir())
|
||
shortcut_name = _safe_shortcut_name(name) if name else _default_shortcut_name(account)
|
||
shortcut_path = os.path.join(root, shortcut_name + ".lnk")
|
||
shortcut_path = os.path.abspath(str(shortcut_path))
|
||
if not shortcut_path.lower().endswith(".lnk"):
|
||
shortcut_path += ".lnk"
|
||
os.makedirs(os.path.dirname(shortcut_path), exist_ok=True)
|
||
|
||
working_directory = os.path.dirname(chrome_path) or os.getcwd()
|
||
script = "; ".join(
|
||
[
|
||
"$shell = New-Object -ComObject WScript.Shell",
|
||
f"$shortcut = $shell.CreateShortcut({_ps_single_quote(shortcut_path)})",
|
||
f"$shortcut.TargetPath = {_ps_single_quote(chrome_path)}",
|
||
f"$shortcut.Arguments = {_ps_single_quote(arguments)}",
|
||
f"$shortcut.WorkingDirectory = {_ps_single_quote(working_directory)}",
|
||
f"$shortcut.IconLocation = {_ps_single_quote(chrome_path)}",
|
||
"$shortcut.Save()",
|
||
]
|
||
)
|
||
try:
|
||
subprocess.run(
|
||
[
|
||
"powershell",
|
||
"-NoProfile",
|
||
"-ExecutionPolicy",
|
||
"Bypass",
|
||
"-Command",
|
||
script,
|
||
],
|
||
check=True,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
except OSError as exc:
|
||
raise ChromeLaunchError(f"生成快捷方式失败: {exc}") from exc
|
||
except subprocess.CalledProcessError as exc:
|
||
error = (exc.stderr or exc.stdout or str(exc)).strip()
|
||
raise ChromeLaunchError(f"生成快捷方式失败: {error}") from exc
|
||
return shortcut_path
|
||
|
||
|
||
def launch_chrome(account, config=None, initial_url=None) -> subprocess.Popen:
|
||
"""Launch Chrome for one account and return the process handle."""
|
||
|
||
args = build_launch_args(account, config=config, initial_url=initial_url)
|
||
try:
|
||
return subprocess.Popen(args)
|
||
except OSError as exc:
|
||
raise ChromeLaunchError(f"启动 Chrome 失败: {exc}") from exc
|
||
|
||
|
||
def _json_version_url(port, host=CDP_HOST):
|
||
return f"http://{host}:{int(port)}/json/version"
|
||
|
||
|
||
def _fetch_json_version(port, host=CDP_HOST, timeout=1.0):
|
||
request = urllib.request.Request(_json_version_url(port, host=host), method="GET")
|
||
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||
with opener.open(request, timeout=timeout) as response:
|
||
if response.status != 200:
|
||
raise ChromeLaunchError(f"CDP 返回状态码: {response.status}")
|
||
body = response.read()
|
||
return json.loads(body.decode("utf-8"))
|
||
|
||
|
||
def is_running(port, host=CDP_HOST) -> bool:
|
||
"""Return whether a Chrome CDP endpoint responds on /json/version."""
|
||
|
||
try:
|
||
_fetch_json_version(port, host=host, timeout=1.0)
|
||
return True
|
||
except (OSError, urllib.error.URLError, json.JSONDecodeError, ChromeLaunchError):
|
||
return False
|
||
|
||
|
||
def wait_debug_ready(port, timeout=60, host=CDP_HOST) -> bool:
|
||
"""Poll until the Chrome CDP endpoint is ready or timeout expires."""
|
||
|
||
deadline = time.monotonic() + float(timeout)
|
||
while True:
|
||
if is_running(port, host=host):
|
||
return True
|
||
if time.monotonic() >= deadline:
|
||
return False
|
||
time.sleep(0.25)
|