sub2api_client.py

"""Push local Grok accounts into a sub2api (Wei-Shaw/sub2api) instance.

sub2api admin flow used here:

1. ``POST /api/v1/auth/login`` → JWT
2. ``GET  /api/v1/admin/groups`` → pick group_id (or create)
3. Prefer direct OAuth create with local access/refresh tokens:
   ``POST /api/v1/admin/accounts``  platform=grok type=oauth
4. Fallback when only SSO cookies are available:
   ``POST /api/v1/admin/grok/sso-to-oauth`` then create account

Config is stored under settings key ``sub2api_config``.
"""

from __future__ import annotations

import json
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Any

import accounts

_DEFAULT_TIMEOUT = 45.0
_USER_AGENT = "grokcli-2api-sub2api-push/1.0"


# ---------------------------------------------------------------------------
# Settings helpers
# ---------------------------------------------------------------------------

def _default_config() -> dict[str, Any]:
    return {
        "enabled": False,
        "base_url": "",
        "email": "",
        "password": "",
        "group_id": None,
        "group_name": "",
        "auto_create_group": True,
        # After protocol registration succeeds and the account is imported
        # locally, automatically push it into sub2api (requires enabled + URL).
        "auto_push_on_register": False,
        # How many accounts to push in parallel (local → sub2api)
        "concurrency": 4,
        # Per-account capacity written into sub2api account.concurrency
        "account_concurrency": 3,
        "account_priority": 50,
        "account_rate_multiplier": 1.0,
        "notes_prefix": "grokcli-2api",
        # Cached JWT (optional; refreshed on 401)
        "token": "",
        "token_expires_at": 0,
    }


def _normalize_config(raw: Any, *, include_secrets: bool = True) -> dict[str, Any]:
    base = _default_config()
    if not isinstance(raw, dict):
        return base
    out = dict(base)
    out["enabled"] = bool(raw.get("enabled", False))
    out["base_url"] = str(raw.get("base_url") or raw.get("url") or "").strip().rstrip("/")
    out["email"] = str(raw.get("email") or raw.get("username") or "").strip()
    pw = raw.get("password")
    if include_secrets:
        out["password"] = "" if pw is None else str(pw)
    else:
        out["password"] = ""
        out["has_password"] = bool(str(pw or "").strip())
    gid = raw.get("group_id")
    if gid in (None, "", 0, "0"):
        out["group_id"] = None
    else:
        try:
            out["group_id"] = int(gid)
        except (TypeError, ValueError):
            out["group_id"] = None
    out["group_name"] = str(raw.get("group_name") or "").strip()
    out["auto_create_group"] = bool(raw.get("auto_create_group", True))
    # Accept several aliases from UI / older drafts
    auto_push = raw.get("auto_push_on_register")
    if auto_push is None:
        auto_push = raw.get("auto_import_on_register")
    if auto_push is None:
        auto_push = raw.get("auto_push_registration")
    out["auto_push_on_register"] = bool(auto_push)
    try:
        conc = int(raw.get("concurrency") or 4)
    except (TypeError, ValueError):
        conc = 4
    out["concurrency"] = max(1, min(16, conc))
    # Account capacity on sub2api side (also accept legacy aliases)
    try:
        acc_conc = int(
            raw.get("account_concurrency")
            if raw.get("account_concurrency") is not None
            else raw.get("account_capacity")
            if raw.get("account_capacity") is not None
            else 3
        )
    except (TypeError, ValueError):
        acc_conc = 3
    out["account_concurrency"] = max(1, min(100, acc_conc))
    try:
        prio = int(
            raw.get("account_priority")
            if raw.get("account_priority") is not None
            else 50
        )
    except (TypeError, ValueError):
        prio = 50
    out["account_priority"] = max(0, min(100, prio))
    try:
        rate = float(
            raw.get("account_rate_multiplier")
            if raw.get("account_rate_multiplier") is not None
            else raw.get("rate_multiplier")
            if raw.get("rate_multiplier") is not None
            else 1.0
        )
    except (TypeError, ValueError):
        rate = 1.0
    out["account_rate_multiplier"] = max(0.1, min(10.0, rate))
    out["notes_prefix"] = str(raw.get("notes_prefix") or "grokcli-2api").strip() or "grokcli-2api"
    if include_secrets:
        out["token"] = str(raw.get("token") or "").strip()
        try:
            out["token_expires_at"] = float(raw.get("token_expires_at") or 0)
        except (TypeError, ValueError):
            out["token_expires_at"] = 0.0
    else:
        out["has_token"] = bool(str(raw.get("token") or "").strip())
        out["token_expires_at"] = float(raw.get("token_expires_at") or 0) or 0
    return out


