Boss Cli

name: CI

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]
  workflow_call:

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install system dependencies
        run: sudo apt-get update && sudo apt-get install -y libsecret-1-dev

      - name: Setup uv
        uses: astral-sh/setup-uv@v6

      - name: Install dependencies
        run: uv sync --all-extras

      - name: Lint
        run: uv run ruff check .

      - name: Test
        run: uv run pytest tests/test_cli.py -v

name: Publish to PyPI

on:
  push:
    tags:
      - "v*"
  workflow_dispatch:

jobs:
  verify:
    uses: ./.github/workflows/ci.yml

  publish:
    needs: verify
    runs-on: ubuntu-latest
    environment: pypi
    permissions:
      id-token: write

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Setup uv
        uses: astral-sh/setup-uv@v6

      - name: Build package
        run: uv build

      - name: Publish to PyPI
        uses: pypa/gh-action-pypi-publish@release/v1

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
*.egg-info/
dist/
build/
*.egg

# Virtual environments
.venv/
venv/
env/

# IDE
.idea/
.vscode/
*.swp
*.swo
*~

# OS
.DS_Store
Thumbs.db

# Testing
.coverage
htmlcov/
.pytest_cache/
.mypy_cache/

# Config (DO NOT commit credentials)
credential.json

"""Boss CLI — A terminal client for Boss Zhipin (BOSS直聘)."""

__version__ = "0.3.5"

"""Authentication for Boss Zhipin.

Strategy:
1. Try loading saved credential from ~/.config/boss-cli/credential.json
2. Try extracting cookies from local browsers via browser-cookie3
3. Fallback: QR code login in terminal
"""

from __future__ import annotations

import glob
import hashlib
import json
import logging
import os
import platform
import shutil
import subprocess
import sys
import tempfile
import time
from typing import Any

import httpx
import qrcode

from boss_cli.constants import (
    AUTH_HEALTH_CACHE_TTL_S,
    BASE_URL,
    CONFIG_DIR,
    CREDENTIAL_FILE,
    HEADERS,
    REQUIRED_COOKIES,
    QR_CODE_URL,
    QR_DISPATCHER_URL,
    QR_RANDKEY_URL,
    QR_SCAN_LOGIN_URL,
    QR_SCAN_URL,
)

logger = logging.getLogger(__name__)

# Credential TTL: warn and attempt refresh after 7 days
CREDENTIAL_TTL_DAYS = 7
_CREDENTIAL_TTL_SECONDS = CREDENTIAL_TTL_DAYS * 86400

# QR poll config
POLL_TIMEOUT_S = 240  # 4 minutes
_AUTH_HEALTH_CACHE: dict[str, tuple[float, dict[str, Any]]] = {}


# ── Credential data class ───────────────────────────────────────────

class Credential:
    """Holds Boss Zhipin session cookies."""

    def __init__(self, cookies: dict[str, str]):
        self.cookies = cookies

    @property
    def is_valid(self) -> bool:
        return bool(self.cookies)

    @property
    def missing_required_cookies(self) -> list[str]:
        return sorted(REQUIRED_COOKIES - set(self.cookies))

    @property
    def has_required_cookies(self) -> bool:
        return not self.missing_required_cookies

    def to_dict(self) -> dict[str, Any]:
        return {"cookies": self.cookies, "saved_at": time.time()}

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> Credential:
        return cls(cookies=data.get("cookies", {}))

    def as_cookie_header(self) -> str:
        return "; ".join(f"{k}={v}" for k, v in self.cookies.items())


# ── Credential persistence ──────────────────────────────────────────

def save_credential(credential: Credential) -> None:
    """Save credential to config file."""
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)
    CREDENTIAL_FILE.write_text(json.dumps(credential.to_dict(), indent=2, ensure_ascii=False), encoding="utf-8")
    CREDENTIAL_FILE.chmod(0o600)
    logger.info("Credential saved to %s", CREDENTIAL_FILE)


def load_credential() -> Credential | None:
    """Load credential from saved file with TTL-based auto-refresh.

    If saved cookies are older than 7 days, automatically attempt to
    refresh from the browser before falling back to stale cookies.
    """
    if not CREDENTIAL_FILE.exists():
        return None
    try:
        data = json.loads(CREDENTIAL_FILE.read_text(encoding="utf-8"))
        cred = Credential.from_dict(data)
        if not cred.is_valid:
            return None
        if not cred.has_required_cookies:
            missing = cred.missing_required_cookies
            # __zp_stoken__ is generated by client-side JS and cannot be
            # obtained via QR login.  Don't reject an otherwise valid
            # credential just because this one cookie is absent.
            if missing != ["__zp_stoken__"]:
                logger.warning(
                    "Saved credential missing required cookies: %s",
                    ", ".join(missing),
                )
                clear_credential()
                return None
            logger.debug("Credential missing __zp_stoken__ (JS-generated), continuing")

        # Check TTL — auto-refresh if stale
        saved_at = data.get("saved_at", 0)
        if saved_at and (time.time() - saved_at) > _CREDENTIAL_TTL_SECONDS:
            logger.info(
                "Credential older than %d days, attempting browser refresh",
                CREDENTIAL_TTL_DAYS,
            )
            fresh, _ = extract_browser_credential()
            if fresh:
                logger.info("Auto-refreshed credential from browser")
                return fresh
            logger.warning(
                "Cookie refresh failed; using existing cookies (age: %d+ days)",
                CREDENTIAL_TTL_DAYS,
            )
        return cred
    except (json.JSONDecodeError, KeyError) as e:
        logger.warning("Failed to load saved credential: %s", e)
    return None


def clear_credential() -> None:
    """Remove saved credential file."""
    if CREDENTIAL_FILE.exists():
        CREDENTIAL_FILE.unlink()
        logger.info("Credential removed: %s", CREDENTIAL_FILE)
    _AUTH_HEALTH_CACHE.clear()


# ── Keychain / environment diagnostics ──────────────────────────────

_KEYCHAIN_ERROR_KEYWORDS = (
    "key for cookie decryption",
    "safe storage",
    "keychain",
    "secretstorage",
    "dpapi",
    "cryptunprotectdata",
    "win32crypt",
)


def _diagnose_extraction_issues(diagnostics: list[str]) -> str | None:
    """Analyse extraction diagnostics for platform-specific issues.

    Returns a user-friendly hint string, or None.
    """
    lowered = " ".join(diagnostics).lower()
    if not any(kw in lowered for kw in _KEYCHAIN_ERROR_KEYWORDS):
        return None

    is_ssh = bool(os.environ.get("SSH_CLIENT") or os.environ.get("SSH_TTY") or os.environ.get("SSH_CONNECTION"))

    if sys.platform == "darwin":
        if is_ssh:
            return (
                "macOS Keychain is locked (SSH session detected).\n"
                "  Fix: security unlock-keychain ~/Library/Keychains/login.keychain-db\n"
                "  Then retry the command."
            )
        return (
            "macOS Keychain permission denied — your terminal is not authorized to read browser cookie encryption keys.\n"
            "  Fix: Open Keychain Access → search for \"<Browser> Safe Storage\" → Access Control → add your Terminal app.\n"
            "  Or click \"Always Allow\" when the Keychain authorization popup appears."
        )
    if sys.platform == "win32":
        return (
            "Windows DPAPI cookie decryption failed.\n"
            "  Possible causes:\n"
            "  1. Chrome is running (locks the cookie database). Try closing Chrome first.\n"
            "  2. browser_cookie3 may not support the latest Chrome cookie encryption format.\n"
            "  3. If Chrome was running, admin privileges may be required for VSS shadowcopy.\n"
            "  Workaround: Set BOSS_COOKIES environment variable manually (see boss login --help)."
        )
    # Linux: gnome-keyring / SecretStorage issues
    return (
        "System keyring access failed — the cookie encryption key could not be retrieved.\n"
        "  If running headless or via SSH, ensure your keyring daemon is unlocked."
    )


# ── Environment variable fallback ───────────────────────────────────

def load_from_env() -> Credential | None:
    """Load cookies from BOSS_COOKIES environment variable.

    Format: "key1=val1; key2=val2; ..."
    """
    raw = os.environ.get("BOSS_COOKIES", "").strip()
    if not raw:
        return None
    cookies: dict[str, str] = {}
    for part in raw.split(";"):
        part = part.strip()
        if "=" not in part:
            continue
        k, v = part.split("=", 1)
        k, v = k.strip(), v.strip()
        if k and v:
            cookies[k] = v
    if not cookies:
        logger.debug("BOSS_COOKIES env set but no valid key=value pairs found")
        return None
    cred = Credential(cookies=cookies)
    logger.info("Loaded %d cookies from BOSS_COOKIES environment variable", len(cookies))
    return cred


