"""Account management services for the PySide6 accounts tab.""" from __future__ import annotations import os from datetime import datetime from typing import Optional from urllib.parse import urlparse from . import appconfig, chrome, db, editor from . import config as account_config DEFAULT_REGION_HOST = editor.DEFAULT_REGION_HOST 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 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 launch_for_login(account_or_alias, path=None, config=None): account = resolve_account(account_or_alias, path=path, config=config) return chrome.launch_chrome(account, config=config) 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: return f"未登录({reason})" return "未登录"