account-manager/scripts/service/browser_service.py

# -*- coding: utf-8 -*-
"""
Chromium/Edge browser profile opener.

Uses system Chrome/Edge channel, Playwright launch_persistent_context.
Does NOT download Playwright Chromium.
Does NOT check login status.
Does NOT write secret to SQLite.
"""
import json
import os
import subprocess
import sys
import tempfile

from db.connection import get_conn, init_db
from db.accounts_repo import get_account_by_id, get_platform_from_db
from util.logging_config import (
    get_skill_logger,
    subprocess_env_with_trace,
)
from util.platforms import PLATFORMS

# 子进程入口脚本与本模块同目录（独立解释器执行，非 import）
_SERVICE_DIR = os.path.dirname(os.path.abspath(__file__))
_SCRIPTS_ROOT = os.path.dirname(_SERVICE_DIR)


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"


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_open(account_id):
    """
    打开该账号的持久化浏览器，仅用于肉眼核对。
    不写数据库、不判定是否已登录。
    """
    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")
        return

    channel = resolve_chromium_channel()
    if not channel:
        _print_browser_install_hint()
        return

    try:
        import playwright  # noqa: F401
    except ImportError:
        print(
            "ERROR:需要 playwright Python 包。"
        )
        return

    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 为空。")
        return
    os.makedirs(profile_dir, exist_ok=True)

    conn = get_conn()
    try:
        url = resolve_account_browser_url(target, conn=conn)
    finally:
        conn.close()

    log.info(
        "browser_open_context profile_dir=%s url=%s channel=%s platform=%s",
        profile_dir,
        url,
        channel,
        target.get("platform_key"),
    )

    browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge"
    print(f"正在打开 [{target['account_label']}] 的 {browser_name}（仅查看，不写入数据库）")
    print(f"地址：{url}")
    print("本命令不做登录检测。")

    cfg = {"channel": channel, "profile_dir": profile_dir, "url": url}
    with tempfile.NamedTemporaryFile(
        mode="w", suffix=".json", delete=False, encoding="utf-8"
    ) as jf:
        json.dump(cfg, jf, ensure_ascii=False)
        cfg_path = jf.name
    try:
        subprocess.run(
            [sys.executable, "-m", "service.open_child_runner", cfg_path],
            cwd=_SCRIPTS_ROOT,
            env=subprocess_env_with_trace(),
        )
        log.info("browser_open_close account_id=%s channel=%s", account_id, channel)
    except Exception as e:
        log.error("browser_open_error account_id=%s error=%s", account_id, str(e))
        raise
    finally:
        try:
            os.unlink(cfg_path)
        except OSError:
            pass