# ── Browser cookie extraction ───────────────────────────────────────

# Chromium-based browser base directories
_CHROMIUM_BASE_DIRS: dict[str, str] = {
    "chrome": os.path.join("Google", "Chrome"),
    "edge": "Microsoft Edge",
    "brave": os.path.join("BraveSoftware", "Brave-Browser"),
}

# Default browser order for extraction
_DEFAULT_BROWSER_ORDER = ["chrome", "edge", "firefox", "brave"]

# Optional browsers (added if browser_cookie3 supports them)
_OPTIONAL_BROWSERS = [("Arc", "arc"), ("Chromium", "chromium"), ("Vivaldi", "vivaldi"), ("Opera", "opera")]


def _get_browser_order(cookie_source: str | None = None) -> list[str]:
    """Return browser extraction order, optionally prioritizing a specific browser."""
    if cookie_source:
        target = cookie_source.lower()
        return [target] + [b for b in _DEFAULT_BROWSER_ORDER if b != target]
    return _DEFAULT_BROWSER_ORDER


def _iter_chrome_cookie_files(browser_name: str) -> list[str]:
    """Return cookie file paths for all Chrome profiles."""
    base_dir = _CHROMIUM_BASE_DIRS.get(browser_name)
    if base_dir is None:
        return []

    if sys.platform == "darwin":
        root = os.path.join(os.path.expanduser("~"), "Library", "Application Support", base_dir)
    elif sys.platform == "win32":
        if browser_name == "edge":
            root = os.path.join(os.environ.get("LOCALAPPDATA", ""), "Microsoft", "Edge", "User Data")
        else:
            root = os.path.join(os.environ.get("LOCALAPPDATA", ""), base_dir)
    else:
        if browser_name == "edge":
            root = os.path.join(os.path.expanduser("~"), ".config", "microsoft-edge")
        else:
            root = os.path.join(os.path.expanduser("~"), ".config", base_dir)

    if not os.path.isdir(root):
        return []

    paths: list[str] = []
    default_cookies = os.path.join(root, "Default", "Cookies")
    if os.path.exists(default_cookies):
        paths.append(default_cookies)

    profile_dirs = sorted(glob.glob(os.path.join(root, "Profile *")))
    for profile_dir in profile_dirs:
        cookie_file = os.path.join(profile_dir, "Cookies")
        if os.path.exists(cookie_file):
            paths.append(cookie_file)

    return paths


def _extract_cookies_from_jar(jar: Any, source: str = "unknown") -> dict[str, str] | None:
    """Extract zhipin.com cookies from a browser_cookie3 cookie jar."""
    cookies: dict[str, str] = {}
    for cookie in jar:
        domain = cookie.domain or ""
        if "zhipin.com" in domain:
            if cookie.name and cookie.value:
                cookies[cookie.name] = cookie.value
    if cookies:
        logger.debug("Found %d zhipin cookies from %s", len(cookies), source)
        return cookies
    return None


def _extract_in_process(cookie_source: str | None = None) -> tuple[Credential | None, list[str]]:
    """Extract cookies in the main process.

    On macOS, Chrome encrypts cookies using a key stored in the system Keychain.
    Child processes do NOT inherit the parent's Keychain authorization, so
    browser_cookie3 must run in the main process to decrypt cookies.

    Returns (Credential | None, diagnostics_list).
    """
    try:
        import browser_cookie3 as bc3
    except ImportError:
        logger.debug("browser_cookie3 not installed, skipping in-process extraction")
        return None, ["browser-cookie3 not installed"]

    browser_fns: dict[str, Any] = {
        "chrome": bc3.chrome,
        "firefox": bc3.firefox,
        "edge": bc3.edge,
        "brave": bc3.brave,
    }
    # Add optional browsers if supported
    for display_name, attr in _OPTIONAL_BROWSERS:
        fn = getattr(bc3, attr, None)
        if fn:
            browser_fns[attr] = fn

    diagnostics: list[str] = []
    attempts: list[str] = []

    for name in _get_browser_order(cookie_source):
        fn = browser_fns.get(name)
        if fn is None:
            continue

        if name in _CHROMIUM_BASE_DIRS:
            # Chromium-based: iterate all profiles
            cookie_files = _iter_chrome_cookie_files(name)
            if not cookie_files:
                # No profile dirs found — try the default (no cookie_file arg)
                try:
                    jar = fn(domain_name=".zhipin.com")
                except Exception as e:
                    logger.debug("%s in-process extraction failed: %s", name, e)
                    attempts.append(f"{name}={type(e).__name__}")
                    diagnostics.append(f"{name}: {e}")
                    continue
                cookies = _extract_cookies_from_jar(jar, source=f"{name}(in-process)")
                if cookies:
                    cred = Credential(cookies=cookies)
                    logger.info("Found cookies in %s (in-process, default)", name)
                    return cred, diagnostics
                attempts.append(f"{name}=no-cookies")
                continue

            for cookie_file in cookie_files:
                profile_name = os.path.basename(os.path.dirname(cookie_file))
                try:
                    jar = fn(cookie_file=cookie_file, domain_name=".zhipin.com")
                except Exception as e:
                    logger.debug("%s[%s] in-process extraction failed: %s", name, profile_name, e)
                    attempts.append(f"{name}[{profile_name}]={type(e).__name__}")
                    diagnostics.append(f"{name}[{profile_name}]: {e}")
                    continue
                cookies = _extract_cookies_from_jar(jar, source=f"{name}[{profile_name}](in-process)")
                if cookies:
                    cred = Credential(cookies=cookies)
                    logger.info("Found cookies in %s profile '%s' (in-process)", name, profile_name)
                    return cred, diagnostics
                attempts.append(f"{name}[{profile_name}]=no-cookies")
        else:
            # Non-Chromium (Firefox): use default behavior
            try:
                jar = fn(domain_name=".zhipin.com")
            except Exception as e:
                logger.debug("%s in-process extraction failed: %s", name, e)
                attempts.append(f"{name}={type(e).__name__}")
                diagnostics.append(f"{name}: {e}")
                continue
            cookies = _extract_cookies_from_jar(jar, source=f"{name}(in-process)")
            if cookies:
                cred = Credential(cookies=cookies)
                logger.info("Found cookies in %s (in-process)", name)
                return cred, diagnostics
            attempts.append(f"{name}=no-cookies")

    if attempts:
        logger.debug("In-process extraction attempts: %s", ", ".join(attempts))
    return None, diagnostics


