474 lines
14 KiB
Python
474 lines
14 KiB
Python
"""Account management services for the PySide6 accounts tab."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
from urllib.parse import urlparse
|
|
|
|
from . import appconfig, cdp, chrome, db, editor
|
|
from . import config as account_config
|
|
|
|
|
|
DEFAULT_REGION_HOST = editor.DEFAULT_REGION_HOST
|
|
INITIAL_LOGIN_TAB_WAIT_SECONDS = 2.0
|
|
INITIAL_LOGIN_TAB_POLL_SECONDS = 0.1
|
|
SAFE_STARTUP_PAGE_PREFIXES = (
|
|
"about:blank",
|
|
"chrome://newtab",
|
|
"chrome://new-tab-page",
|
|
"chrome://welcome",
|
|
"chrome://intro",
|
|
)
|
|
|
|
|
|
class AccountError(RuntimeError):
|
|
"""Raised when account management cannot complete an operation."""
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now().isoformat(timespec="seconds")
|
|
|
|
|
|
def _db_path(path=None, config=None) -> str:
|
|
return path or appconfig.db_path(config)
|
|
|
|
|
|
def _trim(value) -> str:
|
|
return str(value or "").strip()
|
|
|
|
|
|
def _optional_text(value) -> Optional[str]:
|
|
text = _trim(value)
|
|
return text or None
|
|
|
|
|
|
def _required_text(value, field_name: str) -> str:
|
|
text = _trim(value)
|
|
if not text:
|
|
raise AccountError(f"{field_name}不能为空")
|
|
return text
|
|
|
|
|
|
def _account_value(account, field, default=None):
|
|
if isinstance(account, dict):
|
|
return account.get(field, default)
|
|
return getattr(account, field, default)
|
|
|
|
|
|
def normalize_region_host(value=None) -> str:
|
|
text = _trim(value) or DEFAULT_REGION_HOST
|
|
if "://" in text:
|
|
text = urlparse(text).netloc
|
|
text = text.strip("/")
|
|
if "/" in text:
|
|
text = text.split("/", 1)[0]
|
|
return text or DEFAULT_REGION_HOST
|
|
|
|
|
|
def normalize_debug_port(value) -> int:
|
|
try:
|
|
port = int(value)
|
|
except (TypeError, ValueError) as exc:
|
|
raise AccountError("调试端口必须是数字") from exc
|
|
if port <= 0 or port > 65535:
|
|
raise AccountError("调试端口必须在 1-65535 范围内")
|
|
return port
|
|
|
|
|
|
def preview_user_data_dir(alias, config=None) -> str:
|
|
slug = account_config.make_slug(alias)
|
|
root = appconfig.user_data_root(config)
|
|
return os.path.abspath(os.path.join(str(root), slug))
|
|
|
|
|
|
def mask_password(password) -> str:
|
|
return "******" if password else ""
|
|
|
|
|
|
def next_debug_port(config=None, path=None) -> int:
|
|
cfg = appconfig.load_config() if config is None else config
|
|
start, end = appconfig.debug_port_range(cfg)
|
|
database_path = _db_path(path, cfg)
|
|
db.init_db(database_path)
|
|
used = {int(account.debug_port) for account in db.list_accounts(path=database_path)}
|
|
for port in range(start, end + 1):
|
|
if port not in used:
|
|
return port
|
|
return end + 1
|
|
|
|
|
|
def _assert_debug_port_available(port, database_path, ignore_alias=None) -> None:
|
|
used_by = [
|
|
account.alias
|
|
for account in db.list_accounts(path=database_path)
|
|
if int(account.debug_port) == int(port) and account.alias != ignore_alias
|
|
]
|
|
if used_by:
|
|
raise AccountError(f"调试端口 {port} 已被账号 {used_by[0]} 使用")
|
|
|
|
|
|
def list_accounts(path=None, config=None):
|
|
database_path = _db_path(path, config)
|
|
db.init_db(database_path)
|
|
return db.list_accounts(path=database_path)
|
|
|
|
|
|
def get_account(alias, path=None, config=None):
|
|
database_path = _db_path(path, config)
|
|
db.init_db(database_path)
|
|
account = db.get_account_by_alias(alias, path=database_path)
|
|
if account is None:
|
|
raise AccountError(f"账号不存在: {alias}")
|
|
return account
|
|
|
|
|
|
def create_account(
|
|
account_name,
|
|
alias,
|
|
region_host=None,
|
|
debug_port=None,
|
|
password=None,
|
|
note=None,
|
|
path=None,
|
|
config=None,
|
|
):
|
|
cfg = appconfig.load_config() if config is None else config
|
|
database_path = _db_path(path, cfg)
|
|
db.init_db(database_path)
|
|
|
|
account_name = _required_text(account_name, "账号名")
|
|
alias = _required_text(alias, "别名")
|
|
region_host = normalize_region_host(region_host)
|
|
port = normalize_debug_port(
|
|
next_debug_port(config=cfg, path=database_path) if debug_port is None else debug_port
|
|
)
|
|
_assert_debug_port_available(port, database_path)
|
|
slug = account_config.make_slug(alias)
|
|
user_data_dir = account_config.ensure_user_data_dir(slug, config=cfg)
|
|
|
|
try:
|
|
return db.add_account(
|
|
account_name,
|
|
alias,
|
|
region_host,
|
|
port,
|
|
password=_optional_text(password),
|
|
note=_optional_text(note),
|
|
slug=slug,
|
|
user_data_dir=user_data_dir,
|
|
path=database_path,
|
|
)
|
|
except db.DbError as exc:
|
|
raise AccountError(str(exc)) from exc
|
|
|
|
|
|
def update_account(
|
|
original_alias,
|
|
account_name,
|
|
alias,
|
|
region_host=None,
|
|
debug_port=None,
|
|
password=None,
|
|
note=None,
|
|
path=None,
|
|
config=None,
|
|
):
|
|
cfg = appconfig.load_config() if config is None else config
|
|
database_path = _db_path(path, cfg)
|
|
existing = get_account(original_alias, path=database_path, config=cfg)
|
|
|
|
account_name = _required_text(account_name, "账号名")
|
|
alias = _required_text(alias, "别名")
|
|
region_host = normalize_region_host(region_host)
|
|
port = normalize_debug_port(debug_port)
|
|
_assert_debug_port_available(port, database_path, ignore_alias=existing.alias)
|
|
|
|
if alias != existing.alias:
|
|
slug = account_config.make_slug(alias)
|
|
user_data_dir = account_config.ensure_user_data_dir(slug, config=cfg)
|
|
else:
|
|
slug = existing.slug
|
|
user_data_dir = os.path.abspath(existing.user_data_dir)
|
|
os.makedirs(user_data_dir, exist_ok=True)
|
|
|
|
try:
|
|
db.update_account(
|
|
original_alias,
|
|
path=database_path,
|
|
account_name=account_name,
|
|
alias=alias,
|
|
region_host=region_host,
|
|
slug=slug,
|
|
user_data_dir=user_data_dir,
|
|
debug_port=port,
|
|
password=_optional_text(password),
|
|
note=_optional_text(note),
|
|
)
|
|
return get_account(alias, path=database_path, config=cfg)
|
|
except db.DbError as exc:
|
|
raise AccountError(str(exc)) from exc
|
|
|
|
|
|
def delete_account(alias, path=None, config=None) -> None:
|
|
database_path = _db_path(path, config)
|
|
db.init_db(database_path)
|
|
db.delete_account(alias, path=database_path)
|
|
|
|
|
|
def resolve_account(account_or_alias, path=None, config=None):
|
|
if isinstance(account_or_alias, str):
|
|
return get_account(account_or_alias, path=path, config=config)
|
|
return account_or_alias
|
|
|
|
|
|
def _debug_host(account) -> str:
|
|
return f"{chrome.CDP_HOST}:{normalize_debug_port(_account_value(account, 'debug_port'))}"
|
|
|
|
|
|
def _seller_portal_url(account) -> str:
|
|
return f"https://{normalize_region_host(_account_value(account, 'region_host'))}/portal/"
|
|
|
|
|
|
def _page_tabs(host):
|
|
return [
|
|
tab
|
|
for tab in cdp.http_get("/json", host=host)
|
|
if tab.get("type") == "page"
|
|
]
|
|
|
|
|
|
def _find_login_tab_in_tabs(account, tabs):
|
|
region_host = normalize_region_host(_account_value(account, 'region_host')).lower()
|
|
seller_tabs = []
|
|
login_tabs = []
|
|
for tab in tabs:
|
|
url = str(tab.get("url") or "")
|
|
lower_url = url.lower()
|
|
if region_host in lower_url:
|
|
seller_tabs.append(tab)
|
|
elif "accounts.shopee." in lower_url and "/seller/login" in lower_url:
|
|
login_tabs.append(tab)
|
|
return (seller_tabs or login_tabs or [None])[0]
|
|
|
|
|
|
def _find_login_tab(account, host):
|
|
return _find_login_tab_in_tabs(account, _page_tabs(host))
|
|
|
|
|
|
def _wait_for_initial_login_tab(account, host, timeout=INITIAL_LOGIN_TAB_WAIT_SECONDS):
|
|
timeout = max(0.0, float(timeout))
|
|
deadline = time.monotonic() + timeout
|
|
last_tabs = []
|
|
probe_error = None
|
|
first_probe = True
|
|
while first_probe or time.monotonic() < deadline:
|
|
first_probe = False
|
|
try:
|
|
last_tabs = _page_tabs(host)
|
|
probe_error = None
|
|
except Exception as exc:
|
|
probe_error = exc.__class__.__name__
|
|
tab = _find_login_tab_in_tabs(account, last_tabs)
|
|
if tab:
|
|
return tab, last_tabs, probe_error
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
break
|
|
time.sleep(min(INITIAL_LOGIN_TAB_POLL_SECONDS, remaining))
|
|
return None, last_tabs, probe_error
|
|
|
|
|
|
def _safe_startup_page(tabs):
|
|
if len(tabs) != 1:
|
|
return None
|
|
tab = tabs[0]
|
|
url = str(tab.get("url") or "").strip().lower()
|
|
if not url or any(url.startswith(prefix) for prefix in SAFE_STARTUP_PAGE_PREFIXES):
|
|
return tab
|
|
return None
|
|
|
|
|
|
def _navigate_page_target(tab, url):
|
|
websocket_url = tab.get("webSocketDebuggerUrl")
|
|
if not websocket_url:
|
|
raise AccountError("Chrome 初始页面缺少 CDP 地址，无法复用")
|
|
client = cdp.CDP(websocket_url)
|
|
try:
|
|
result = client.send("Page.navigate", {"url": url})
|
|
finally:
|
|
client.close()
|
|
if result.get("errorText"):
|
|
raise AccountError("Chrome 初始页面导航到蝦皮卖家中心失败")
|
|
|
|
|
|
def _activate_login_tab_result(
|
|
tab,
|
|
host,
|
|
portal_url,
|
|
*,
|
|
created_tab=False,
|
|
startup_target_reused=False,
|
|
startup_page_navigated=False,
|
|
page_target_count=0,
|
|
fallback_reason=None,
|
|
target_probe_error=None,
|
|
):
|
|
target_id = tab.get("id")
|
|
if target_id:
|
|
cdp.activate_tab(target_id, host=host)
|
|
return {
|
|
"target_id": target_id,
|
|
"url": tab.get("url") or portal_url,
|
|
"created_tab": bool(created_tab),
|
|
"startup_target_reused": bool(startup_target_reused),
|
|
"startup_page_navigated": bool(startup_page_navigated),
|
|
"page_target_count": int(page_target_count),
|
|
"fallback_reason": fallback_reason,
|
|
"target_probe_error": target_probe_error,
|
|
}
|
|
|
|
|
|
def _open_or_activate_login_tab(
|
|
account,
|
|
*,
|
|
newly_launched=False,
|
|
startup_wait_timeout=INITIAL_LOGIN_TAB_WAIT_SECONDS,
|
|
):
|
|
host = _debug_host(account)
|
|
portal_url = _seller_portal_url(account)
|
|
if newly_launched:
|
|
tab, tabs, probe_error = _wait_for_initial_login_tab(
|
|
account,
|
|
host,
|
|
timeout=startup_wait_timeout,
|
|
)
|
|
else:
|
|
tabs = _page_tabs(host)
|
|
tab = _find_login_tab_in_tabs(account, tabs)
|
|
probe_error = None
|
|
if tab:
|
|
return _activate_login_tab_result(
|
|
tab,
|
|
host,
|
|
portal_url,
|
|
startup_target_reused=newly_launched,
|
|
page_target_count=len(tabs),
|
|
target_probe_error=probe_error,
|
|
)
|
|
|
|
fallback_reason = "existing_chrome_no_login_tab"
|
|
if newly_launched:
|
|
startup_tab = _safe_startup_page(tabs)
|
|
if startup_tab:
|
|
try:
|
|
_navigate_page_target(startup_tab, portal_url)
|
|
except Exception as exc:
|
|
fallback_reason = "startup_navigation_failed"
|
|
probe_error = probe_error or exc.__class__.__name__
|
|
else:
|
|
navigated_tab = dict(startup_tab)
|
|
navigated_tab["url"] = portal_url
|
|
return _activate_login_tab_result(
|
|
navigated_tab,
|
|
host,
|
|
portal_url,
|
|
startup_target_reused=True,
|
|
startup_page_navigated=True,
|
|
page_target_count=len(tabs),
|
|
target_probe_error=probe_error,
|
|
)
|
|
elif tabs:
|
|
fallback_reason = "startup_page_not_safe"
|
|
else:
|
|
fallback_reason = "startup_page_missing"
|
|
|
|
tab = cdp.create_tab_info(portal_url, host=host)
|
|
return _activate_login_tab_result(
|
|
tab,
|
|
host,
|
|
portal_url,
|
|
created_tab=True,
|
|
page_target_count=len(tabs) + 1,
|
|
fallback_reason=fallback_reason,
|
|
target_probe_error=probe_error,
|
|
)
|
|
|
|
|
|
def launch_for_login(account_or_alias, path=None, config=None):
|
|
account = resolve_account(account_or_alias, path=path, config=config)
|
|
port = normalize_debug_port(_account_value(account, "debug_port"))
|
|
alias = _account_value(account, "alias")
|
|
portal_url = _seller_portal_url(account)
|
|
|
|
if chrome.is_running(port):
|
|
tab = _open_or_activate_login_tab(account, newly_launched=False)
|
|
return {
|
|
"ok": True,
|
|
"action": "reused",
|
|
"reused": True,
|
|
"launched": False,
|
|
"alias": alias,
|
|
"debug_port": port,
|
|
"pid": None,
|
|
"initial_url": portal_url,
|
|
**tab,
|
|
}
|
|
|
|
process = chrome.launch_chrome(account, config=config, initial_url=portal_url)
|
|
timeout = appconfig.cdp_ready_timeout(config)
|
|
if not chrome.wait_debug_ready(port, timeout=timeout):
|
|
raise AccountError(f"Chrome 已启动，但 CDP 端口 {port} 未在 {timeout} 秒内就绪")
|
|
tab = _open_or_activate_login_tab(account, newly_launched=True)
|
|
return {
|
|
"ok": True,
|
|
"action": "launched",
|
|
"reused": False,
|
|
"launched": True,
|
|
"alias": alias,
|
|
"debug_port": port,
|
|
"pid": getattr(process, "pid", None),
|
|
"initial_url": portal_url,
|
|
**tab,
|
|
}
|
|
|
|
|
|
def create_shortcut(account_or_alias, shortcut_path=None, desktop_dir=None, path=None, config=None) -> str:
|
|
account = resolve_account(account_or_alias, path=path, config=config)
|
|
return chrome.create_shortcut(
|
|
account,
|
|
shortcut_path=shortcut_path,
|
|
desktop_dir=desktop_dir,
|
|
config=config,
|
|
)
|
|
|
|
|
|
def detect_login(account_or_alias, timeout=8, path=None, config=None) -> dict:
|
|
database_path = _db_path(path, config)
|
|
account = resolve_account(account_or_alias, path=database_path, config=config)
|
|
status = editor.login_status(account, timeout=timeout)
|
|
if status.get("logged_in"):
|
|
db.update_account(account.alias, path=database_path, last_login_at=_now())
|
|
return status
|
|
|
|
|
|
def login_status_text(status) -> str:
|
|
if not status:
|
|
return "未知"
|
|
if status.get("logged_in"):
|
|
return "已登录"
|
|
reason = status.get("reason")
|
|
if reason == "LOGIN_PAGE":
|
|
return "未登录"
|
|
if reason == "NO_SESSION_COOKIE":
|
|
return "未登录"
|
|
if reason == "LOGIN_CHECK_TARGET_UNAVAILABLE" or str(reason).startswith(
|
|
"LOGIN_CHECK_FAILED"
|
|
):
|
|
return "登录状态暂不可确认"
|
|
if reason:
|
|
return f"未登录({reason})"
|
|
return "未登录"
|