def get_sub2api_config(*, include_secrets: bool = True) -> dict[str, Any]:
    try:
        from settings_store import _get_setting_value  # type: ignore

        raw = _get_setting_value("sub2api_config", None)
    except Exception:
        raw = None
    return _normalize_config(raw, include_secrets=include_secrets)


def set_sub2api_config(patch: dict[str, Any] | None, *, replace: bool = False) -> dict[str, Any]:
    """Merge or replace sub2api_config. Empty password keeps previous."""
    if patch is None:
        patch = {}
    if not isinstance(patch, dict):
        raise ValueError("sub2api_config must be an object")
    current = get_sub2api_config(include_secrets=True)
    if replace:
        merged = _normalize_config(patch, include_secrets=True)
        # Preserve password/token when UI sends blank password intentionally keep
        if not str(merged.get("password") or "").strip() and current.get("password"):
            merged["password"] = current["password"]
        if not str(merged.get("token") or "").strip() and current.get("token"):
            merged["token"] = current.get("token") or ""
            merged["token_expires_at"] = current.get("token_expires_at") or 0
    else:
        merged = dict(current)
        for k, v in patch.items():
            if k in ("password", "token") and (v is None or str(v).strip() == ""):
                continue  # keep existing secret
            merged[k] = v
        merged = _normalize_config(merged, include_secrets=True)
    try:
        from settings_store import _set_setting_value  # type: ignore

        _set_setting_value("sub2api_config", merged)
    except Exception as e:  # noqa: BLE001
        raise RuntimeError(f"failed to persist sub2api_config: {e}") from e
    return get_sub2api_config(include_secrets=True)


def public_sub2api_config() -> dict[str, Any]:
    """Admin UI payload — secrets redacted."""
    cfg = get_sub2api_config(include_secrets=True)
    return _normalize_config(cfg, include_secrets=False)


# ---------------------------------------------------------------------------
# HTTP
# ---------------------------------------------------------------------------

def _urljoin(base: str, path: str) -> str:
    base = (base or "").rstrip("/")
    if not path.startswith("/"):
        path = "/" + path
    return base + path