def _extract_via_subprocess(cookie_source: str | None = None) -> tuple[Credential | None, list[str]]:
    """Extract cookies via subprocess (fallback if in-process fails, e.g. SQLite lock).

    Returns (Credential | None, diagnostics_list).
    """
    extract_script = '''
import glob, json, os, sys
try:
    import browser_cookie3 as bc3
except ImportError:
    print(json.dumps({"error": "not_installed"}))
    sys.exit(0)

target = sys.argv[1] if len(sys.argv) > 1 else None

CHROMIUM_BASE_DIRS = {
    "chrome": os.path.join("Google", "Chrome"),
    "edge": "Microsoft Edge",
    "brave": os.path.join("BraveSoftware", "Brave-Browser"),
}

def iter_cookie_files(browser_name):
    base_dir = CHROMIUM_BASE_DIRS.get(browser_name)
    if base_dir is None:
        return []
    if sys.platform == "darwin":
        root = os.path.join(os.path.expanduser("~"), "Library", "Application Support", base_dir)
    elif sys.platform == "win32":
        if browser_name == "edge":
            root = os.path.join(os.environ.get("LOCALAPPDATA", ""), "Microsoft", "Edge", "User Data")
        else:
            root = os.path.join(os.environ.get("LOCALAPPDATA", ""), base_dir)
    else:
        if browser_name == "edge":
            root = os.path.join(os.path.expanduser("~"), ".config", "microsoft-edge")
        else:
            root = os.path.join(os.path.expanduser("~"), ".config", base_dir)
    if not os.path.isdir(root):
        return []
    paths = []
    d = os.path.join(root, "Default", "Cookies")
    if os.path.exists(d):
        paths.append(d)
    for pd in sorted(glob.glob(os.path.join(root, "Profile *"))):
        cf = os.path.join(pd, "Cookies")
        if os.path.exists(cf):
            paths.append(cf)
    return paths

browsers = [
    ("chrome", bc3.chrome),
    ("firefox", bc3.firefox),
    ("edge", bc3.edge),
    ("brave", bc3.brave),
]
for name, attr in [("arc", "arc"), ("chromium", "chromium"), ("vivaldi", "vivaldi"), ("opera", "opera")]:
    fn = getattr(bc3, attr, None)
    if fn:
        browsers.append((name, fn))

if target:
    target_lower = target.lower()
    browsers = [(n, fn) for n, fn in browsers if n.lower() == target_lower]
    if not browsers:
        print(json.dumps({"error": f"unsupported_browser: {target}"}))
        sys.exit(0)

attempts = []
for name, loader in browsers:
    if name in CHROMIUM_BASE_DIRS:
        cookie_files = iter_cookie_files(name)
        if not cookie_files:
            try:
                cj = loader(domain_name=".zhipin.com")
                cookies = {c.name: c.value for c in cj if "zhipin.com" in (c.domain or "")}
                if cookies:
                    print(json.dumps({"browser": name, "cookies": cookies}))
                    sys.exit(0)
                attempts.append(f"{name}=no-cookies")
            except Exception as exc:
                attempts.append(f"{name}={type(exc).__name__}: {exc}")
            continue
        for cf in cookie_files:
            pname = os.path.basename(os.path.dirname(cf))
            try:
                cj = loader(cookie_file=cf, domain_name=".zhipin.com")
                cookies = {c.name: c.value for c in cj if "zhipin.com" in (c.domain or "")}
                if cookies:
                    print(json.dumps({"browser": name, "cookies": cookies}))
                    sys.exit(0)
                attempts.append(f"{name}[{pname}]=no-cookies")
            except Exception as exc:
                attempts.append(f"{name}[{pname}]={type(exc).__name__}: {exc}")
    else:
        try:
            cj = loader(domain_name=".zhipin.com")
            cookies = {c.name: c.value for c in cj if "zhipin.com" in (c.domain or "")}
            if cookies:
                print(json.dumps({"browser": name, "cookies": cookies}))
                sys.exit(0)
            attempts.append(f"{name}=no-cookies")
        except Exception as exc:
            attempts.append(f"{name}={type(exc).__name__}: {exc}")

print(json.dumps({"error": "no_cookies", "attempts": attempts}))
'''

    diagnostics: list[str] = []
    try:
        cmd = [sys.executable, "-c", extract_script]
        if cookie_source:
            cmd.append(cookie_source)

        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=15,
        )

        if result.returncode != 0:
            logger.debug("Cookie extraction subprocess failed: %s", result.stderr)
            return None, diagnostics

        output = result.stdout.strip()
        if not output:
            return None, diagnostics

        data = json.loads(output)
        if "error" in data:
            attempts = data.get("attempts") or []
            if attempts:
                logger.debug("Subprocess extraction attempts: %s", ", ".join(str(a) for a in attempts))
                diagnostics.extend(str(a) for a in attempts)
            if data["error"] == "not_installed":
                logger.debug("browser-cookie3 not installed, skipping")
            else:
                logger.debug("No valid Boss Zhipin cookies found: %s", data["error"])
            return None, diagnostics

        cookies = data["cookies"]
        browser_name = data["browser"]
        cred = Credential(cookies=cookies)
        logger.info("Found cookies in %s (subprocess, %d cookies)", browser_name, len(cookies))
        return cred, diagnostics

    except subprocess.TimeoutExpired:
        logger.warning("Cookie extraction timed out (browser may be running)")
        diagnostics.append("subprocess: timed out")
        return None, diagnostics
    except (json.JSONDecodeError, KeyError) as e:
        logger.warning("Cookie extraction parse error: %s", e)
        return None, diagnostics


def extract_browser_credential(cookie_source: str | None = None) -> tuple[Credential | None, list[str]]:
    """Extract Boss Zhipin cookies from local browsers.

    Strategy:
    1. Try in-process first (required on macOS for Keychain access)
    2. Fall back to subprocess (handles SQLite lock when browser is running)

    Args:
        cookie_source: Optional browser name to extract from (e.g., 'chrome', 'firefox').
                       If None, tries all supported browsers in order.

    Returns:
        (Credential | None, diagnostics_list)
    """
    all_diagnostics: list[str] = []

    # 1. In-process (works on macOS, may fail with SQLite lock)
    cred, diag = _extract_in_process(cookie_source)
    all_diagnostics.extend(diag)
    if cred:
        if not cred.has_required_cookies:
            logger.warning(
                "In-process cookies missing required keys: %s",
                ", ".join(cred.missing_required_cookies),
            )
        else:
            save_credential(cred)
            return cred, all_diagnostics

    # 2. Subprocess fallback (handles SQLite lock, but fails on macOS Keychain)
    logger.debug("In-process extraction failed, trying subprocess fallback")
    cred, diag = _extract_via_subprocess(cookie_source)
    all_diagnostics.extend(diag)
    if cred:
        if not cred.has_required_cookies:
            logger.warning(
                "Subprocess cookies missing required keys: %s",
                ", ".join(cred.missing_required_cookies),
            )
        else:
            save_credential(cred)
            return cred, all_diagnostics

    if not cred:
        logger.warning("Cookie extraction failed in both in-process and subprocess modes")
    return None, all_diagnostics


# ── QR Code terminal rendering ──────────────────────────────────────

def _render_qr_half_blocks(matrix: list[list[bool]]) -> str:
    """Render QR matrix using Unicode half-block characters (▀▄█ and space).

    Same approach as xiaohongshu-cli: two rows are combined into one
    terminal line using half-block glyphs, halving the vertical space.
    """
    if not matrix:
        return ""

    # Add 1-module quiet zone
    size = len(matrix)
    padded = [[False] * (size + 2)]
    for row in matrix:
        padded.append([False] + list(row) + [False])
    padded.append([False] * (size + 2))
    matrix = padded
    rows = len(matrix)

    # Check terminal width
    term_cols = shutil.get_terminal_size(fallback=(80, 24)).columns
    qr_width = len(matrix[0])
    if qr_width > term_cols:
        logger.warning("Terminal too narrow (%d) for QR (%d)", term_cols, qr_width)
        return ""

    lines: list[str] = []
    for y in range(0, rows, 2):
        line = ""
        top_row = matrix[y]
        bottom_row = matrix[y + 1] if y + 1 < rows else [False] * len(top_row)
        for x in range(len(top_row)):
            top = top_row[x]
            bottom = bottom_row[x]
            if top and bottom:
                line += "█"
            elif top and not bottom:
                line += "▀"
            elif not top and bottom:
                line += "▄"
            else:
                line += " "
        lines.append(line)
    return "\n".join(lines)


def _display_qr_in_terminal(data: str) -> bool:
    """Display *data* as a QR code in the terminal using Unicode half-blocks.

    Returns True on success.
    """
    qr = qrcode.QRCode(error_correction=qrcode.constants.ERROR_CORRECT_L)
    qr.add_data(data)
    qr.make(fit=True)
    modules = qr.get_matrix()

    rendered = _render_qr_half_blocks(modules)
    if rendered:
        print(rendered)
        return True

    # Fallback to basic ASCII
    qr2 = qrcode.QRCode(
        error_correction=qrcode.constants.ERROR_CORRECT_L,
        box_size=1,
        border=1,
    )
    qr2.add_data(data)
    qr2.make(fit=True)
    qr2.print_ascii(invert=True)
    return True


