cmshoppe/app/accounts.py

"""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 "未登录"