def _http_json(
    method: str,
    url: str,
    *,
    headers: dict[str, str] | None = None,
    body: Any = None,
    timeout: float = _DEFAULT_TIMEOUT,
) -> tuple[int, Any, str]:
    data = None
    hdrs = {
        "Accept": "application/json",
        "User-Agent": _USER_AGENT,
    }
    if headers:
        hdrs.update(headers)
    if body is not None:
        data = json.dumps(body, ensure_ascii=False).encode("utf-8")
        hdrs["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=hdrs, method=method.upper())
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read().decode("utf-8", errors="replace")
            status = int(getattr(resp, "status", 200) or 200)
            try:
                parsed = json.loads(raw) if raw.strip() else None
            except json.JSONDecodeError:
                parsed = raw
            return status, parsed, raw
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", errors="replace") if e.fp else str(e)
        try:
            parsed = json.loads(raw) if raw.strip() else None
        except json.JSONDecodeError:
            parsed = raw
        return int(e.code), parsed, raw
    except Exception as e:  # noqa: BLE001
        return 0, None, str(e)


def _api_error_message(status: int, parsed: Any, raw: str) -> str:
    if isinstance(parsed, dict):
        # sub2api style: {code:N, message:"...", data:...} (HTTP may still be 200)
        code = parsed.get("code")
        if code not in (None, 0, "0", 200, "200"):
            msg = parsed.get("message") or parsed.get("msg") or parsed.get("error")
            if isinstance(msg, dict):
                msg = msg.get("message") or msg.get("msg") or str(msg)
            if msg:
                return f"HTTP {status}: {msg}"
            return f"HTTP {status}: code={code}"
        for k in ("message", "error", "detail", "msg"):
            v = parsed.get(k)
            if v and not isinstance(v, (dict, list)):
                return f"HTTP {status}: {v}"
        err = parsed.get("error")
        if isinstance(err, dict) and err.get("message"):
            return f"HTTP {status}: {err.get('message')}"
    text = (raw or "").strip()
    if text:
        return f"HTTP {status}: {text[:300]}"
    return f"HTTP {status}"


def _unwrap_data(parsed: Any) -> Any:
    """Unwrap sub2api `{code, data}` / `{data: ...}` envelopes."""
    if not isinstance(parsed, dict):
        return parsed
    code = parsed.get("code")
    if code not in (None, 0, "0", 200, "200"):
        return parsed
    if "data" in parsed:
        return parsed.get("data")
    return parsed


# ---------------------------------------------------------------------------
# Auth / groups
# ---------------------------------------------------------------------------

def login(cfg: dict[str, Any] | None = None, *, force: bool = False) -> dict[str, Any]:
    """Login to sub2api admin; cache JWT in settings when successful."""
    cfg = cfg or get_sub2api_config(include_secrets=True)
    base = cfg.get("base_url") or ""
    if not base:
        raise ValueError("sub2api base_url is required")
    email = cfg.get("email") or ""
    password = cfg.get("password") or ""
    if not email or not password:
        raise ValueError("sub2api email/password is required")

    token = str(cfg.get("token") or "").strip()
    exp = float(cfg.get("token_expires_at") or 0)
    if not force and token and exp > time.time() + 60:
        return {"ok": True, "token": token, "cached": True, "expires_at": exp}

    status, parsed, raw = _http_json(
        "POST",
        _urljoin(base, "/api/v1/auth/login"),
        body={"email": email, "password": password},
        timeout=30,
    )
    if status < 200 or status >= 300:
        raise RuntimeError(_api_error_message(status, parsed, raw) or "login failed")
    if isinstance(parsed, dict):
        code = parsed.get("code")
        if code not in (None, 0, "0", 200, "200"):
            raise RuntimeError(_api_error_message(status, parsed, raw) or "login failed")
    # Response shapes:
    #   {access_token, expires_in}
    #   {code:0, data:{access_token, expires_in, ...}}  (sub2api)
    #   {data:{token:...}}
    data = _unwrap_data(parsed)
    if not isinstance(data, dict):
        data = parsed if isinstance(parsed, dict) else {}
    new_token = (
        data.get("access_token")
        or data.get("token")
        or (parsed.get("access_token") if isinstance(parsed, dict) else None)
        or (parsed.get("token") if isinstance(parsed, dict) else None)
        or ""
    )
    new_token = str(new_token).strip()
    if not new_token:
        raise RuntimeError(f"login response missing token: {raw[:200]}")
    # Default 12h cache if server omits exp
    expires_in = data.get("expires_in") or parsed.get("expires_in") or 12 * 3600
    try:
        expires_in = float(expires_in)
    except (TypeError, ValueError):
        expires_in = 12 * 3600
    expires_at = time.time() + max(300.0, expires_in)
    try:
        set_sub2api_config({"token": new_token, "token_expires_at": expires_at})
    except Exception:
        pass
    return {"ok": True, "token": new_token, "cached": False, "expires_at": expires_at}


def _auth_headers(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def _request_authed(
    method: str,
    path: str,
    *,
    cfg: dict[str, Any] | None = None,
    body: Any = None,
    timeout: float = _DEFAULT_TIMEOUT,
    retry_login: bool = True,
) -> tuple[int, Any, str]:
    cfg = cfg or get_sub2api_config(include_secrets=True)
    base = cfg.get("base_url") or ""
    if not base:
        raise ValueError("sub2api base_url is required")
    try:
        auth = login(cfg, force=False)
    except Exception:
        if not retry_login:
            raise
        auth = login(cfg, force=True)
    token = auth["token"]
    status, parsed, raw = _http_json(
        method,
        _urljoin(base, path),
        headers=_auth_headers(token),
        body=body,
        timeout=timeout,
    )
    if status in (401, 403) and retry_login:
        auth = login(cfg, force=True)
        status, parsed, raw = _http_json(
            method,
            _urljoin(base, path),
            headers=_auth_headers(auth["token"]),
            body=body,
            timeout=timeout,
        )
    return status, parsed, raw


def list_groups(cfg: dict[str, Any] | None = None) -> list[dict[str, Any]]:
    status, parsed, raw = _request_authed("GET", "/api/v1/admin/groups", cfg=cfg)
    if status < 200 or status >= 300:
        raise RuntimeError(_api_error_message(status, parsed, raw))
    if isinstance(parsed, dict):
        code = parsed.get("code")
        if code not in (None, 0, "0", 200, "200"):
            raise RuntimeError(_api_error_message(status, parsed, raw))
    data = _unwrap_data(parsed)
    items: Any = data
    if isinstance(data, dict):
        items = (
            data.get("items")
            or data.get("groups")
            or data.get("list")
            or data.get("data")
            or []
        )
        if isinstance(items, dict):
            items = items.get("items") or items.get("list") or []
    elif isinstance(parsed, dict) and not isinstance(data, list):
        # fallback legacy
        items = (
            parsed.get("items")
            or parsed.get("groups")
            or parsed.get("list")
            or []
        )
    if not isinstance(items, list):
        return []
    out: list[dict[str, Any]] = []
    for g in items:
        if not isinstance(g, dict):
            continue
        out.append(
            {
                "id": g.get("id"),
                "name": g.get("name") or g.get("title") or "",
                "platform": g.get("platform") or g.get("platform_id") or "",
                "description": g.get("description") or "",
                "status": g.get("status"),
                "account_count": g.get("account_count") or g.get("accounts_count"),
            }
        )
    return out


def create_group(
    name: str,
    *,
    platform: str = "grok",
    description: str = "",
    cfg: dict[str, Any] | None = None,
) -> dict[str, Any]:
    name = str(name or "").strip()
    if not name:
        raise ValueError("group name is required")
    body = {
        "name": name,
        "platform": platform or "grok",
        "description": description or "created by grokcli-2api",
        "rate_multiplier": 1.0,
        "is_exclusive": False,
    }
    status, parsed, raw = _request_authed(
        "POST", "/api/v1/admin/groups", cfg=cfg, body=body
    )
    if status < 200 or status >= 300:
        raise RuntimeError(_api_error_message(status, parsed, raw))
    if isinstance(parsed, dict):
        code = parsed.get("code")
        if code not in (None, 0, "0", 200, "200"):
            raise RuntimeError(_api_error_message(status, parsed, raw))
    data = _unwrap_data(parsed)
    if not isinstance(data, dict):
        data = parsed if isinstance(parsed, dict) else {"raw": parsed}
    return data


def resolve_group_id(cfg: dict[str, Any] | None = None) -> int:
    """Return configured group_id, matching by name, or auto-create."""
    cfg = cfg or get_sub2api_config(include_secrets=True)
    if cfg.get("group_id"):
        return int(cfg["group_id"])
    name = str(cfg.get("group_name") or "").strip() or "grokcli-2api"
    groups = list_groups(cfg)
    for g in groups:
        if str(g.get("name") or "").strip() == name:
            gid = int(g["id"])
            try:
                set_sub2api_config({"group_id": gid, "group_name": name})
            except Exception:
                pass
            return gid
        # also match platform-filtered same name ignore case
        if str(g.get("name") or "").strip().lower() == name.lower():
            gid = int(g["id"])
            try:
                set_sub2api_config({"group_id": gid, "group_name": g.get("name") or name})
            except Exception:
                pass
            return gid
    if not cfg.get("auto_create_group", True):
        raise RuntimeError(f"group not found: {name}")
    created = create_group(name, platform="grok", cfg=cfg)
    gid = created.get("id")
    if gid is None:
        # re-list
        for g in list_groups(cfg):
            if str(g.get("name") or "").strip() == name:
                gid = g.get("id")
                break
    if gid is None:
        raise RuntimeError(f"failed to create group {name}: {created}")
    gid_i = int(gid)
    try:
        set_sub2api_config({"group_id": gid_i, "group_name": name})
    except Exception:
        pass
    return gid_i


# ---------------------------------------------------------------------------
# Account push
# ---------------------------------------------------------------------------

def _local_account_entry(account_id: str) -> tuple[str, dict[str, Any]] | None:
    data = accounts.read_auth_map()
    if not data:
        return None
    aid = str(account_id or "").strip()
    if aid in data and isinstance(data[aid], dict):
        return aid, data[aid]
    # fuzzy match by email / suffix
    for k, v in data.items():
        if not isinstance(v, dict):
            continue
        if str(v.get("email") or "").strip() == aid:
            return k, v
        if k.endswith(aid) or aid.endswith(k):
            return k, v
    return None


def _entry_tokens(entry: dict[str, Any]) -> tuple[str, str]:
    access = (
        entry.get("key")
        or entry.get("access_token")
        or entry.get("token")
        or ""
    )
    refresh = entry.get("refresh_token") or ""
    return str(access).strip(), str(refresh).strip()


def _entry_sso_candidates(entry: dict[str, Any], account_id: str) -> list[str]:
    """Best-effort SSO cookie extraction (often empty after token import)."""
    out: list[str] = []
    for k in ("sso", "sso_cookie", "sso_token", "session_cookie"):
        v = entry.get(k)
        if isinstance(v, str) and v.strip():
            out.append(v.strip())
    sc = entry.get("session_cookies")
    if isinstance(sc, dict):
        for k in ("sso", "SSO", "session", "token"):
            v = sc.get(k)
            if isinstance(v, str) and v.strip():
                out.append(v.strip())
    elif isinstance(sc, list):
        for item in sc:
            if isinstance(item, str) and item.strip():
                out.append(item.strip())
            elif isinstance(item, dict):
                for k in ("sso", "value", "token"):
                    v = item.get(k)
                    if isinstance(v, str) and v.strip():
                        out.append(v.strip())
    # Dedup preserve order
    seen: set[str] = set()
    uniq: list[str] = []
    for x in out:
        if x not in seen:
            seen.add(x)
            uniq.append(x)
    return uniq


def _expires_at_iso(entry: dict[str, Any], access_token: str) -> str | None:
    exp = entry.get("expires_at")
    if isinstance(exp, (int, float)) and exp > 0:
        try:
            return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(float(exp)))
        except Exception:
            pass
    if isinstance(exp, str) and exp.strip():
        return exp.strip()
    # JWT payload exp
    try:
        import base64

        parts = access_token.split(".")
        if len(parts) >= 2:
            pad = "=" * (-len(parts[1]) % 4)
            payload = json.loads(base64.urlsafe_b64decode(parts[1] + pad))
            if payload.get("exp"):
                return time.strftime(
                    "%Y-%m-%dT%H:%M:%SZ", time.gmtime(float(payload["exp"]))
                )
    except Exception:
        pass
    return None


def sso_to_oauth(
    sso_tokens: list[str],
    *,
    cfg: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    tokens = [str(x).strip() for x in sso_tokens if str(x).strip()]
    if not tokens:
        return []
    status, parsed, raw = _request_authed(
        "POST",
        "/api/v1/admin/grok/sso-to-oauth",
        cfg=cfg,
        body={"sso_tokens": tokens, "proxy_id": None},
        timeout=120,
    )
    if status < 200 or status >= 300:
        raise RuntimeError(_api_error_message(status, parsed, raw))
    if isinstance(parsed, dict):
        code = parsed.get("code")
        if code not in (None, 0, "0", 200, "200"):
            raise RuntimeError(_api_error_message(status, parsed, raw))
    data = _unwrap_data(parsed)
    if isinstance(data, dict) and isinstance(data.get("results"), list):
        return data["results"]
    if isinstance(data, list):
        return data
    if isinstance(parsed, dict) and isinstance(parsed.get("results"), list):
        return parsed["results"]
    return []


def create_grok_oauth_account(
    *,
    name: str,
    group_id: int,
    access_token: str,
    refresh_token: str = "",
    email: str = "",
    expires_at: str | None = None,
    notes: str = "",
    cfg: dict[str, Any] | None = None,
) -> dict[str, Any]:
    credentials: dict[str, Any] = {
        "access_token": access_token,
        "email": email or "",
    }
    if refresh_token:
        credentials["refresh_token"] = refresh_token
    if expires_at:
        credentials["expires_at"] = expires_at
    live_cfg = cfg or get_sub2api_config(include_secrets=True)
    try:
        acc_conc = int(live_cfg.get("account_concurrency") or 3)
    except (TypeError, ValueError):
        acc_conc = 3
    acc_conc = max(1, min(100, acc_conc))
    try:
        acc_prio = int(live_cfg.get("account_priority") if live_cfg.get("account_priority") is not None else 50)
    except (TypeError, ValueError):
        acc_prio = 50
    acc_prio = max(0, min(100, acc_prio))
    try:
        acc_rate = float(live_cfg.get("account_rate_multiplier") or 1.0)
    except (TypeError, ValueError):
        acc_rate = 1.0
    acc_rate = max(0.1, min(10.0, acc_rate))
    body: dict[str, Any] = {
        "name": name[:200] if name else (email or "grok-account")[:200],
        "platform": "grok",
        "type": "oauth",
        "credentials": credentials,
        "extra": {},
        "proxy_id": None,
        "group_ids": [int(group_id)],
        "concurrency": acc_conc,
        "priority": acc_prio,
        "rate_multiplier": acc_rate,
        "notes": notes or "",
    }
    status, parsed, raw = _request_authed(
        "POST",
        "/api/v1/admin/accounts",
        cfg=cfg,
        body=body,
        timeout=60,
    )
    if status < 200 or status >= 300:
        raise RuntimeError(_api_error_message(status, parsed, raw))
    if isinstance(parsed, dict):
        code = parsed.get("code")
        if code not in (None, 0, "0", 200, "200"):
            raise RuntimeError(_api_error_message(status, parsed, raw))
    data = _unwrap_data(parsed)
    return data if isinstance(data, dict) else {"raw": parsed}


def push_account(
    account_id: str,
    *,
    group_id: int | None = None,
    cfg: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Push one local account to sub2api. Returns result dict."""
    cfg = cfg or get_sub2api_config(include_secrets=True)
    matched = _local_account_entry(account_id)
    if not matched:
        return {"ok": False, "account_id": account_id, "error": "account not found"}
    aid, entry = matched
    email = str(entry.get("email") or "").strip()
    access, refresh = _entry_tokens(entry)
    notes_prefix = str(cfg.get("notes_prefix") or "grokcli-2api")
    notes = f"{notes_prefix}:{aid}"
    name = email or aid
    gid = int(group_id or resolve_group_id(cfg))

    # Path A: direct OAuth with local tokens
    if access:
        try:
            created = create_grok_oauth_account(
                name=name,
                group_id=gid,
                access_token=access,
                refresh_token=refresh,
                email=email,
                expires_at=_expires_at_iso(entry, access),
                notes=notes,
                cfg=cfg,
            )
            return {
                "ok": True,
                "account_id": aid,
                "email": email,
                "method": "oauth_token",
                "group_id": gid,
                "remote": {
                    "id": created.get("id"),
                    "name": created.get("name"),
                },
            }
        except Exception as e:  # noqa: BLE001
            token_err = str(e)
            # fall through to SSO if available
            sso_list = _entry_sso_candidates(entry, aid)
            if not sso_list:
                return {
                    "ok": False,
                    "account_id": aid,
                    "email": email,
                    "error": token_err,
                    "method": "oauth_token",
                }
    else:
        token_err = "missing access_token"
        sso_list = _entry_sso_candidates(entry, aid)
        if not sso_list:
            return {
                "ok": False,
                "account_id": aid,
                "email": email,
                "error": token_err,
                "method": "none",
            }

    # Path B: SSO → OAuth then create
    try:
        results = sso_to_oauth(sso_list, cfg=cfg)
    except Exception as e:  # noqa: BLE001
        return {
            "ok": False,
            "account_id": aid,
            "email": email,
            "error": f"sso-to-oauth failed: {e}",
            "method": "sso",
        }
    # Pick first success
    cred = None
    for r in results:
        if not isinstance(r, dict):
            continue
        if r.get("success") is False:
            continue
        # nested credentials or flat
        c = r.get("credentials") if isinstance(r.get("credentials"), dict) else r
        at = c.get("access_token") or c.get("AccessToken") or ""
        if at:
            cred = c
            if not email:
                email = str(c.get("email") or r.get("email") or "").strip()
            break
    if not cred:
        errs = []
        for r in results:
            if isinstance(r, dict) and (r.get("error") or r.get("message")):
                errs.append(str(r.get("error") or r.get("message")))
        return {
            "ok": False,
            "account_id": aid,
            "email": email,
            "error": "sso-to-oauth produced no credentials"
            + (f": {'; '.join(errs)}" if errs else ""),
            "method": "sso",
        }
    access2 = str(cred.get("access_token") or "").strip()
    refresh2 = str(cred.get("refresh_token") or "").strip()
    exp2 = cred.get("expires_at") or cred.get("expires_in")
    exp_iso = None
    if isinstance(exp2, str):
        exp_iso = exp2
    elif isinstance(exp2, (int, float)) and exp2 > 1_000_000_000:
        exp_iso = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(float(exp2)))
    try:
        created = create_grok_oauth_account(
            name=email or name,
            group_id=gid,
            access_token=access2,
            refresh_token=refresh2,
            email=email,
            expires_at=exp_iso,
            notes=notes,
            cfg=cfg,
        )
        return {
            "ok": True,
            "account_id": aid,
            "email": email,
            "method": "sso",
            "group_id": gid,
            "remote": {
                "id": created.get("id"),
                "name": created.get("name"),
            },
        }
    except Exception as e:  # noqa: BLE001
        return {
            "ok": False,
            "account_id": aid,
            "email": email,
            "error": str(e),
            "method": "sso",
        }


def maybe_auto_push_registered_accounts(
    account_ids: list[str] | None,
    *,
    cfg: dict[str, Any] | None = None,
    source: str = "register-email",
) -> dict[str, Any]:
    """Push freshly registered local accounts into sub2api when configured.

    Safe no-op when:
    - ``auto_push_on_register`` is off
    - sub2api is not enabled / missing URL or credentials
    - no account ids were provided

    Never raises — registration success must not fail because of sub2api push.
    """
    ids = [str(x).strip() for x in (account_ids or []) if str(x).strip()]
    if not ids:
        return {"ok": True, "skipped": True, "reason": "no_accounts", "results": []}
    try:
        live = cfg or get_sub2api_config(include_secrets=True)
    except Exception as e:  # noqa: BLE001
        return {
            "ok": False,
            "skipped": True,
            "reason": f"config_error: {e}",
            "results": [],
        }
    if not bool(live.get("auto_push_on_register")):
        return {
            "ok": True,
            "skipped": True,
            "reason": "auto_push_on_register_disabled",
            "results": [],
        }
    if not bool(live.get("enabled")):
        return {
            "ok": False,
            "skipped": True,
            "reason": "sub2api_disabled",
            "results": [],
        }
    if not str(live.get("base_url") or "").strip():
        return {
            "ok": False,
            "skipped": True,
            "reason": "missing_base_url",
            "results": [],
        }
    if not str(live.get("email") or "").strip() or not (
        str(live.get("password") or "").strip() or str(live.get("token") or "").strip()
    ):
        return {
            "ok": False,
            "skipped": True,
            "reason": "missing_credentials",
            "results": [],
        }

    results: list[dict[str, Any]] = []
    ok_n = 0
    fail_n = 0
    for aid in ids:
        try:
            r = push_account(aid, cfg=live)
        except Exception as e:  # noqa: BLE001
            r = {"ok": False, "account_id": aid, "error": str(e), "method": "auto"}
        if not isinstance(r, dict):
            r = {"ok": False, "account_id": aid, "error": "invalid push result"}
        r.setdefault("source", source)
        results.append(r)
        if r.get("ok"):
            ok_n += 1
        else:
            fail_n += 1
        try:
            # Light pacing so bulk registration does not stampede sub2api.
            time.sleep(0.05)
        except Exception:
            pass
    summary = {
        "ok": fail_n == 0,
        "skipped": False,
        "source": source,
        "total": len(ids),
        "success": ok_n,
        "failed": fail_n,
        "results": results,
    }
    try:
        print(
            f"[sub2api] auto_push_on_register source={source} "
            f"total={len(ids)} ok={ok_n} fail={fail_n}"
        )
    except Exception:
        pass
    return summary


def push_accounts(
    account_ids: list[str] | None = None,
    *,
    group_id: int | None = None,
    cfg: dict[str, Any] | None = None,
    concurrency: int | None = None,
) -> dict[str, Any]:
    """Push selected or all local accounts to sub2api."""
    cfg = cfg or get_sub2api_config(include_secrets=True)
    if not cfg.get("base_url"):
        return {"ok": False, "error": "sub2api base_url not configured"}
    if not cfg.get("email") or not cfg.get("password"):
        return {"ok": False, "error": "sub2api login email/password not configured"}

    # Resolve ids
    data = accounts.read_auth_map() or {}
    if account_ids is None:
        ids = [k for k, v in data.items() if isinstance(v, dict)]
    else:
        ids = [str(x).strip() for x in account_ids if str(x).strip()]
    if not ids:
        return {"ok": True, "total": 0, "success": 0, "failed": 0, "results": []}

    try:
        gid = int(group_id) if group_id else resolve_group_id(cfg)
    except Exception as e:  # noqa: BLE001
        return {"ok": False, "error": f"resolve group failed: {e}"}

    # Ensure login works once
    try:
        login(cfg, force=False)
    except Exception as e:  # noqa: BLE001
        return {"ok": False, "error": f"sub2api login failed: {e}"}

    conc = concurrency if concurrency is not None else int(cfg.get("concurrency") or 4)
    conc = max(1, min(16, int(conc)))

    results: list[dict[str, Any]] = []
    if conc <= 1 or len(ids) == 1:
        for aid in ids:
            results.append(push_account(aid, group_id=gid, cfg=cfg))
    else:
        from concurrent.futures import ThreadPoolExecutor, as_completed

        with ThreadPoolExecutor(max_workers=conc) as pool:
            futs = {
                pool.submit(push_account, aid, group_id=gid, cfg=cfg): aid
                for aid in ids
            }
            for fut in as_completed(futs):
                try:
                    results.append(fut.result())
                except Exception as e:  # noqa: BLE001
                    results.append(
                        {
                            "ok": False,
                            "account_id": futs[fut],
                            "error": str(e),
                        }
                    )

    success = sum(1 for r in results if r.get("ok"))
    failed = len(results) - success
    return {
        "ok": failed == 0,
        "total": len(results),
        "success": success,
        "failed": failed,
        "group_id": gid,
        "results": results,
    }


def test_connection(cfg: dict[str, Any] | None = None) -> dict[str, Any]:
    """Login + list groups smoke test for settings UI."""
    cfg = cfg or get_sub2api_config(include_secrets=True)
    try:
        auth = login(cfg, force=True)
        groups = list_groups(cfg)
        return {
            "ok": True,
            "message": "连接成功",
            "token_cached": bool(auth.get("token")),
            "groups": groups,
            "group_count": len(groups),
        }
    except Exception as e:  # noqa: BLE001
        return {"ok": False, "error": str(e)}