def _open_image_file(path: str) -> None:
    """Open an image file with the system default viewer."""
    system = platform.system()
    try:
        if system == "Darwin":
            subprocess.Popen(["open", path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        elif system == "Windows":
            os.startfile(path)  # type: ignore[attr-defined]
        else:
            subprocess.Popen(["xdg-open", path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    except OSError as exc:
        logger.debug("Failed to open QR image: %s", exc)


async def _fetch_and_display_qr(client: httpx.AsyncClient, qr_id: str) -> None:
    """Fetch the QR code image from Boss API and display it.

    The server-generated QR image contains the correct scannable content
    that the Boss Zhipin APP can recognise. We save it to a temp file and
    open it with the system image viewer, plus render it in the terminal
    as a fallback.
    """
    # Fetch QR image from API
    resp = await client.get(QR_CODE_URL, params={"content": qr_id})
    resp.raise_for_status()

    # Save to temp file
    tmp = tempfile.NamedTemporaryFile(suffix=".png", prefix="boss_qr_", delete=False)
    tmp.write(resp.content)
    tmp.close()
    logger.debug("QR image saved to %s", tmp.name)

    # Try to open with system viewer
    _open_image_file(tmp.name)
    print(f"  📁 二维码图片已保存到: {tmp.name}")

    # Also try terminal rendering — decode the image to find the encoded content
    try:
        from PIL import Image
        from pyzbar.pyzbar import decode as zbar_decode

        img = Image.open(tmp.name)
        decoded = zbar_decode(img)
        if decoded:
            qr_content = decoded[0].data.decode("utf-8")
            logger.debug("Decoded QR content: %s", qr_content)
            _display_qr_in_terminal(qr_content)
    except ImportError:
        # pyzbar / Pillow not installed — terminal QR not available, image viewer is enough
        logger.debug("pyzbar/Pillow not installed, skipping terminal QR rendering")
    except Exception as exc:
        logger.debug("Failed to decode QR image for terminal display: %s", exc)


# ── QR Login flow ───────────────────────────────────────────────────

async def _get_qr_session(client: httpx.AsyncClient) -> dict[str, str]:
    """Step 1: Get QR session (qrId, randKey, secretKey)."""
    resp = await client.post(QR_RANDKEY_URL)
    resp.raise_for_status()
    data = resp.json()
    if data.get("code") != 0:
        raise RuntimeError(f"Failed to get QR session: {data.get('message', 'Unknown error')}")
    return data["zpData"]


async def _wait_for_scan(client: httpx.AsyncClient, qr_id: str) -> bool:
    """Step 3: Long-poll waiting for QR scan."""
    try:
        resp = await client.get(QR_SCAN_URL, params={"uuid": qr_id}, timeout=35)
        resp.raise_for_status()
        data = resp.json()
        return data.get("scaned", False)
    except httpx.ReadTimeout:
        return False


async def _wait_for_confirm(client: httpx.AsyncClient, qr_id: str) -> bool:
    """Step 4: Long-poll waiting for login confirmation.

    Must check the ``login`` field in the JSON body — a 200 status alone
    does NOT mean the user has confirmed on their phone.
    """
    try:
        resp = await client.get(QR_SCAN_LOGIN_URL, params={"qrId": qr_id}, timeout=35)
        resp.raise_for_status()
        data = resp.json()
        return data.get("login", False) is True
    except httpx.ReadTimeout:
        return False


async def _dispatch_login(client: httpx.AsyncClient, qr_id: str) -> Credential:
    """Step 5: Get final login cookies via dispatcher."""
    resp = await client.get(
        QR_DISPATCHER_URL,
        params={"qrId": qr_id, "pk": "header-login"},
    )
    resp.raise_for_status()

    # Extract cookies from response
    cookies = {}
    for name, value in resp.cookies.items():
        cookies[name] = value

    # Also grab cookies accumulated on the client
    for name, value in client.cookies.items():
        cookies[name] = value

    # QR dispatcher can complete before the web session is fully hydrated.
    # Visit the site once to collect any additional auth cookies such as __zp_stoken__.
    try:
        warmup = await client.get("/", timeout=15)
        warmup.raise_for_status()
        for name, value in warmup.cookies.items():
            cookies[name] = value
        for name, value in client.cookies.items():
            cookies[name] = value
    except httpx.HTTPError as exc:
        logger.debug("QR warmup request failed: %s", exc)

    if not cookies:
        raise RuntimeError("Login dispatcher returned no cookies")

    credential = Credential(cookies=cookies)

    # __zp_stoken__ is generated by client-side JavaScript and cannot be
    # obtained via the QR HTTP flow.  Do NOT try to supplement it from
    # browser cookies — the browser's stoken is tied to its own session
    # and will always mismatch the fresh QR session's wt2/wbg/zp_at.
    if not credential.has_required_cookies:
        missing = credential.missing_required_cookies
        if missing == ["__zp_stoken__"]:
            logger.warning(
                "QR login obtained session cookies but __zp_stoken__ is "
                "unavailable (generated by JS). Some APIs may return code=37. "
                "Re-run `boss login` from a browser session to fix."
            )
        else:
            raise RuntimeError(
                "二维码登录未拿到完整的 Web 登录态，缺少关键 Cookie: "
                f"{', '.join(missing)}。请先在浏览器完成登录后重新运行 boss login。"
            )

    return credential


async def qr_login() -> Credential:
    """Full QR code login flow.

    1. Get QR session
    2. Display QR code in terminal (Unicode half-blocks)
    3. Wait for scan (long-polling)
    4. Wait for confirm (long-polling)
    5. Dispatch to get cookies
    """
    async with httpx.AsyncClient(
        base_url=BASE_URL,
        headers=HEADERS,
        follow_redirects=True,
        timeout=httpx.Timeout(30, read=40),
    ) as client:
        # Step 1: Get QR session
        session = await _get_qr_session(client)
        qr_id = session["qrId"]

        # Step 2: Fetch QR image from API and display
        print("\n📱 请使用 Boss 直聘 APP 扫描以下二维码登录:\n")
        await _fetch_and_display_qr(client, qr_id)
        print("\n⏳ 扫码后请在手机上确认登录...")
        print(f"   (QR ID: {qr_id[:20]}...)\n")

        # Step 3: Wait for scan
        max_retries = 6  # ~3 min with 30s timeout each
        scanned = False
        for _ in range(max_retries):
            scanned = await _wait_for_scan(client, qr_id)
            if scanned:
                print("  📲 已扫码，请在手机上确认...")
                break

        if not scanned:
            raise RuntimeError("二维码已过期，请重试 (boss login)")

        # Step 4: Wait for confirm
        confirmed = False
        for _ in range(max_retries):
            confirmed = await _wait_for_confirm(client, qr_id)
            if confirmed:
                break

        if not confirmed:
            raise RuntimeError("确认超时，请重试 (boss login)")

        # Step 5: Dispatch
        credential = await _dispatch_login(client, qr_id)
        save_credential(credential)
        print("\n✅ 登录成功！凭证已保存到", CREDENTIAL_FILE)
        return credential


# ── Unified get_credential ──────────────────────────────────────────

def get_credential() -> Credential | None:
    """Try all auth methods and return credential.

    1. Saved credential file
    2. Environment variable (BOSS_COOKIES)
    3. Browser cookie extraction
    """
    cred = load_credential()
    if cred:
        logger.info("Loaded credential from %s", CREDENTIAL_FILE)
        return cred

    cred = load_from_env()
    if cred:
        logger.info("Loaded credential from BOSS_COOKIES env")
        save_credential(cred)
        return cred

    cred, _ = extract_browser_credential()
    if cred:
        logger.info("Extracted credential from browser")
        return cred

    return None


def _credential_cache_key(credential: Credential) -> str:
    payload = json.dumps(sorted(credential.cookies.items()), ensure_ascii=False, separators=(",", ":"))
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def verify_credential_details(credential: Credential, *, force_refresh: bool = False) -> dict[str, Any]:
    """Verify credential health across the key authenticated flows."""
    if not credential.has_required_cookies:
        missing = ", ".join(credential.missing_required_cookies)
        return {
            "authenticated": False,
            "search_authenticated": False,
            "recommend_authenticated": False,
            "reason": f"缺少关键 Cookie: {missing}",
        }

    from .client import BossClient
    from .exceptions import BossApiError, SessionExpiredError

    cache_key = _credential_cache_key(credential)
    now = time.time()
    if not force_refresh:
        cached = _AUTH_HEALTH_CACHE.get(cache_key)
        if cached and (now - cached[0]) <= AUTH_HEALTH_CACHE_TTL_S:
            return dict(cached[1])

    checks = {
        "search_authenticated": False,
        "recommend_authenticated": False,
    }
    failures: list[str] = []

    with BossClient(credential, request_delay=0.2) as client:
        try:
            client.search_jobs(query="Python", city="100010000", page=1, page_size=1)
            checks["search_authenticated"] = True
        except SessionExpiredError as exc:
            failures.append(f"search: {exc}")
        except BossApiError as exc:
            failures.append(f"search: 登录态校验失败: {exc}")

        try:
            client.get_recommend_jobs(page=1)
            checks["recommend_authenticated"] = True
        except SessionExpiredError as exc:
            failures.append(f"recommend: {exc}")
        except BossApiError as exc:
            failures.append(f"recommend: 登录态校验失败: {exc}")

    authenticated = checks["search_authenticated"]
    result: dict[str, Any] = {
        "authenticated": authenticated,
        **checks,
    }
    if failures:
        result["reason"] = "; ".join(failures)
    _AUTH_HEALTH_CACHE[cache_key] = (time.time(), dict(result))
    return result


def verify_credential(credential: Credential, *, force_refresh: bool = False) -> tuple[bool, str | None]:
    """Verify that the credential can access an authenticated API."""
    result = verify_credential_details(credential, force_refresh=force_refresh)
    return result["authenticated"], result.get("reason")

"""Browser-assisted login enhancement via Camoufox.

Hybrid approach:
1. Complete the QR login flow via HTTP (httpx) to obtain session cookies
   (wt2, wbg, zp_at).
2. Inject those cookies into a Camoufox browser and navigate to the site
   so that client-side JavaScript generates ``__zp_stoken__``.
3. Export all cookies from the browser context.

This gives us the complete cookie set that pure HTTP cannot achieve.

NOTE: Boss Zhipin uses aggressive anti-bot detection that may prevent
``__zp_stoken__`` generation even in Camoufox.  The QR login still
works without it for most APIs (recommend, chat, applied, etc.).
"""

from __future__ import annotations

import asyncio
import logging
import subprocess
import sys
from typing import Any

from .auth import Credential, qr_login, save_credential
from .constants import BASE_URL

logger = logging.getLogger(__name__)

# Cookie domains to export from browser
BROWSER_EXPORT_DOMAINS = (".zhipin.com", "zhipin.com", "www.zhipin.com")


class BrowserLoginUnavailable(RuntimeError):
    """Raised when the camoufox browser backend cannot be started."""


def _ensure_camoufox_ready() -> None:
    """Validate that the Camoufox package and browser binary are available."""
    try:
        import camoufox  # noqa: F401
    except ImportError as exc:
        raise BrowserLoginUnavailable(
            "camoufox 未安装。安装: pip install 'kabi-boss-cli[browser]'"
        ) from exc

    try:
        result = subprocess.run(
            [sys.executable, "-m", "camoufox", "path"],
            capture_output=True,
            text=True,
            timeout=15,
        )
    except (OSError, subprocess.SubprocessError) as exc:
        raise BrowserLoginUnavailable(
            "无法验证 Camoufox 浏览器安装状态。"
        ) from exc

    if result.returncode != 0 or not result.stdout.strip():
        raise BrowserLoginUnavailable(
            "Camoufox 浏览器运行时缺失。运行: python -m camoufox fetch"
        )


def _normalize_browser_cookies(raw_cookies: list[dict[str, Any]]) -> dict[str, str]:
    """Convert Playwright cookie entries into a flat dict, filtering to zhipin.com."""
    cookies: dict[str, str] = {}
    for entry in raw_cookies:
        name = entry.get("name")
        value = entry.get("value")
        domain = entry.get("domain", "")
        if not isinstance(name, str) or not isinstance(value, str):
            continue
        if not any(domain.endswith(d) for d in BROWSER_EXPORT_DOMAINS):
            continue
        cookies[name] = value
    return cookies


def _hydrate_stoken_via_browser(cookies: dict[str, str]) -> dict[str, str]:
    """Inject session cookies into a Camoufox browser and harvest __zp_stoken__.

    Boss Zhipin's client-side JS generates __zp_stoken__ on page load.
    We open a browser with the session cookies already set, visit the
    site, and let JS run.

    NOTE: This may fail if the anti-bot JS fingerprints the browser
    environment and refuses to generate the token.
    """
    from camoufox.sync_api import Camoufox

    playwright_cookies = []
    for name, value in cookies.items():
        playwright_cookies.append({
            "name": name,
            "value": value,
            "domain": ".zhipin.com",
            "path": "/",
        })

    with Camoufox(headless=True) as browser:
        context = browser.new_context()
        context.add_cookies(playwright_cookies)
        page = context.new_page()

        try:
            page.goto(f"{BASE_URL}/", wait_until="networkidle", timeout=20_000)
        except Exception:
            logger.debug("Camoufox page load did not reach networkidle")

        # Give JS time to set cookies
        try:
            page.wait_for_timeout(3000)
        except Exception:
            pass

        result = _normalize_browser_cookies(context.cookies())

    return result


def browser_qr_login(
    *,
    on_status: callable | None = None,
) -> Credential:
    """Hybrid QR login: HTTP for session + Camoufox for __zp_stoken__.

    1. Run the standard HTTP QR login flow (user scans in terminal)
    2. If __zp_stoken__ is missing, try headless Camoufox to generate it
    3. Return the credential (complete or partial)
    """
    _ensure_camoufox_ready()

    def _emit(msg: str) -> None:
        if on_status:
            on_status(msg)
        else:
            print(msg)

    # Step 1: Complete QR login via HTTP (reuse existing flow)
    cred = asyncio.run(qr_login())

    # Step 2: If __zp_stoken__ is missing, try to hydrate via browser
    if "__zp_stoken__" not in cred.cookies:
        _emit("\n🔧 正在通过浏览器补全 __zp_stoken__...")

        try:
            enriched = _hydrate_stoken_via_browser(cred.cookies)
        except Exception as exc:
            logger.warning("Browser __zp_stoken__ hydration failed: %s", exc)
            _emit("⚠️  浏览器补全 __zp_stoken__ 失败")
            return cred

        if "__zp_stoken__" in enriched:
            merged = {**cred.cookies, **enriched}
            cred = Credential(cookies=merged)
            save_credential(cred)
            _emit("✅ __zp_stoken__ 补全成功！所有接口可正常使用")
        else:
            _emit("⚠️  浏览器未能生成 __zp_stoken__（Boss 直聘反爬检测）")
            _emit("   recommend/chat/applied 等接口仍可使用，search 可能受限")

    return cred

"""CLI entry point for Boss CLI.

Usage:
    boss login / status / logout
    boss search <keyword> [--city C] [--salary S] [--exp E] [--degree D]
    boss recommend [--page N]
    boss me / applied / interviews / chat
    boss greet <securityId>
    boss batch-greet <keyword> [-n N] [--city C] [--dry-run]
    boss cities
"""

from __future__ import annotations

import logging

import click

from . import __version__
from .commands import auth, personal, recruiter, search, social


@click.group()
@click.version_option(version=__version__, prog_name="boss")
@click.option("-v", "--verbose", is_flag=True, help="Enable verbose logging (show request URLs, timing)")
@click.pass_context
def cli(ctx, verbose: bool) -> None:
    """Boss CLI — 在终端使用 BOSS 直聘 🤝"""
    ctx.ensure_object(dict)
    if verbose:
        logging.basicConfig(level=logging.INFO, format="%(name)s %(message)s")
    else:
        logging.basicConfig(level=logging.WARNING)


# ─── Auth commands ───────────────────────────────────────────────────

cli.add_command(auth.login)
cli.add_command(auth.logout)
cli.add_command(auth.status)
cli.add_command(auth.me)

# ─── Search & Browse commands ────────────────────────────────────────

cli.add_command(search.search)
cli.add_command(search.recommend)
cli.add_command(search.detail)
cli.add_command(search.show)
cli.add_command(search.export)
cli.add_command(search.history)
cli.add_command(search.cities)

# ─── Personal Center commands ────────────────────────────────────────

cli.add_command(personal.applied)
cli.add_command(personal.interviews)

# ─── Social commands ────────────────────────────────────────────────

cli.add_command(social.chat_list)
cli.add_command(social.greet)
cli.add_command(social.batch_greet)

# ─── Recruiter (Boss) commands ──────────────────────────────────────

cli.add_command(recruiter.recruiter)


if __name__ == "__main__":
    cli()

"""API client for Boss Zhipin with rate limiting, retry, and anti-detection."""

from __future__ import annotations

import logging
import random
import time
import urllib.parse
from collections import deque
from typing import Any

import httpx

from .constants import (
    BASE_URL,
    BOSS_CHAT_GEEK_INFO_URL,
    BOSS_CHATTED_JOB_LIST_URL,
    BOSS_EXCHANGE_CONTENT_URL,
    BOSS_EXCHANGE_REQUEST_URL,
    BOSS_FRIEND_DETAIL_URL,
    BOSS_FRIEND_LABELS_URL,
    BOSS_FRIEND_LIST_URL,
    BOSS_FRIEND_NOTE_URL,
    BOSS_GREET_REC_SORT_URL,
    BOSS_GREET_SORT_LIST_URL,
    BOSS_HISTORY_MSG_URL,
    BOSS_INTERVIEW_INVITE_URL,
    BOSS_INTERVIEW_LIST_URL,
    BOSS_JOB_OFFLINE_URL,
    BOSS_JOB_ONLINE_URL,
    BOSS_LAST_MSG_URL,
    BOSS_REMOVE_FILTER_URL,
    BOSS_SEARCH_GEEK_URL,
    BOSS_SEND_MSG_URL,
    BOSS_SESSION_ENTER_URL,
    BOSS_VIEW_GEEK_URL,
    CITY_CODES,
    DELIVER_LIST_URL,
    FRIEND_ADD_URL,
    FRIEND_LIST_URL,
    GEEK_GET_JOB_URL,
    HEADERS,
    INTERVIEW_DATA_URL,
    JOB_CARD_URL,
    JOB_DETAIL_URL,
    JOB_HISTORY_URL,
    JOB_SEARCH_URL,
    RESUME_BASEINFO_URL,
    RESUME_EXPECT_URL,
    RESUME_STATUS_URL,
    USER_INFO_URL,
    WEB_BOSS_CHAT_URL,
    WEB_GEEK_CHAT_URL,
    WEB_GEEK_HISTORY_URL,
    WEB_GEEK_JOB_URL,
    WEB_GEEK_RECOMMEND_URL,
)
from .exceptions import BossApiError, ParamError, RateLimitError, SessionExpiredError

logger = logging.getLogger(__name__)


class BossClient:
    """Boss Zhipin API client with Gaussian jitter, exponential backoff, and session-stable identity.

    Anti-detection strategy:
    - Gaussian jitter delay between requests (~1s mean, σ=0.3)
    - 5% chance of a random long pause (2-5s) to mimic reading behavior
    - Exponential backoff on HTTP 429/5xx (up to 3 retries)
    - Response cookies merged back into session jar
    - Request counter for monitoring
    """

    def __init__(
        self,
        credential: object | None = None,
        timeout: float = 30.0,
        request_delay: float = 1.0,
        max_retries: int = 3,
    ):
        self.credential = credential
        self._timeout = timeout
        self._request_delay = request_delay
        self._base_request_delay = request_delay
        self._max_retries = max_retries
        self._last_request_time = 0.0
        self._request_count = 0
        self._rate_limit_count = 0
        self._recent_request_times: deque[float] = deque(maxlen=12)
        self._http: httpx.Client | None = None

    def _build_client(self) -> httpx.Client:
        cookies = {}
        if self.credential:
            cookies = self.credential.cookies
        return httpx.Client(
            base_url=BASE_URL,
            headers=dict(HEADERS),
            cookies=cookies,
            follow_redirects=True,
            timeout=httpx.Timeout(self._timeout),
        )

    @property
    def client(self) -> httpx.Client:
        if not self._http:
            raise RuntimeError("Client not initialized. Use 'with BossClient() as client:'")
        return self._http

    def __enter__(self) -> BossClient:
        self._http = self._build_client()
        return self

    def __exit__(self, *args: Any) -> None:
        if self._http:
            self._http.close()
            self._http = None

    # ── Rate limiting ───────────────────────────────────────────────

    def _rate_limit_delay(self) -> None:
        """Enforce minimum delay with Gaussian jitter to mimic human browsing."""
        if self._request_delay <= 0:
            return
        elapsed = time.time() - self._last_request_time
        if elapsed < self._request_delay:
            # Gaussian jitter: mean=0.3, σ=0.15, clamped to [0, ∞)
            jitter = max(0, random.gauss(0.3, 0.15))
            # 5% chance of a long pause to mimic reading
            if random.random() < 0.05:
                jitter += random.uniform(2.0, 5.0)
            sleep_time = self._request_delay - elapsed + jitter
            logger.debug("Rate-limit delay: %.2fs", sleep_time)
            time.sleep(sleep_time)

        burst_penalty = self._burst_penalty_delay()
        if burst_penalty > 0:
            logger.debug("Burst penalty delay: %.2fs", burst_penalty)
            time.sleep(burst_penalty)

    def _burst_penalty_delay(self) -> float:
        """Add extra delay when a burst pattern looks less like human browsing."""
        if not self._recent_request_times:
            return 0.0

        now = time.time()
        recent_15s = sum(1 for ts in self._recent_request_times if now - ts <= 15)
        recent_45s = sum(1 for ts in self._recent_request_times if now - ts <= 45)

        if recent_45s >= 6:
            return random.uniform(4.0, 7.0)
        if recent_15s >= 3:
            return random.uniform(1.2, 2.8)
        return 0.0

    def _mark_request(self) -> None:
        now = time.time()
        self._last_request_time = now
        self._request_count += 1
        self._recent_request_times.append(now)

    @property
    def request_stats(self) -> dict[str, int | float]:
        """Return current request statistics."""
        return {
            "request_count": self._request_count,
            "last_request_time": self._last_request_time,
        }

    # ── Response handling ───────────────────────────────────────────

    def _merge_response_cookies(self, resp: httpx.Response) -> None:
        """Persist response Set-Cookie headers back into the session jar."""
        for name, value in resp.cookies.items():
            if value:
                self.client.cookies.set(name, value)

    def _headers_for_request(self, url: str, params: dict[str, Any] | None = None) -> dict[str, str]:
        """Build browser-like headers, including endpoint-specific Referer and zp_token."""
        headers = dict(HEADERS)
        # Add security headers that the boss web app sends with every request
        headers["X-Requested-With"] = "XMLHttpRequest"
        bst = self.client.cookies.get("bst", "")
        if bst:
            headers["zp_token"] = bst
        if url == JOB_SEARCH_URL:
            query = ""
            if params and params.get("query"):
                query = f"?{urllib.parse.urlencode({'query': params['query']})}"
            headers["Referer"] = f"{WEB_GEEK_JOB_URL}{query}"
        elif url == GEEK_GET_JOB_URL and params and params.get("tag") == 5:
            headers["Referer"] = WEB_GEEK_RECOMMEND_URL
        elif url == GEEK_GET_JOB_URL:
            headers["Referer"] = WEB_GEEK_CHAT_URL
        elif url in (JOB_CARD_URL, JOB_DETAIL_URL):
            headers["Referer"] = WEB_GEEK_JOB_URL
        elif url == JOB_HISTORY_URL:
            headers["Referer"] = WEB_GEEK_HISTORY_URL
        elif url in (FRIEND_LIST_URL, FRIEND_ADD_URL):
            headers["Referer"] = WEB_GEEK_CHAT_URL
        # Recruiter (boss) endpoints
        elif url == BOSS_SEARCH_GEEK_URL:
            headers["Referer"] = f"{BASE_URL}/web/chat/search"
        elif url in (BOSS_VIEW_GEEK_URL, BOSS_SEND_MSG_URL):
            headers["Referer"] = WEB_BOSS_CHAT_URL
        elif url in (BOSS_FRIEND_LIST_URL, BOSS_FRIEND_DETAIL_URL, BOSS_LAST_MSG_URL,
                      BOSS_HISTORY_MSG_URL, BOSS_CHAT_GEEK_INFO_URL, BOSS_FRIEND_LABELS_URL,
                      BOSS_FRIEND_NOTE_URL, BOSS_GREET_SORT_LIST_URL, BOSS_GREET_REC_SORT_URL,
                      BOSS_CHATTED_JOB_LIST_URL, BOSS_INTERVIEW_LIST_URL,
                      BOSS_EXCHANGE_REQUEST_URL, BOSS_EXCHANGE_CONTENT_URL,
                      BOSS_INTERVIEW_INVITE_URL, BOSS_REMOVE_FILTER_URL,
                      BOSS_SESSION_ENTER_URL):
            headers["Referer"] = WEB_BOSS_CHAT_URL
        return headers

    def _handle_response(self, data: dict[str, Any], action: str) -> dict[str, Any]:
        """Validate API response and return zpData, raise typed exceptions."""
        code = data.get("code", -1)

        if code == 0:
            return data.get("zpData", {})

        message = data.get("message", "Unknown error")

        if code == 37:
            raise SessionExpiredError()
        if code in (17, 19):
            raise ParamError(message, code=code)
        if code in (121, 122):
            raise BossApiError(
                f"{action}: 请求被安全系统拦截 (code={code})。"
                "此操作需要浏览器环境的安全验证，CLI 暂不支持。"
                "请在 BOSS直聘 网页端完成此操作。",
                code=code, response=data,
            )
        if code == 9:
            # Rate limited — auto-cooldown with exponential backoff
            self._rate_limit_count += 1
            cooldown = min(60, 10 * (2 ** (self._rate_limit_count - 1)))
            self._request_delay = max(self._request_delay, self._base_request_delay * 2)
            logger.warning(
                "Rate limited (count=%d), cooling down %.0fs, delay raised to %.1fs",
                self._rate_limit_count, cooldown, self._request_delay,
            )
            time.sleep(cooldown)
            raise RateLimitError()

        raise BossApiError(f"{action}: {message} (code={code})", code=code, response=data)

    # ── Request with retry ──────────────────────────────────────────

    def _request(self, method: str, url: str, **kwargs) -> dict[str, Any]:
        """Execute HTTP request with rate-limit delay, retry, and cookie merge."""
        self._rate_limit_delay()
        last_exc: Exception | None = None
        params = kwargs.get("params")
        merged_headers = self._headers_for_request(url, params=params)
        request_headers = kwargs.pop("headers", None)
        if request_headers:
            merged_headers.update(request_headers)

        for attempt in range(self._max_retries):
            t0 = time.time()
            try:
                resp = self.client.request(method, url, headers=merged_headers, **kwargs)
                elapsed = time.time() - t0
                self._merge_response_cookies(resp)
                self._mark_request()

                logger.info(
                    "[#%d] %s %s → %d (%.2fs)",
                    self._request_count, method, url[:60], resp.status_code, elapsed,
                )

                # Retry on server errors
                if resp.status_code in (429, 500, 502, 503, 504):
                    wait = (2 ** attempt) + random.uniform(0, 1)
                    logger.warning(
                        "HTTP %d from %s, retrying in %.1fs (attempt %d/%d)",
                        resp.status_code, url[:80], wait, attempt + 1, self._max_retries,
                    )
                    time.sleep(wait)
                    continue

                # For non-server errors (4xx except 404), raise immediately
                if resp.status_code == 404:
                    # Some endpoints return 404 when anti-bot blocks the request
                    text = resp.text
                    if text.strip().startswith("{"):
                        return resp.json()
                    raise BossApiError(f"接口不存在: {url} (HTTP 404)", code=404)

                resp.raise_for_status()

                # Check for HTML responses (redirect to login page)
                text = resp.text
                if text.startswith("<"):
                    raise BossApiError(f"Received HTML instead of JSON from {url} (possible auth redirect)")

                return resp.json()

            except (httpx.TimeoutException, httpx.NetworkError) as exc:
                elapsed = time.time() - t0
                last_exc = exc
                wait = (2 ** attempt) + random.uniform(0, 1)
                logger.warning(
                    "[#%d] %s %s → Network error: %s (%.2fs), retrying in %.1fs (attempt %d/%d)",
                    self._request_count + 1, method, url[:60], exc, elapsed, wait,
                    attempt + 1, self._max_retries,
                )
                time.sleep(wait)

        if last_exc:
            raise BossApiError(f"Request failed after {self._max_retries} retries: {last_exc}") from last_exc
        raise BossApiError(f"Request failed after {self._max_retries} retries")

    def _get(self, url: str, params: dict[str, Any] | None = None, action: str = "") -> dict[str, Any]:
        """GET request with response validation and rate-limit retry."""
        data = self._request("GET", url, params=params)
        try:
            result = self._handle_response(data, action)
            # Reset rate-limit counter on success
            self._rate_limit_count = 0
            return result
        except RateLimitError:
            # Auto-retry once after cooldown (cooldown already happened in _handle_response)
            logger.info("Retrying after rate-limit cooldown...")
            data = self._request("GET", url, params=params)
            result = self._handle_response(data, action)
            self._rate_limit_count = 0
            return result

    # ── Job Search & Browse ─────────────────────────────────────────

    def search_jobs(
        self,
        query: str,
        city: str = "101010100",
        page: int = 1,
        page_size: int = 15,
        experience: str | None = None,
        degree: str | None = None,
        salary: str | None = None,
        industry: str | None = None,
        scale: str | None = None,
        stage: str | None = None,
        job_type: str | None = None,
    ) -> dict[str, Any]:
        """Search jobs."""
        params: dict[str, Any] = {
            "query": query,
            "city": city,
            "page": page,
            "pageSize": page_size,
        }
        if experience:
            params["experience"] = experience
        if degree:
            params["degree"] = degree
        if salary:
            params["salary"] = salary
        if industry:
            params["industry"] = industry
        if scale:
            params["scale"] = scale
        if stage:
            params["stage"] = stage
        if job_type:
            params["jobType"] = job_type
        return self._get(JOB_SEARCH_URL, params=params, action="搜索职位")

    def get_recommend_jobs(self, page: int = 1) -> dict[str, Any]:
        """Get personalized job recommendations.

        The live web page currently loads recommendation cards from
        ``/wapi/zprelation/interaction/geekGetJob`` with tag=5 rather
        than the older ``/wapi/zpgeek/pc/recommend/job/list.json`` path.
        Normalize that payload back into the CLI's historical shape.
        """
        data = self._get(
            GEEK_GET_JOB_URL,
            params={"page": page, "tag": 5, "isActive": "true"},
            action="推荐职位",
        )
        if "jobList" in data:
            return data

        card_list = data.get("cardList", [])
        return {
            "jobList": card_list,
            "hasMore": data.get("hasMore", False),
            "totalCount": data.get("totalCount", len(card_list)),
            "page": data.get("page", page),
            "startIndex": data.get("startIndex", 0),
            "type": data.get("type", 2),
            "lid": data.get("lid", ""),
        }

    def get_job_card(self, security_id: str, lid: str) -> dict[str, Any]:
        """Get job card info (hover preview)."""
        return self._get(JOB_CARD_URL, params={"securityId": security_id, "lid": lid}, action="职位卡片")

    def get_job_detail(self, security_id: str, lid: str = "") -> dict[str, Any]:
        """Get detailed information for a specific job."""
        params: dict[str, str] = {"securityId": security_id}
        if lid:
            params["lid"] = lid
        return self._get(JOB_DETAIL_URL, params=params, action="职位详情")

    # ── Personal Center ─────────────────────────────────────────────

    def get_user_info(self) -> dict[str, Any]:
        """Get current user info (userId, name, avatar, etc.)."""
        return self._get(USER_INFO_URL, action="用户信息")

    def get_resume_baseinfo(self) -> dict[str, Any]:
        """Get resume basic info (full profile: name, age, degree, etc.)."""
        return self._get(RESUME_BASEINFO_URL, action="简历基本信息")

    def get_resume_expect(self) -> dict[str, Any]:
        """Get job expectations (desired position, salary, city)."""
        return self._get(RESUME_EXPECT_URL, action="求职期望")

    def get_resume_status(self) -> dict[str, Any]:
        """Get resume status."""
        return self._get(RESUME_STATUS_URL, action="简历状态")

    def get_deliver_list(self, page: int = 1) -> dict[str, Any]:
        """Get list of jobs applied to (已投递)."""
        return self._get(DELIVER_LIST_URL, params={"page": page}, action="已投递列表")

    def get_interview_data(self) -> dict[str, Any]:
        """Get interview data (面试)."""
        return self._get(INTERVIEW_DATA_URL, action="面试数据")

    def get_job_history(self, page: int = 1) -> dict[str, Any]:
        """Get job browsing history."""
        return self._get(JOB_HISTORY_URL, params={"page": page}, action="浏览历史")

    # ── Social / Chat ───────────────────────────────────────────────

    def get_friend_list(self) -> dict[str, Any]:
        """Get geek friend list (沟通过的 Boss)."""
        return self._get(FRIEND_LIST_URL, action="好友列表")

    def add_friend(self, security_id: str, lid: str = "") -> dict[str, Any]:
        """Send greeting to a Boss (打招呼 / 投递简历)."""
        params: dict[str, str] = {"securityId": security_id}
        if lid:
            params["lid"] = lid
        return self._get(FRIEND_ADD_URL, params=params, action="打招呼")

    def get_geek_job(self, security_id: str) -> dict[str, Any]:
        """Get interacted job info."""
        return self._get(GEEK_GET_JOB_URL, params={"securityId": security_id}, action="互动职位")

    # ── Recruiter (Boss) Mode ────────────────────────────────────────

    def _post(self, url: str, data: dict[str, Any] | None = None, action: str = "", json_body: bool = False) -> dict[str, Any]:
        """POST request with form-encoded or JSON body, response validation, and rate-limit retry."""
        kwargs = {"json": data} if json_body else {"data": data}
        resp = self._request("POST", url, **kwargs)
        try:
            result = self._handle_response(resp, action)
            self._rate_limit_count = 0
            return result
        except RateLimitError:
            logger.info("Retrying after rate-limit cooldown...")
            resp = self._request("POST", url, **kwargs)
            result = self._handle_response(resp, action)
            self._rate_limit_count = 0
            return result

    def get_boss_chatted_jobs(self) -> list[dict[str, Any]]:
        """Get list of jobs the boss has posted (chatted job list)."""
        return self._get(BOSS_CHATTED_JOB_LIST_URL, action="招聘职位列表")

    def get_boss_friend_list(self, label_id: int = 0, enc_job_id: str = "", sort: str = "", page: int = 1) -> dict[str, Any]:
        """Get boss friend list (candidates who have chatted)."""
        data: dict[str, Any] = {"labelId": label_id, "page": page}
        if enc_job_id:
            data["encJobId"] = enc_job_id
        if sort:
            data["sort"] = sort
        return self._post(BOSS_FRIEND_LIST_URL, data=data, action="候选人列表")

    def get_boss_friend_details(self, friend_ids: list[int]) -> dict[str, Any]:
        """Get detailed info for boss friends (candidates)."""
        ids_str = ",".join(str(fid) for fid in friend_ids)
        return self._post(BOSS_FRIEND_DETAIL_URL, data={"friendIds": ids_str}, action="候选人详情")

    def get_boss_last_messages(self, friend_ids: list[int], src: int = 0) -> list[dict[str, Any]]:
        """Get last message for each friend."""
        ids_str = ",".join(str(fid) for fid in friend_ids)
        return self._post(BOSS_LAST_MSG_URL, data={"friendIds": ids_str, "src": src}, action="最近消息")

    def get_boss_chat_history(self, gid: int, count: int = 20, max_msg_id: int = 0) -> dict[str, Any]:
        """Get chat history with a specific candidate."""
        params: dict[str, Any] = {"gid": gid, "c": count, "src": 0}
        if max_msg_id:
            params["maxMsgId"] = max_msg_id
        return self._get(BOSS_HISTORY_MSG_URL, params=params, action="聊天记录")

    def get_boss_chat_geek_info(
        self, encrypt_geek_id: str, security_id: str, job_id: int,
    ) -> dict[str, Any]:
        """Get detailed info for a candidate in chat context."""
        return self._get(
            BOSS_CHAT_GEEK_INFO_URL,
            params={"encryptGeekId": encrypt_geek_id, "securityId": security_id, "jobId": job_id},
            action="候选人信息",
        )

    def get_boss_friend_labels(self) -> dict[str, Any]:
        """Get recruiter's friend labels/tags."""
        return self._get(BOSS_FRIEND_LABELS_URL, action="标签列表")

    def get_boss_greet_list(self, enc_job_id: str = "", page: int = 1) -> dict[str, Any]:
        """Get list of new greetings (candidates who greeted the boss)."""
        params: dict[str, Any] = {"page": page}
        if enc_job_id:
            params["encJobId"] = enc_job_id
        return self._get(BOSS_GREET_SORT_LIST_URL, params=params, action="新招呼列表")

    def get_boss_greet_rec_list(self, enc_job_id: str = "", page: int = 1) -> dict[str, Any]:
        """Get recommended greeting sort list."""
        params: dict[str, Any] = {"page": page}
        if enc_job_id:
            params["encJobId"] = enc_job_id
        return self._get(BOSS_GREET_REC_SORT_URL, params=params, action="推荐招呼排序")

    def get_boss_interview_list(self) -> dict[str, Any]:
        """Get boss interview list."""
        return self._get(BOSS_INTERVIEW_LIST_URL, action="面试列表")

    def search_geeks(
        self, query: str, city: str = "101020100", page: int = 1,
        experience: str | None = None, degree: str | None = None,
        salary: str | None = None, encrypt_job_id: str = "",
    ) -> dict[str, Any]:
        """Search candidates (geeks) as a recruiter."""
        params: dict[str, Any] = {
            "query": query, "city": city, "page": page,
        }
        if encrypt_job_id:
            params["encryptJobId"] = encrypt_job_id
        if experience:
            params["experience"] = experience
        if degree:
            params["degree"] = degree
        if salary:
            params["salary"] = salary
        return self._get(BOSS_SEARCH_GEEK_URL, params=params, action="搜索候选人")

    def get_boss_recommend_geeks(self, page: int = 1, enc_job_id: str = "") -> dict[str, Any]:
        """Get recommended candidates (new greetings sorted by recommendation)."""
        params: dict[str, Any] = {"page": page}
        if enc_job_id:
            params["encJobId"] = enc_job_id
        return self._get(BOSS_GREET_REC_SORT_URL, params=params, action="推荐候选人")

    def get_boss_view_geek(
        self, encrypt_geek_id: str, encrypt_job_id: str, security_id: str = "",
    ) -> dict[str, Any]:
        """Get full candidate resume/profile view."""
        params: dict[str, Any] = {
            "encryptGeekId": encrypt_geek_id,
            "encryptJobId": encrypt_job_id,
        }
        if security_id:
            params["securityId"] = security_id
        return self._get(BOSS_VIEW_GEEK_URL, params=params, action="候选人简历")

    def boss_send_message(self, gid: int, content: str) -> dict[str, Any]:
        """Send a text message to a candidate as a recruiter."""
        return self._post(
            BOSS_SEND_MSG_URL,
            data={"gid": gid, "content": content},
            action="发送消息",
        )

    def boss_job_offline(self, encrypt_job_id: str) -> dict[str, Any]:
        """Take a job posting offline (close)."""
        return self._post(BOSS_JOB_OFFLINE_URL, data={"encryptJobId": encrypt_job_id}, action="关闭职位")

    def boss_job_online(self, encrypt_job_id: str) -> dict[str, Any]:
        """Bring a job posting online (reopen)."""
        return self._post(BOSS_JOB_ONLINE_URL, data={"encryptJobId": encrypt_job_id}, action="开启职位")

    # ── Recruiter Chat Actions ────────────────────────────────────────

    def boss_exchange_request(self, uid: int, job_id: int, exchange_type: int) -> dict[str, Any]:
        """Request exchange with candidate.

        exchange_type: 1=phone, 2=wechat, 3=resume
        """
        return self._post(
            BOSS_EXCHANGE_REQUEST_URL,
            data={"type": exchange_type, "uid": uid, "jobId": job_id, "gid": uid},
            action="交换请求",
        )

    def boss_get_exchange_content(self, uid: int) -> dict[str, Any]:
        """Get exchanged contact info (phone/wechat) for a candidate."""
        return self._post(
            BOSS_EXCHANGE_CONTENT_URL,
            data={"uid": uid},
            action="查看交换内容",
        )

    def boss_interview_invite(
        self, encrypt_geek_id: str, encrypt_job_id: str, security_id: str,
        address: str = "", start_time: str = "", description: str = "",
    ) -> dict[str, Any]:
        """Invite candidate for an interview."""
        data: dict[str, Any] = {
            "encryptGeekId": encrypt_geek_id,
            "encryptJobId": encrypt_job_id,
            "securityId": security_id,
        }
        if address:
            data["address"] = address
        if start_time:
            data["startTime"] = start_time
        if description:
            data["description"] = description
        return self._post(BOSS_INTERVIEW_INVITE_URL, data=data, action="约面试", json_body=True)

    def boss_mark_unsuitable(self, encrypt_geek_id: str, encrypt_job_id: str) -> dict[str, Any]:
        """Mark candidate as unsuitable."""
        return self._post(
            BOSS_REMOVE_FILTER_URL,
            data={"encryptGeekId": encrypt_geek_id, "encryptJobId": encrypt_job_id},
            action="标记不合适",
        )

    def boss_session_enter(self, geek_id: str, expect_id: str, job_id: str, security_id: str) -> dict[str, Any]:
        """Enter a chat session with a candidate (required before sending messages)."""
        return self._post(
            BOSS_SESSION_ENTER_URL,
            data={"geekId": geek_id, "expectId": expect_id, "jobId": job_id, "securityId": security_id},
            action="进入会话",
        )


# ── City resolution ─────────────────────────────────────────────────

def resolve_city(name: str) -> str:
    """Resolve city name to code, passthrough if already a code."""
    if name.isdigit() and len(name) >= 6:
        return name
    return CITY_CODES.get(name, CITY_CODES["全国"])


def list_cities() -> dict[str, str]:
    """Return all supported city name -> code mappings."""
    return dict(CITY_CODES)