Xhs Cli

name: CI

on:
  pull_request:
  push:
    branches:
      - main

jobs:
  lint-and-test:
    name: Lint and test (Python ${{ matrix.python-version }})
    runs-on: ubuntu-latest
    strategy:
      fail-fast: false
      matrix:
        python-version: ["3.10", "3.12"]

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}

      - name: Set up uv
        uses: astral-sh/setup-uv@v5

      - name: Install dependencies
        run: uv sync --dev

      - name: Run ruff
        run: uv run ruff check .

      - name: Run tests
        run: uv run pytest -q

name: Publish to PyPI

# Trigger: push a version tag like v0.1.0
on:
  push:
    tags:
      - "v*"

jobs:
  build:
    name: Build distribution
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install uv
        uses: astral-sh/setup-uv@v5

      - name: Build package
        run: uv build

      - name: Upload dist artifacts
        uses: actions/upload-artifact@v4
        with:
          name: dist
          path: dist/

  publish:
    name: Publish to PyPI
    needs: build
    runs-on: ubuntu-latest
    environment: pypi
    permissions:
      id-token: write  # Required for Trusted Publisher (OIDC)
    steps:
      - name: Download dist artifacts
        uses: actions/download-artifact@v4
        with:
          name: dist
          path: dist/

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

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

# Virtual environments
.venv/
venv/

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

# OS
.DS_Store
Thumbs.db

# Project specific
.xhs-cli/

                                 Apache License
                           Version 2.0, January 2004
                        http://www.apache.org/licenses/

   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION

   1. Definitions.

      "License" shall mean the terms and conditions for use, reproduction,
      and distribution as defined by Sections 1 through 9 of this document.

      "Licensor" shall mean the copyright owner or entity authorized by
      the copyright owner that is granting the License.

      "Legal Entity" shall mean the union of the acting entity and all
      other entities that control, are controlled by, or are under common
      control with that entity. For the purposes of this definition,
      "control" means (i) the power, direct or indirect, to cause the
      direction or management of such entity, whether by contract or
      otherwise, or (ii) ownership of fifty percent (50%) or more of the
      outstanding shares, or (iii) beneficial ownership of such entity.

      "You" (or "Your") shall mean an individual or Legal Entity
      exercising permissions granted by this License.

      "Source" form shall mean the preferred form for making modifications,
      including but not limited to software source code, documentation
      source, and configuration files.

      "Object" form shall mean any form resulting from mechanical
      transformation or translation of a Source form, including but
      not limited to compiled object code, generated documentation,
      and conversions to other media types.

      "Work" shall mean the work of authorship, whether in Source or
      Object form, made available under the License, as indicated by a
      copyright notice that is included in or attached to the work
      (an example is provided in the Appendix below).

      "Derivative Works" shall mean any work, whether in Source or Object
      form, that is based on (or derived from) the Work and for which the
      editorial revisions, annotations, elaborations, or other modifications
      represent, as a whole, an original work of authorship. For the purposes
      of this License, Derivative Works shall not include works that remain
      separable from, or merely link (or bind by name) to the interfaces of,
      the Work and Derivative Works thereof.

      "Contribution" shall mean any work of authorship, including
      the original version of the Work and any modifications or additions
      to that Work or Derivative Works thereof, that is intentionally
      submitted to the Licensor for inclusion in the Work by the copyright owner
      or by an individual or Legal Entity authorized to submit on behalf of
      the copyright owner. For the purposes of this definition, "submitted"
      means any form of electronic, verbal, or written communication sent
      to the Licensor or its representatives, including but not limited to
      communication on electronic mailing lists, source code control systems,
      and issue tracking systems that are managed by, or on behalf of, the
      Licensor for the purpose of discussing and improving the Work, but
      excluding communication that is conspicuously marked or otherwise
      designated in writing by the copyright owner as "Not a Contribution."

      "Contributor" shall mean Licensor and any individual or Legal Entity
      on behalf of whom a Contribution has been received by the Licensor and
      subsequently incorporated within the Work.

   2. Grant of Copyright License. Subject to the terms and conditions of
      this License, each Contributor hereby grants to You a perpetual,
      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
      copyright license to reproduce, prepare Derivative Works of,
      publicly display, publicly perform, sublicense, and distribute the
      Work and such Derivative Works in Source or Object form.

   3. Grant of Patent License. Subject to the terms and conditions of
      this License, each Contributor hereby grants to You a perpetual,
      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
      (except as stated in this section) patent license to make, have made,
      use, offer to sell, sell, import, and otherwise transfer the Work,
      where such license applies only to those patent claims licensable
      by such Contributor that are necessarily infringed by their
      Contribution(s) alone or by combination of their Contribution(s)
      with the Work to which such Contribution(s) was submitted. If You
      institute patent litigation against any entity (including a
      cross-claim or counterclaim in a lawsuit) alleging that the Work
      or a Contribution incorporated within the Work constitutes direct
      or contributory patent infringement, then any patent licenses
      granted to You under this License for that Work shall terminate
      as of the date such litigation is filed.

   4. Redistribution. You may reproduce and distribute copies of the
      Work or Derivative Works thereof in any medium, with or without
      modifications, and in Source or Object form, provided that You
      meet the following conditions:

      (a) You must give any other recipients of the Work or
          Derivative Works a copy of this License; and

      (b) You must cause any modified files to carry prominent notices
          stating that You changed the files; and

      (c) You must retain, in the Source form of any Derivative Works
          that You distribute, all copyright, patent, trademark, and
          attribution notices from the Source form of the Work,
          excluding those notices that do not pertain to any part of
          the Derivative Works; and

      (d) If the Work includes a "NOTICE" text file as part of its
          distribution, then any Derivative Works that You distribute must
          include a readable copy of the attribution notices contained
          within such NOTICE file, excluding any notices that do not
          pertain to any part of the Derivative Works, in at least one
          of the following places: within a NOTICE text file distributed
          as part of the Derivative Works; within the Source form or
          documentation, if provided along with the Derivative Works; or,
          within a display generated by the Derivative Works, if and
          wherever such third-party notices normally appear. The contents
          of the NOTICE file are for informational purposes only and
          do not modify the License. You may add Your own attribution
          notices within Derivative Works that You distribute, alongside
          or as an addendum to the NOTICE text from the Work, provided
          that such additional attribution notices cannot be construed
          as modifying the License.

      You may add Your own copyright statement to Your modifications and
      may provide additional or different license terms and conditions
      for use, reproduction, or distribution of Your modifications, or
      for any such Derivative Works as a whole, provided Your use,
      reproduction, and distribution of the Work otherwise complies with
      the conditions stated in this License.

   5. Submission of Contributions. Unless You explicitly state otherwise,
      any Contribution intentionally submitted for inclusion in the Work
      by You to the Licensor shall be under the terms and conditions of
      this License, without any additional terms or conditions.
      Notwithstanding the above, nothing herein shall supersede or modify
      the terms of any separate license agreement you may have executed
      with Licensor regarding such Contributions.

   6. Trademarks. This License does not grant permission to use the trade
      names, trademarks, service marks, or product names of the Licensor,
      except as required for reasonable and customary use in describing the
      origin of the Work and reproducing the content of the NOTICE file.

   7. Disclaimer of Warranty. Unless required by applicable law or
      agreed to in writing, Licensor provides the Work (and each
      Contributor provides its Contributions) on an "AS IS" BASIS,
      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
      implied, including, without limitation, any warranties or conditions
      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
      PARTICULAR PURPOSE. You are solely responsible for determining the
      appropriateness of using or redistributing the Work and assume any
      risks associated with Your exercise of permissions under this License.

   8. Limitation of Liability. In no event and under no legal theory,
      whether in tort (including negligence), contract, or otherwise,
      unless required by applicable law (such as deliberate and grossly
      negligent acts) or agreed to in writing, shall any Contributor be
      liable to You for damages, including any direct, indirect, special,
      incidental, or consequential damages of any character arising as a
      result of this License or out of the use or inability to use the
      Work (including but not limited to damages for loss of goodwill,
      work stoppage, computer failure or malfunction, or any and all
      other commercial damages or losses), even if such Contributor
      has been advised of the possibility of such damages.

   9. Accepting Warranty or Additional Liability. While redistributing
      the Work or Derivative Works thereof, You may choose to offer,
      and charge a fee for, acceptance of support, warranty, indemnity,
      or other liability obligations and/or rights consistent with this
      License. However, in accepting such obligations, You may act only
      on Your own behalf and on Your sole responsibility, not on behalf
      of any other Contributor, and only if You agree to indemnify,
      defend, and hold each Contributor harmless for any liability
      incurred by, or claims asserted against, such Contributor by reason
      of your accepting any such warranty or additional liability.

   END OF TERMS AND CONDITIONS

   Copyright 2026 jackwener

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "xhs-cli"
version = "0.1.4"
description = "A CLI for Xiaohongshu (小红书) — search, read notes, view profiles"
readme = "README.md"
license = { file = "LICENSE" }
requires-python = ">=3.8"
authors = [{ name = "jackwener", email = "jakevingoo@gmail.com" }]
keywords = ["xiaohongshu", "xhs", "cli", "redbook", "小红书"]
classifiers = [
    "Development Status :: 3 - Alpha",
    "Environment :: Console",
    "Intended Audience :: Developers",
    "License :: OSI Approved :: Apache Software License",
    "Programming Language :: Python :: 3",
    "Programming Language :: Python :: 3.8",
    "Programming Language :: Python :: 3.9",
    "Programming Language :: Python :: 3.10",
    "Programming Language :: Python :: 3.11",
    "Programming Language :: Python :: 3.12",
    "Programming Language :: Python :: 3.13",
    "Topic :: Utilities",
]
dependencies = [
    "camoufox[geoip]>=0.4",
    "playwright>=1.40",
    "browser-cookie3>=0.19",
    "qrcode>=7.4",
    "click>=8.0",
    "rich>=13.0",
]

[project.urls]
Homepage = "https://github.com/jackwener/xhs-cli"
Repository = "https://github.com/jackwener/xhs-cli"
Issues = "https://github.com/jackwener/xhs-cli/issues"

[project.scripts]
xhs = "xhs_cli.cli:cli"

[dependency-groups]
dev = [
    "pytest>=8.3.5",
    "ruff>=0.11.0",
]

[tool.pytest.ini_options]
markers = [
    "integration: requires valid login session (deselect with '-m \"not integration\"')",
    "live_mutation: integration tests that may mutate account state (like/favorite/comment/post)",
]
addopts = "-m 'not integration'"

[tool.ruff]
line-length = 100
target-version = "py38"

[tool.ruff.lint]
select = ["E", "F", "I", "B", "UP"]

#!/usr/bin/env bash
set -euo pipefail

ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$ROOT_DIR"

echo "[smoke] checking local saved session..."
if ! uv run python -m xhs_cli.cli status >/dev/null 2>&1; then
  echo "[smoke] no valid saved session. run 'uv run python -m xhs_cli.cli login' first."
  exit 1
fi

echo "[smoke] validating session usability via whoami..."
if ! uv run python -m xhs_cli.cli whoami >/dev/null 2>&1; then
  echo "[smoke] saved cookies exist but session is expired/invalid."
  echo "[smoke] run 'uv run python -m xhs_cli.cli login' to refresh auth."
  exit 1
fi

echo "[smoke] probing profile endpoint reachability..."
if ! USER_ID="$(uv run python - <<'PY'
import json
import subprocess
import sys

proc = subprocess.run(
    [sys.executable, "-m", "xhs_cli.cli", "whoami", "--json"],
    capture_output=True,
    text=True,
)
if proc.returncode != 0:
    sys.stderr.write(proc.stdout)
    sys.stderr.write(proc.stderr)
    sys.exit(1)

try:
    data = json.loads(proc.stdout)
except json.JSONDecodeError:
    sys.stderr.write(proc.stdout)
    sys.stderr.write(proc.stderr)
    sys.exit(1)
for key in ("userInfo", "basicInfo", "basic_info"):
    sub = data.get(key, {})
    if isinstance(sub, dict):
        uid = sub.get("userId") or sub.get("user_id")
        if uid:
            print(uid)
            sys.exit(0)
uid = data.get("userId") or data.get("user_id") or data.get("id")
if uid:
    print(uid)
    sys.exit(0)
sys.exit(1)
PY
)"; then
  echo "[smoke] failed to resolve user_id from whoami."
  exit 1
fi

if ! uv run python -m xhs_cli.cli user "$USER_ID" --json >/dev/null 2>&1; then
  echo "[smoke] profile endpoint is blocked (likely security verification / risk control)."
  echo "[smoke] refresh login and retry from a normal network."
  exit 1
fi

MARK_EXPR="integration and not live_mutation"
if [[ "${XHS_SMOKE_MUTATION:-0}" == "1" ]]; then
  MARK_EXPR="integration"
fi

echo "[smoke] running integration smoke tests with marker: $MARK_EXPR"
uv run pytest tests/test_integration.py -v --override-ini="addopts=" -m "$MARK_EXPR" "$@"

"""Shared fixtures for xhs-cli tests."""

from __future__ import annotations

import pytest


@pytest.fixture
def tmp_config_dir(tmp_path, monkeypatch):
    """Override auth module's CONFIG_DIR to use a temp directory.

    Also disables browser cookie extraction to avoid subprocess calls.
    """
    import xhs_cli.auth as auth_module

    monkeypatch.setattr(auth_module, "CONFIG_DIR", tmp_path)
    monkeypatch.setattr(auth_module, "COOKIE_FILE", tmp_path / "cookies.json")
    monkeypatch.setattr(auth_module, "TOKEN_CACHE_FILE", tmp_path / "token_cache.json")
    # Prevent actual browser cookie extraction in tests
    monkeypatch.setattr(auth_module, "_extract_browser_cookies", lambda: None)
    return tmp_path


@pytest.fixture
def sample_cookie_str():
    return "a1=abc123; web_session=xyz789; webId=test_id"


@pytest.fixture
def sample_cookie_dict():
    return {"a1": "abc123", "web_session": "xyz789", "webId": "test_id"}

"""Unit tests for xhs_cli.auth module (pure functions, no browser)."""

from __future__ import annotations

import json
import os
import stat

from xhs_cli.auth import (
    _browser_response_payload,
    _dict_to_cookie_str,
    _has_required_cookies,
    _normalize_browser_cookies,
    _render_qr_half_blocks,
    _unwrap_browser_response_payload,
    clear_cookies,
    cookie_str_to_dict,
    get_cookie_string,
    get_saved_cookie_string,
    load_xsec_token,
    save_cookies,
    save_token_cache,
)


class TestCookieStrToDict:
    def test_basic(self):
        result = cookie_str_to_dict("a1=xxx; web_session=yyy")
        assert result == {"a1": "xxx", "web_session": "yyy"}

    def test_empty_string(self):
        assert cookie_str_to_dict("") == {}

    def test_single_cookie(self):
        assert cookie_str_to_dict("a1=abc") == {"a1": "abc"}

    def test_value_with_equals(self):
        result = cookie_str_to_dict("token=abc=def=ghi")
        assert result == {"token": "abc=def=ghi"}

    def test_whitespace_handling(self):
        result = cookie_str_to_dict("  a1 = xxx ;  b = yyy  ")
        assert result == {"a1": "xxx", "b": "yyy"}

    def test_no_equals(self):
        result = cookie_str_to_dict("invalid_cookie")
        assert result == {}

    def test_mixed_valid_invalid(self):
        result = cookie_str_to_dict("a1=xxx; bad; web_session=yyy")
        assert result == {"a1": "xxx", "web_session": "yyy"}


class TestDictToCookieStr:
    def test_basic(self):
        result = _dict_to_cookie_str({"a1": "xxx", "b": "yyy"})
        assert "a1=xxx" in result
        assert "b=yyy" in result

    def test_empty(self):
        assert _dict_to_cookie_str({}) == ""

    def test_roundtrip(self):
        original = {"a1": "abc", "web_session": "def"}
        cookie_str = _dict_to_cookie_str(original)
        parsed = cookie_str_to_dict(cookie_str)
        assert parsed == original


class TestHasRequiredCookies:
    def test_has_a1_and_web_session(self):
        assert _has_required_cookies({"a1": "val", "web_session": "sess", "other": "x"})

    def test_missing_a1(self):
        assert not _has_required_cookies({"web_session": "val"})

    def test_missing_web_session(self):
        assert not _has_required_cookies({"a1": "val"})

    def test_empty(self):
        assert not _has_required_cookies({})


class TestSaveAndLoadCookies:
    def test_save_and_load(self, tmp_config_dir, sample_cookie_str):
        save_cookies(sample_cookie_str)

        # Verify file exists
        cookie_file = tmp_config_dir / "cookies.json"
        assert cookie_file.exists()

        # Verify contents
        data = json.loads(cookie_file.read_text())
        assert "cookies" in data
        assert data["cookies"]["a1"] == "abc123"

    def test_file_permissions(self, tmp_config_dir, sample_cookie_str):
        save_cookies(sample_cookie_str)
        cookie_file = tmp_config_dir / "cookies.json"
        mode = stat.S_IMODE(os.stat(cookie_file).st_mode)
        assert mode == 0o600

    def test_load_roundtrip(self, tmp_config_dir, sample_cookie_str):
        save_cookies(sample_cookie_str)
        loaded = get_cookie_string()
        assert loaded is not None
        parsed = cookie_str_to_dict(loaded)
        assert parsed["a1"] == "abc123"
        assert parsed["web_session"] == "xyz789"

    def test_load_nonexistent(self, tmp_config_dir):
        assert get_cookie_string() is None

    def test_get_saved_cookie_string(self, tmp_config_dir, sample_cookie_str):
        save_cookies(sample_cookie_str)
        loaded = get_saved_cookie_string()
        assert loaded is not None
        parsed = cookie_str_to_dict(loaded)
        assert parsed["a1"] == "abc123"

    def test_save_cookies_handles_chmod_oserror(
        self,
        tmp_config_dir,
        sample_cookie_str,
        monkeypatch,
    ):
        from pathlib import Path

        def _chmod_with_failure(path_obj: Path, mode: int):
            if path_obj.name == "cookies.json":
                raise OSError("chmod not supported")
            return None

        monkeypatch.setattr(Path, "chmod", _chmod_with_failure)
        save_cookies(sample_cookie_str)
        cookie_file = tmp_config_dir / "cookies.json"
        assert cookie_file.exists()


class TestClearCookies:
    def test_clear(self, tmp_config_dir, sample_cookie_str):
        save_cookies(sample_cookie_str)
        save_token_cache({"note1": "token1"})
        removed = clear_cookies()
        assert "cookies.json" in removed
        assert "token_cache.json" in removed

    def test_clear_nothing(self, tmp_config_dir):
        removed = clear_cookies()
        assert removed == []


class TestTokenCache:
    def test_save_and_load(self, tmp_config_dir):
        save_token_cache({"note1": "token_a", "note2": "token_b"})
        assert load_xsec_token("note1") == "token_a"
        assert load_xsec_token("note2") == "token_b"

    def test_load_missing(self, tmp_config_dir):
        save_token_cache({"note1": "token_a"})
        assert load_xsec_token("nonexistent") == ""

    def test_load_no_cache_file(self, tmp_config_dir):
        assert load_xsec_token("anything") == ""

    def test_merge(self, tmp_config_dir):
        save_token_cache({"note1": "token_a"})
        save_token_cache({"note2": "token_b"})

        assert load_xsec_token("note1") == "token_a"
        assert load_xsec_token("note2") == "token_b"

    def test_overwrite(self, tmp_config_dir):
        save_token_cache({"note1": "old"})
        save_token_cache({"note1": "new"})
        assert load_xsec_token("note1") == "new"

    def test_token_cache_file_permissions(self, tmp_config_dir):
        save_token_cache({"note1": "token"})
        token_file = tmp_config_dir / "token_cache.json"
        mode = stat.S_IMODE(os.stat(token_file).st_mode)
        assert mode == 0o600


class TestQrHalfBlockRender:
    def test_empty_matrix(self):
        assert _render_qr_half_blocks([]) == ""

    def test_block_character_mapping(self):
        matrix = [
            [True, False],
            [True, False],
        ]
        rendered = _render_qr_half_blocks(matrix)
        assert "█" in rendered

    def test_half_block_top_and_bottom(self):
        top_only = [
            [True],
            [False],
        ]
        bottom_only = [
            [False],
            [True],
        ]
        assert "▀" in _render_qr_half_blocks(top_only)
        assert "▄" in _render_qr_half_blocks(bottom_only)


class TestBrowserAssistedQrHelpers:
    def test_normalize_browser_cookies_filters_domain_and_name(self):
        raw = [
            {"name": "a1", "value": "cookie-a1", "domain": ".xiaohongshu.com"},
            {"name": "web_session", "value": "cookie-session", "domain": ".xiaohongshu.com"},
            {"name": "ignored", "value": "x", "domain": ".xiaohongshu.com"},
            {"name": "webId", "value": "wrong-domain", "domain": ".example.com"},
        ]
        assert _normalize_browser_cookies(raw) == {
            "a1": "cookie-a1",
            "web_session": "cookie-session",
        }

    def test_unwrap_browser_response_payload_prefers_data_envelope(self):
        payload = {"success": True, "data": {"url": "https://example.com/qr"}}
        assert _unwrap_browser_response_payload(payload) == {"url": "https://example.com/qr"}

    def test_browser_response_payload_rejects_non_json_dict(self):
        class _Response:
            url = "https://www.xiaohongshu.com/api/sns/web/v1/login/qrcode/create"

            def json(self):
                return ["not", "a", "dict"]

        import pytest

        with pytest.raises(Exception, match="unexpected payload"):
            _browser_response_payload(_Response())

class TestQrCodeLogin:
    def test_qrcode_login_delegates_to_browser_assisted_flow(self, monkeypatch):
        monkeypatch.setattr(
            "xhs_cli.auth._browser_assisted_qrcode_login",
            lambda: "a1=browser; web_session=browser-session",
        )

        from xhs_cli.auth import qrcode_login

        assert qrcode_login() == "a1=browser; web_session=browser-session"

"""Unit tests for CLI commands (no browser needed)."""

from __future__ import annotations

from contextlib import contextmanager

import pytest
from click.testing import CliRunner

import xhs_cli.cli as cli_module
from xhs_cli import __version__
from xhs_cli.cli import cli
from xhs_cli.exceptions import DataFetchError


@pytest.fixture
def runner():
    return CliRunner()


class TestCliVersion:
    def test_version_flag(self, runner):
        result = runner.invoke(cli, ["--version"])
        assert result.exit_code == 0
        assert "xhs-cli" in result.output
        assert __version__ in result.output


class TestCliHelp:
    def test_main_help(self, runner):
        result = runner.invoke(cli, ["--help"])
        assert result.exit_code == 0
        # Should list all command groups
        for cmd in ["login", "logout", "status", "whoami", "search", "read",
                     "feed", "topics", "user", "user-posts", "followers",
                     "following", "like", "unlike", "comment", "favorite",
                     "unfavorite", "favorites", "post", "delete"]:
            assert cmd in result.output, f"Command '{cmd}' not found in --help output"

    def test_search_help(self, runner):
        result = runner.invoke(cli, ["search", "--help"])
        assert result.exit_code == 0
        assert "--json" in result.output

    def test_post_help(self, runner):
        result = runner.invoke(cli, ["post", "--help"])
        assert result.exit_code == 0
        assert "--image" in result.output
        assert "--content" in result.output
        assert "--json" in result.output

    def test_favorites_help(self, runner):
        result = runner.invoke(cli, ["favorites", "--help"])
        assert result.exit_code == 0
        assert "--max" in result.output


class TestStatusNotLoggedIn:
    def test_status_no_cookies(self, runner, tmp_config_dir):
        result = runner.invoke(cli, ["status"])
        assert result.exit_code == 1
        assert "Not logged in" in result.output

    def test_status_uses_saved_cookie_only(self, runner, tmp_config_dir, monkeypatch):
        monkeypatch.setattr(cli_module, "get_saved_cookie_string", lambda: "a1=abc")
        monkeypatch.setattr(
            cli_module,
            "get_cookie_string",
            lambda: pytest.fail("status should not trigger browser cookie extraction"),
        )
        result = runner.invoke(cli, ["status"])
        assert result.exit_code == 0
        assert "saved cookies" in result.output


class TestLoginCookieValidation:
    def test_login_valid_cookie(self, runner, tmp_config_dir):
        result = runner.invoke(cli, ["login", "--cookie", "a1=abc; web_session=xyz"])
        assert result.exit_code == 0
        assert "Cookie saved" in result.output

    def test_login_invalid_cookie(self, runner, tmp_config_dir):
        result = runner.invoke(cli, ["login", "--cookie", "bad_cookie_string"])
        assert result.exit_code == 1
        assert "Invalid cookie" in result.output

    def test_login_cookie_missing_web_session(self, runner, tmp_config_dir):
        result = runner.invoke(cli, ["login", "--cookie", "a1=abc_only"])
        assert result.exit_code == 1
        assert "Invalid cookie" in result.output

    def test_login_cookie_name_collision_is_invalid(self, runner, tmp_config_dir):
        result = runner.invoke(cli, ["login", "--cookie", "my_a1=abc; my_web_session=xyz"])
        assert result.exit_code == 1
        assert "Invalid cookie" in result.output

    def test_login_empty_cookie(self, runner, tmp_config_dir, monkeypatch):
        monkeypatch.setattr(
            cli_module,
            "qrcode_login",
            lambda: pytest.fail("qrcode_login should not be called for empty --cookie"),
        )
        result = runner.invoke(cli, ["login", "--cookie", ""])
        assert result.exit_code == 1
        assert "Invalid cookie" in result.output

    def test_login_verify_transient_error_does_not_clear(self, runner, tmp_config_dir, monkeypatch):
        called = {"qrcode": False}
        monkeypatch.setattr(cli_module, "get_cookie_string", lambda: "a1=abc")
        monkeypatch.setattr(cli_module, "_verify_cookies", lambda _cookie_dict: None)
        monkeypatch.setattr(
            cli_module,
            "clear_cookies",
            lambda: pytest.fail("clear_cookies should not be called on transient verify errors"),
        )
        monkeypatch.setattr(
            cli_module,
            "qrcode_login",
            lambda: called.__setitem__("qrcode", True) or "a1=new",
        )

        result = runner.invoke(cli, ["login"])
        assert result.exit_code == 0
        assert "Unable to verify cookies" in result.output
        assert called["qrcode"] is False

    def test_login_verify_invalid_clears_stale_cookies(self, runner, tmp_config_dir, monkeypatch):
        called = {"cleared": False}
        monkeypatch.setattr(cli_module, "get_cookie_string", lambda: "a1=abc")
        monkeypatch.setattr(cli_module, "_verify_cookies", lambda _cookie_dict: False)
        monkeypatch.setattr(cli_module, "_probe_session_usability", lambda _cookie_dict: True)
        monkeypatch.setattr(
            cli_module,
            "clear_cookies",
            lambda: called.__setitem__("cleared", True) or ["cookies.json"],
        )
        monkeypatch.setattr(cli_module, "qrcode_login", lambda: "a1=new; web_session=new")

        result = runner.invoke(cli, ["login"])
        assert result.exit_code == 0
        assert called["cleared"]

    def test_login_verify_ok_but_probe_fails_triggers_refresh(
        self,
        runner,
        tmp_config_dir,
        monkeypatch,
    ):
        called = {"cleared": False, "qrcode": False}
        probe_results = iter([False, True])
        monkeypatch.setattr(
            cli_module,
            "get_cookie_string",
            lambda: "a1=abc; web_session=xyz",
        )
        monkeypatch.setattr(cli_module, "_verify_cookies", lambda _cookie_dict: True)
        monkeypatch.setattr(
            cli_module,
            "_probe_session_usability",
            lambda _cookie_dict: next(probe_results),
        )
        monkeypatch.setattr(
            cli_module,
            "clear_cookies",
            lambda: called.__setitem__("cleared", True) or ["cookies.json"],
        )
        monkeypatch.setattr(
            cli_module,
            "qrcode_login",
            lambda: called.__setitem__("qrcode", True) or "a1=new; web_session=new",
        )

        result = runner.invoke(cli, ["login"])
        assert result.exit_code == 0
        assert called["cleared"] is True
        assert called["qrcode"] is True

    def test_login_qrcode_success_but_probe_fails_exits(self, runner, tmp_config_dir, monkeypatch):
        called = {"cleared": False}
        monkeypatch.setattr(cli_module, "qrcode_login", lambda: "a1=new; web_session=new")
        monkeypatch.setattr(cli_module, "_probe_session_usability", lambda _cookie_dict: False)
        monkeypatch.setattr(
            cli_module,
            "clear_cookies",
            lambda: called.__setitem__("cleared", True) or ["cookies.json"],
        )

        result = runner.invoke(cli, ["login", "--qrcode"])
        assert result.exit_code == 1
        assert called["cleared"] is True
        assert "session is still limited" in result.output

class TestLogout:
    def test_logout_with_cookies(self, runner, tmp_config_dir):
        # First save some cookies
        runner.invoke(cli, ["login", "--cookie", "a1=abc; web_session=xyz"])
        # Then logout
        result = runner.invoke(cli, ["logout"])
        assert result.exit_code == 0
        assert "Logged out" in result.output

    def test_logout_no_cookies(self, runner, tmp_config_dir):
        result = runner.invoke(cli, ["logout"])
        assert result.exit_code == 0
        assert "No saved cookies" in result.output


class _FakeVerifyClient:
    def __init__(self, info, should_raise=False):
        self._info = info
        self._should_raise = should_raise

    def __enter__(self):
        if self._should_raise:
            raise RuntimeError("boom")
        return self

    def __exit__(self, exc_type, exc, tb):
        return False

    def get_self_info(self):
        return self._info


class TestVerifyCookies:
    def test_guest_session_is_invalid(self, monkeypatch):
        monkeypatch.setattr(
            "xhs_cli.client.XhsClient",
            lambda _cookie_dict: _FakeVerifyClient(
                {"userInfo": {"userId": "u123", "guest": True}}
            ),
        )
        assert cli_module._verify_cookies({"a1": "x", "web_session": "y"}) is False

    def test_profile_with_nickname_is_valid(self, monkeypatch):
        monkeypatch.setattr(
            "xhs_cli.client.XhsClient",
            lambda _cookie_dict: _FakeVerifyClient(
                {"userPageData": {"basicInfo": {"nickname": "Alice", "userId": "u1"}}}
            ),
        )
        assert cli_module._verify_cookies({"a1": "x", "web_session": "y"}) is True

    def test_transient_error_returns_none(self, monkeypatch):
        monkeypatch.setattr(
            "xhs_cli.client.XhsClient",
            lambda _cookie_dict: _FakeVerifyClient({}, should_raise=True),
        )
        assert cli_module._verify_cookies({"a1": "x", "web_session": "y"}) is None


class _FakeProbeClient:
    def __init__(self, feed=None, error=None):
        self._feed = feed if feed is not None else []
        self._error = error

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, tb):
        return False

    def get_feed(self):
        if self._error:
            raise self._error
        return self._feed


class TestProbeSessionUsability:
    def test_probe_success(self, monkeypatch):
        monkeypatch.setattr(
            "xhs_cli.client.XhsClient",
            lambda _cookie_dict: _FakeProbeClient(feed=[]),
        )
        assert cli_module._probe_session_usability({"a1": "x", "web_session": "y"}) is True

    def test_probe_data_fetch_error(self, monkeypatch):
        monkeypatch.setattr(
            "xhs_cli.client.XhsClient",
            lambda _cookie_dict: _FakeProbeClient(error=DataFetchError("bad state")),
        )
        assert cli_module._probe_session_usability({"a1": "x", "web_session": "y"}) is False

    def test_probe_transient_error(self, monkeypatch):
        monkeypatch.setattr(
            "xhs_cli.client.XhsClient",
            lambda _cookie_dict: _FakeProbeClient(error=RuntimeError("temporary")),
        )
        assert cli_module._probe_session_usability({"a1": "x", "web_session": "y"}) is None


class _FakeDataClient:
    def search_notes(self, _keyword):
        return [
            "bad-item",
            {
                "id": "n1",
                "xsec_token": "tok1",
                "note_card": {
                    "display_title": "title",
                    "user": {"nickname": "alice"},
                    "interact_info": {"liked_count": 12},
                },
            },
        ]

    def get_followers(self, _user_id):
        return ["bad-user", {"nickname": "bob", "userId": "u1"}]


@contextmanager
def _fake_client_ctx(client):
    yield client


class TestCliRobustness:
    def test_search_skips_non_dict_items(self, runner, monkeypatch):
        monkeypatch.setattr(cli_module, "_get_client", lambda: _fake_client_ctx(_FakeDataClient()))
        monkeypatch.setattr(cli_module, "save_token_cache", lambda *_args, **_kwargs: None)
        result = runner.invoke(cli, ["search", "coffee"])
        assert result.exit_code == 0
        assert "Search: coffee" in result.output

    def test_followers_skips_non_dict_items(self, runner, monkeypatch):
        monkeypatch.setattr(cli_module, "_get_client", lambda: _fake_client_ctx(_FakeDataClient()))
        result = runner.invoke(cli, ["followers", "u123"])
        assert result.exit_code == 0
        assert "Followers" in result.output

"""Unit tests for xhs_cli.client module (no real browser)."""

from __future__ import annotations

import pytest

from xhs_cli.client import XhsClient
from xhs_cli.exceptions import DataFetchError, LoginError


class _FakePage:
    def __init__(self, url: str, evaluate_result):
        self.url = url
        self._evaluate_result = evaluate_result

    def evaluate(self, _script, *_args):
        return self._evaluate_result


class _FakeProfilePage:
    def __init__(self, evaluate_result, url: str = "https://www.xiaohongshu.com/user/profile/u123"):
        self._evaluate_result = evaluate_result
        self.url = url

    def goto(self, *_args, **_kwargs):
        return None

    def evaluate(self, _script, *_args):
        return self._evaluate_result

    def text_content(self, _selector):
        return ""


class _FakeWaitPage:
    def __init__(self, url: str, body: str):
        self.url = url
        self._body = body

    def evaluate(self, _script, *_args):
        return False

    def text_content(self, _selector):
        return self._body


class TestGetNoteComments:
    def test_extracts_note_comments_and_applies_max_limit(self):
        client = XhsClient({})
        client._page = _FakePage(
            "https://www.xiaohongshu.com/explore/note123",
            {"comments": [{"id": "c1"}, {"id": "c2"}]},
        )

        comments = client.get_note_comments("note123", max_comments=1)
        assert comments == [{"id": "c1"}]

    def test_navigates_to_target_note_when_page_mismatch(self, monkeypatch):
        client = XhsClient({})
        client._page = _FakePage(
            "https://www.xiaohongshu.com/explore/other",
            [{"id": "c1"}],
        )
        called = {"value": False}

        def _fake_nav(note_id: str, xsec_token: str):
            called["value"] = True
            assert note_id == "note123"
            assert xsec_token == "tok"
            client._page.url = f"https://www.xiaohongshu.com/explore/{note_id}"

        monkeypatch.setattr(client, "_navigate_to_note", _fake_nav)
        comments = client.get_note_comments("note123", xsec_token="tok", max_comments=10)
        assert called["value"]
        assert comments == [{"id": "c1"}]


class TestPublishResultHeuristic:
    def test_success_indicator_in_page_text(self):
        assert XhsClient._is_publish_success("发布成功", "https://creator.xiaohongshu.com/publish/publish")

    def test_success_when_redirected_away_from_publish_url(self):
        assert XhsClient._is_publish_success(
            "",
            "https://creator.xiaohongshu.com/note/123",
            "123",
        )

    def test_failure_when_no_success_signal_and_still_on_publish_page(self):
        assert not XhsClient._is_publish_success(
            "",
            "https://creator.xiaohongshu.com/publish/publish",
        )

    def test_failure_when_redirected_without_publish_signal_or_note_id(self):
        assert not XhsClient._is_publish_success(
            "",
            "https://creator.xiaohongshu.com/login",
        )

    def test_failure_when_only_generic_success_word_present(self):
        assert not XhsClient._is_publish_success(
            "Operation success",
            "https://creator.xiaohongshu.com/publish/publish",
        )

    def test_extract_note_id_from_explore_url(self):
        note_id = XhsClient._extract_note_id_from_url("https://www.xiaohongshu.com/explore/abc123")
        assert note_id == "abc123"

    def test_extract_note_id_from_query(self):
        note_id = XhsClient._extract_note_id_from_url(
            "https://creator.xiaohongshu.com/publish/success?noteId=xyz987"
        )
        assert note_id == "xyz987"


class TestGetUserInfoFallback:
    def test_returns_unwrapped_user_object_when_key_fields_missing(self, monkeypatch):
        client = XhsClient({})
        client._page = _FakeProfilePage({"nickname": "TestUser", "userId": "u123"})
        monkeypatch.setattr(client, "_human_wait", lambda *_args, **_kwargs: None)
        monkeypatch.setattr(client, "_wait_for_data", lambda *_args, **_kwargs: None)

        info = client.get_user_info("u123")
        assert info["nickname"] == "TestUser"

    def test_returns_minimal_fallback_when_state_missing(self, monkeypatch):
        client = XhsClient({})
        client._page = _FakeProfilePage(None)
        monkeypatch.setattr(client, "_human_wait", lambda *_args, **_kwargs: None)
        monkeypatch.setattr(client, "_wait_for_data", lambda *_args, **_kwargs: None)

        info = client.get_user_info("u123")
        assert info == {"userInfo": {"userId": "u123"}}


class TestWaitForData:
    def test_raises_login_error_when_verification_page_detected(self, monkeypatch):
        client = XhsClient({})
        client._page = _FakeWaitPage(
            "https://www.xiaohongshu.com/website-login/captcha?verifyUuid=abc",
            "Security Verification",
        )
        monkeypatch.setattr("xhs_cli.client.time.sleep", lambda *_args, **_kwargs: None)

        with pytest.raises(LoginError, match="security verification"):
            client._wait_for_data(
                "() => false",
                timeout=0.01,
                desc="user profile",
                raise_on_timeout=True,
            )

    def test_raises_data_fetch_error_when_not_blocked(self, monkeypatch):
        client = XhsClient({})
        client._page = _FakeWaitPage(
            "https://www.xiaohongshu.com/user/profile/u123",
            "normal page body",
        )
        monkeypatch.setattr("xhs_cli.client.time.sleep", lambda *_args, **_kwargs: None)

        with pytest.raises(DataFetchError, match="user profile"):
            client._wait_for_data(
                "() => false",
                timeout=0.01,
                desc="user profile",
                raise_on_timeout=True,
            )

"""Integration smoke tests — require a valid local saved login session.

These tests are marked with @pytest.mark.integration and are deselected by default.
Run explicitly with:

    uv run pytest tests/test_integration.py -v --override-ini="addopts="

By default, mutation tests are marked with @pytest.mark.live_mutation so they
can be excluded safely from day-to-day smoke runs.
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
import time

import pytest
from click.testing import CliRunner

from xhs_cli.cli import cli

pytestmark = pytest.mark.integration


@pytest.fixture(scope="module")
def runner():
    return CliRunner()


def _run_cli(*args: str, timeout: int = 90) -> subprocess.CompletedProcess:
    """Run xhs command via subprocess to avoid event loop conflicts."""
    return subprocess.run(
        [sys.executable, "-m", "xhs_cli.cli", *args],
        capture_output=True,
        text=True,
        timeout=timeout,
    )


def _run_cli_json(*args: str, timeout: int = 90):
    """Run command and parse JSON output."""
    result = _run_cli(*args, timeout=timeout)
    assert result.returncode == 0, (
        f"command failed: {' '.join(args)}\n{result.stdout}{result.stderr}"
    )
    return json.loads(result.stdout)


def _extract_user_id(data: dict) -> str:
    """Extract user_id from whoami JSON output."""
    for sub_key in ["userInfo", "basicInfo", "basic_info"]:
        sub = data.get(sub_key, {})
        if isinstance(sub, dict):
            uid = sub.get("userId", "") or sub.get("user_id", "")
            if uid:
                return str(uid)
    return str(data.get("userId", "") or data.get("user_id", "") or data.get("id", ""))


def _extract_note_from_search_items(items: list[dict]) -> dict[str, str]:
    """Pick one note_id/xsec_token pair from search/feed style items."""
    for item in items:
        if not isinstance(item, dict):
            continue
        note_id = str(item.get("id", "") or item.get("noteId", "") or item.get("note_id", ""))
        xsec_token = str(item.get("xsec_token", "") or item.get("xsecToken", ""))
        if note_id:
            return {"note_id": note_id, "xsec_token": xsec_token}
    return {"note_id": "", "xsec_token": ""}


def _note_cli_args(note: dict[str, str]) -> list[str]:
    args = [note["note_id"]]
    token = note.get("xsec_token", "")
    if token:
        args.extend(["--xsec-token", token])
    return args


def _find_note_id_by_title(user_id: str, title: str) -> str:
    """Find a recently posted note_id from user-posts JSON output by title."""
    posts = _run_cli_json("user-posts", user_id, "--json", timeout=120)
    if not isinstance(posts, list):
        return ""
    for item in posts:
        if not isinstance(item, dict):
            continue
        note_card = item.get("note_card", item.get("noteCard", item))
        if not isinstance(note_card, dict):
            note_card = item
        note_title = str(
            note_card.get("display_title", "")
            or note_card.get("displayTitle", "")
            or note_card.get("title", "")
        )
        if title and title in note_title:
            return str(item.get("id", "") or item.get("noteId", "") or item.get("note_id", ""))
    return ""


@pytest.fixture(scope="module")
def user_id() -> str:
    whoami_data = _run_cli_json("whoami", "--json")
    uid = _extract_user_id(whoami_data)
    if not uid:
        pytest.skip("Cannot extract user_id from whoami")
    return uid


@pytest.fixture(scope="module")
def sample_note() -> dict[str, str]:
    search_items = _run_cli_json("search", "咖啡", "--json", timeout=120)
    if isinstance(search_items, list):
        note = _extract_note_from_search_items(search_items)
        if note["note_id"]:
            return note

    feed_items = _run_cli_json("feed", "--json", timeout=120)
    if isinstance(feed_items, list):
        note = _extract_note_from_search_items(feed_items)
        if note["note_id"]:
            return note

    pytest.skip("No note_id available from search/feed")


# ===== Auth =====


class TestAuth:
    def test_status(self, runner):
        result = runner.invoke(cli, ["status"])
        assert result.exit_code == 0
        assert "Logged in" in result.output

    def test_whoami(self):
        result = _run_cli("whoami")
        assert result.returncode == 0, f"whoami failed: {result.stdout}{result.stderr}"

    def test_whoami_json(self):
        data = _run_cli_json("whoami", "--json")
        assert isinstance(data, dict)


# ===== Search / Read =====


class TestSearchAndRead:
    def test_search(self, runner):
        result = runner.invoke(cli, ["search", "咖啡"])
        assert result.exit_code == 0

    def test_search_json(self):
        data = _run_cli_json("search", "咖啡", "--json", timeout=120)
        assert isinstance(data, list)

    def test_read(self, sample_note):
        result = _run_cli("read", *_note_cli_args(sample_note), timeout=120)
        assert result.returncode == 0, f"read failed: {result.stdout}{result.stderr}"

    def test_read_json_with_comments(self, sample_note):
        args = ["read", *_note_cli_args(sample_note), "--comments", "--json"]
        data = _run_cli_json(*args, timeout=120)
        assert isinstance(data, dict)
        assert "note" in data


# ===== Feed / Topics =====


class TestDiscovery:
    def test_feed(self, runner):
        result = runner.invoke(cli, ["feed"])
        assert result.exit_code == 0

    def test_feed_json(self):
        data = _run_cli_json("feed", "--json", timeout=120)
        assert isinstance(data, list)

    def test_topics(self, runner):
        result = runner.invoke(cli, ["topics", "旅行"])
        assert result.exit_code == 0

    def test_topics_json(self):
        data = _run_cli_json("topics", "旅行", "--json", timeout=120)
        assert isinstance(data, list)


# ===== User =====


class TestUser:
    def test_user(self, user_id):
        result = _run_cli("user", user_id, timeout=120)
        assert result.returncode == 0, f"user failed: {result.stdout}{result.stderr}"

    def test_user_posts(self, user_id):
        result = _run_cli("user-posts", user_id, timeout=120)
        assert result.returncode == 0, f"user-posts failed: {result.stdout}{result.stderr}"

    def test_followers_json(self, user_id):
        data = _run_cli_json("followers", user_id, "--json", timeout=120)
        assert isinstance(data, list)

    def test_following_json(self, user_id):
        data = _run_cli_json("following", user_id, "--json", timeout=120)
        assert isinstance(data, list)


# ===== Favorites =====


class TestFavorites:
    def test_favorites(self, user_id):
        result = _run_cli("favorites", "--max", "3", timeout=120)
        assert result.returncode == 0, f"favorites failed: {result.stdout}{result.stderr}"

    def test_favorites_json(self, user_id):
        data = _run_cli_json("favorites", "--max", "3", "--json", timeout=120)
        assert isinstance(data, list)


# ===== Optional mutation smoke =====


@pytest.mark.live_mutation
class TestMutation:
    def test_like_then_unlike(self, sample_note):
        result = _run_cli("like", *_note_cli_args(sample_note), timeout=120)
        assert result.returncode == 0, f"like failed: {result.stdout}{result.stderr}"

        result = _run_cli("unlike", *_note_cli_args(sample_note), timeout=120)
        assert result.returncode == 0, f"unlike failed: {result.stdout}{result.stderr}"

    def test_favorite_then_unfavorite(self, sample_note):
        result = _run_cli("favorite", *_note_cli_args(sample_note), timeout=120)
        assert result.returncode == 0, f"favorite failed: {result.stdout}{result.stderr}"

        result = _run_cli("unfavorite", *_note_cli_args(sample_note), timeout=120)
        assert result.returncode == 0, f"unfavorite failed: {result.stdout}{result.stderr}"

    def test_comment_optional(self, sample_note):
        text = os.getenv("XHS_SMOKE_COMMENT_TEXT", "").strip()
        if not text:
            pytest.skip("XHS_SMOKE_COMMENT_TEXT not set; skip comment smoke")

        args = ["comment", *_note_cli_args(sample_note), text]
        result = _run_cli(*args, timeout=120)
        assert result.returncode == 0, f"comment failed: {result.stdout}{result.stderr}"

    def test_post_optional(self):
        title_prefix = os.getenv("XHS_SMOKE_POST_TITLE", "Smoke test post").strip()
        title = f"{title_prefix} {int(time.time())}"
        content = os.getenv("XHS_SMOKE_POST_CONTENT", "posted by smoke test").strip()
        images_raw = os.getenv("XHS_SMOKE_POST_IMAGES", "").strip()

        if not images_raw:
            pytest.skip("XHS_SMOKE_POST_IMAGES not set; skip post smoke")

        image_paths = [p.strip() for p in images_raw.split(",") if p.strip()]
        if not image_paths:
            pytest.skip("No usable image path in XHS_SMOKE_POST_IMAGES")

        args = ["post", title]
        for path in image_paths:
            args.extend(["--image", path])
        if content:
            args.extend(["--content", content])
        args.append("--json")

        result = _run_cli(*args, timeout=180)
        combined_output = f"{result.stdout}{result.stderr}"
        if result.returncode != 0 and "Creator platform login required" in combined_output:
            pytest.skip("creator platform login is not available in current local session")
        assert result.returncode == 0, f"post failed: {result.stdout}{result.stderr}"
        payload = json.loads(result.stdout)
        assert payload.get("success") is True

        note_id = str(payload.get("note_id", ""))
        if not note_id:
            user_id = _extract_user_id(_run_cli_json("whoami", "--json", timeout=120))
            if user_id:
                note_id = _find_note_id_by_title(user_id, title)
        assert note_id, "post succeeded but note_id could not be resolved"

        delete_result = _run_cli("delete", note_id, timeout=120)
        assert delete_result.returncode == 0, (
            f"delete failed: {delete_result.stdout}{delete_result.stderr}"
        )

"""xhs-cli: A CLI for Xiaohongshu (小红书)"""

__version__ = "0.1.4"

"""Authentication for Xiaohongshu.

Strategy:
1. Try loading saved cookies from ~/.xhs-cli/cookies.json
2. Try extracting cookies from local Chrome/Firefox via browser-cookie3
3. Fallback: QR code login via API + terminal display
"""

from __future__ import annotations

import json
import logging
from pathlib import Path
from typing import Any

from .exceptions import LoginError

logger = logging.getLogger(__name__)

CONFIG_DIR = Path.home() / ".xhs-cli"
COOKIE_FILE = CONFIG_DIR / "cookies.json"
# Cache file for xsec_token: maps note_id -> xsec_token so users don't
# need to copy-paste tokens manually after search.
TOKEN_CACHE_FILE = CONFIG_DIR / "token_cache.json"

# a1 is required for signing; web_session is required for a stable logged-in session.
REQUIRED_COOKIES = {"a1", "web_session"}
LOGIN_URL = "https://www.xiaohongshu.com/login"
QR_CREATE_ENDPOINT = "/api/sns/web/v1/login/qrcode/create"
QR_USERINFO_ENDPOINT = "/api/qrcode/userinfo"
QR_STATUS_ENDPOINT = "/api/sns/web/v1/login/qrcode/status"
BROWSER_EXPORT_COOKIE_NAMES = (
    "a1",
    "webId",
    "web_session",
    "web_session_sec",
    "id_token",
    "websectiga",
    "sec_poison_id",
    "xsecappid",
    "gid",
    "abRequestId",
    "webBuild",
    "loadts",
)


def get_saved_cookie_string() -> str | None:
    """Load only saved cookies from local config file.

    This helper never triggers browser extraction and has no write side effects.
    """
    return _load_saved_cookies()


def get_cookie_string() -> str | None:
    """Try all auth methods in order. Returns cookie string or None."""
    # 1. Saved cookies
    cookie = _load_saved_cookies()
    if cookie:
        logger.info("Loaded saved cookies from %s", COOKIE_FILE)
        return cookie

    # 2. browser-cookie3
    cookie = _extract_browser_cookies()
    if cookie:
        logger.info("Extracted cookies from local browser")
        save_cookies(cookie)
        return cookie

    return None


def _load_saved_cookies() -> str | None:
    """Load cookies from saved file."""
    if not COOKIE_FILE.exists():
        return None

    try:
        data = json.loads(COOKIE_FILE.read_text())
        cookies = data.get("cookies", {})
        if _has_required_cookies(cookies):
            return _dict_to_cookie_str(cookies)
    except (json.JSONDecodeError, KeyError) as e:
        logger.warning("Failed to load saved cookies: %s", e)

    return None


def _extract_browser_cookies() -> str | None:
    """Extract xiaohongshu cookies from local browsers using browser-cookie3.

    Runs extraction in a subprocess with timeout to avoid hanging
    when the browser is running (Chrome DB lock issue).
    """
    import subprocess
    import sys

    # Python script to run in subprocess
    extract_script = '''
import json, sys
try:
    import browser_cookie3 as bc3
except ImportError:
    print(json.dumps({"error": "not_installed"}))
    sys.exit(0)

browsers = [
    ("Chrome", bc3.chrome),
    ("Firefox", bc3.firefox),
    ("Edge", bc3.edge),
    ("Brave", bc3.brave),
]

for name, loader in browsers:
    try:
        cj = loader(domain_name=".xiaohongshu.com")
        cookies = {c.name: c.value for c in cj if "xiaohongshu" in (c.domain or "")}
        if "a1" in cookies and "web_session" in cookies:
            print(json.dumps({"browser": name, "cookies": cookies}))
            sys.exit(0)
    except Exception:
        pass

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

    try:
        result = subprocess.run(
            [sys.executable, "-c", extract_script],
            capture_output=True, text=True, timeout=15,
        )

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

        data = json.loads(result.stdout.strip())

        if "error" in data:
            if data["error"] == "not_installed":
                logger.warning("browser-cookie3 not installed")
            else:
                logger.debug("No valid cookies found in any browser")
            return None

        cookies = data["cookies"]
        browser = data["browser"]
        logger.info("Found valid cookies in %s (%d cookies)", browser, len(cookies))
        return _dict_to_cookie_str(cookies)

    except subprocess.TimeoutExpired:
        logger.warning("Cookie extraction timed out (browser may be running). "
                       "Try closing your browser or use `xhs login --cookie <string>`")
        return None
    except (json.JSONDecodeError, KeyError) as e:
        logger.warning("Cookie extraction parse error: %s", e)
        return None


def qrcode_login() -> str:
    """Login via QR code displayed to the user."""
    return _browser_assisted_qrcode_login()


def _browser_assisted_qrcode_login() -> str:
    """Login via QR code using network responses instead of page DOM heuristics."""
    import time

    from camoufox.sync_api import Camoufox

    print("🔑 Starting QR code login...")

    with Camoufox(headless=True) as browser:
        page = browser.new_page()
        state = {"last_status": -1}

        def _handle_response(response) -> None:
            if QR_USERINFO_ENDPOINT not in response.url:
                return
            try:
                payload = _browser_response_payload(response)
            except Exception as exc:
                logger.debug("Failed to parse QR poll response: %s", exc)
                return

            code_status = int(payload.get("codeStatus", -1))
            if code_status == state["last_status"]:
                return
            state["last_status"] = code_status

            if code_status == 1:
                print("📲 Scanned! Waiting for confirmation...")
            elif code_status == 2:
                print("✅ Login confirmed!")

        page.on("response", _handle_response)

        try:
            with page.expect_response(
                lambda response: (
                    QR_CREATE_ENDPOINT in response.url
                    and response.request.method == "POST"
                ),
                timeout=20_000,
            ) as qr_response_info:
                page.goto(LOGIN_URL, wait_until="domcontentloaded", timeout=20_000)
        except Exception as exc:
            raise LoginError("Failed to load Xiaohongshu login page in Camoufox.") from exc

        qr_payload = _browser_response_payload(qr_response_info.value)
        qr_url = str(qr_payload.get("url", "")).strip()
        if not qr_url:
            raise LoginError(f"QR login did not expose a QR URL: {qr_payload}")

        print("\n📱 Scan the QR code below with the Xiaohongshu app:\n")
        if not _display_qr_text_in_terminal(qr_url):
            print(f"QR URL: {qr_url}")
        print("\n⏳ Waiting for QR code scan...")

        try:
            with page.expect_response(
                lambda response: (
                    QR_STATUS_ENDPOINT in response.url
                    and response.request.method == "GET"
                ),
                timeout=240_000,
            ) as completion_info:
                pass
        except Exception as exc:
            raise LoginError("QR code login timed out after 4 minutes") from exc

        completion_response = completion_info.value
        _raise_for_browser_response(completion_response)
        completion_data = _browser_response_payload(completion_response)
        _wait_for_browser_login_settled(page)
        time.sleep(1)

        cookies = _normalize_browser_cookies(page.context.cookies())
        login_info = completion_data.get("login_info", {})
        if not isinstance(login_info, dict):
            login_info = {}

        session = login_info.get("session") or completion_data.get("session")
        secure_session = login_info.get("secure_session") or completion_data.get("secure_session")
        if isinstance(session, str) and session:
            cookies["web_session"] = session
        if isinstance(secure_session, str) and secure_session:
            cookies["web_session_sec"] = secure_session

        if not _has_required_cookies(cookies):
            raise LoginError(
                "QR login succeeded, but exported cookies were incomplete: "
                f"keys={', '.join(sorted(cookies.keys()))}"
            )

        cookie_str = _dict_to_cookie_str(cookies)
        save_cookies(cookie_str)
        return cookie_str
def _normalize_browser_cookies(raw_cookies: list[dict[str, Any]]) -> dict[str, str]:
    """Convert browser cookies into the local persisted cookie shape."""
    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 name not in BROWSER_EXPORT_COOKIE_NAMES:
            continue
        if not isinstance(domain, str) or "xiaohongshu.com" not in domain:
            continue
        cookies[name] = value
    return cookies


def _unwrap_browser_response_payload(payload: dict[str, Any]) -> dict[str, Any]:
    """Return the inner data payload when browser responses use a common envelope."""
    data = payload.get("data")
    if isinstance(data, dict):
        return data
    return payload


def _browser_response_payload(response: Any) -> dict[str, Any]:
    """Decode a browser response body as JSON."""
    try:
        data = response.json()
    except Exception as exc:
        raise LoginError(f"Browser response from {response.url} was not valid JSON.") from exc
    if not isinstance(data, dict):
        raise LoginError(
            f"Browser response from {response.url} returned unexpected payload: {data!r}"
        )
    return _unwrap_browser_response_payload(data)


def _raise_for_browser_response(response: Any) -> None:
    """Raise a domain error for QR completion failures."""
    status = getattr(response, "status", None)
    if status in (461, 471):
        verify_type = response.headers.get("verifytype", "unknown")
        verify_uuid = response.headers.get("verifyuuid", "unknown")
        raise LoginError(
            "QR login requires verification. "
            f"verify_type={verify_type} verify_uuid={verify_uuid}"
        )
    if status and status >= 400:
        try:
            body = response.text()
        except Exception:
            body = "<unavailable>"
        raise LoginError(f"QR login failed: HTTP {status} body={body[:300]}")


def _wait_for_browser_login_settled(page: Any) -> None:
    """Wait briefly for the browser session and post-login page state to stabilize."""
    try:
        page.wait_for_url("**/explore*", timeout=5_000)
    except Exception:
        logger.debug("QR login did not navigate to /explore before timeout")

    try:
        response = page.wait_for_response(
            lambda resp: "/api/sns/web/v2/user/me" in resp.url and resp.request.method == "GET",
            timeout=5_000,
        )
    except Exception:
        logger.debug("QR login did not observe a post-login user/me response before timeout")
        return

    try:
        payload = _browser_response_payload(response)
    except Exception as exc:
        logger.debug("Failed to parse browser user/me response after QR login: %s", exc)
        return

    if bool(payload.get("guest", False)):
        logger.debug("QR login settled with guest=true in user/me payload: %s", payload)


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

    border = 2
    width = len(matrix[0]) + border * 2
    padded = [[False] * width for _ in range(border)]
    for row in matrix:
        padded.append(([False] * border) + row + ([False] * border))
    padded.extend([[False] * width for _ in range(border)])

    chars = {
        (False, False): " ",
        (True, False): "▀",
        (False, True): "▄",
        (True, True): "█",
    }

    lines = []
    for y in range(0, len(padded), 2):
        top = padded[y]
        bottom = padded[y + 1] if y + 1 < len(padded) else [False] * width
        line = "".join(chars[(top[x], bottom[x])] for x in range(width))
        lines.append(line)
    return "\n".join(lines)


def _display_qr_text_in_terminal(qr_text: str) -> bool:
    """Render QR text as terminal half-block art."""
    try:
        import qrcode
    except ImportError:
        return False

    try:
        qr = qrcode.QRCode(border=0)
        qr.add_data(qr_text)
        qr.make(fit=True)
        print(_render_qr_half_blocks(qr.get_matrix()))
        return True
    except Exception:
        return False


def save_cookies(cookie_str: str):
    """Save cookies to config file."""
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)

    cookies = cookie_str_to_dict(cookie_str)
    data = {"cookies": cookies}

    COOKIE_FILE.write_text(json.dumps(data, indent=2, ensure_ascii=False))
    try:
        COOKIE_FILE.chmod(0o600)  # Owner-only read/write
    except OSError:
        logger.debug("Failed to set permissions on %s", COOKIE_FILE)
    logger.info("Cookies saved to %s", COOKIE_FILE)


def clear_cookies():
    """Remove saved cookies and token cache (for logout)."""
    removed = []
    for f in (COOKIE_FILE, TOKEN_CACHE_FILE):
        if f.exists():
            f.unlink()
            removed.append(f.name)
    if removed:
        logger.info("Removed: %s", ", ".join(removed))
    return removed


def _has_required_cookies(cookies: dict) -> bool:
    return REQUIRED_COOKIES.issubset(cookies.keys())


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


def cookie_str_to_dict(cookie_str: str) -> dict:
    """Parse a cookie header string into a dict.

    Example: "a1=xxx; web_session=yyy" -> {"a1": "xxx", "web_session": "yyy"}
    """
    result = {}
    for item in cookie_str.split(";"):
        item = item.strip()
        if "=" in item:
            k, v = item.split("=", 1)
            result[k.strip()] = v.strip()
    return result


# ===== xsec_token cache =====
# After a search, we cache the note_id -> xsec_token mapping so that
# subsequent commands (note, like, favorite, comment) can automatically
# resolve the token without requiring the user to pass --xsec-token.


def save_token_cache(token_map: dict[str, str]):
    """Save note_id -> xsec_token mapping from search results.

    Merges with any existing cache so tokens from previous searches
    are preserved until overwritten by a new search containing the
    same note_id.
    """
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)

    # Merge with existing cache
    existing = {}
    if TOKEN_CACHE_FILE.exists():
        try:
            existing = json.loads(TOKEN_CACHE_FILE.read_text())
        except (OSError, json.JSONDecodeError):
            pass

    existing.update(token_map)
    TOKEN_CACHE_FILE.write_text(json.dumps(existing, indent=2, ensure_ascii=False))
    try:
        TOKEN_CACHE_FILE.chmod(0o600)
    except OSError:
        logger.debug("Failed to set permissions on %s", TOKEN_CACHE_FILE)
    logger.info("Cached %d xsec_token(s) to %s", len(token_map), TOKEN_CACHE_FILE)


def load_xsec_token(note_id: str) -> str:
    """Look up cached xsec_token for a given note_id.

    Returns the token string if found, or empty string if not cached.
    """
    if not TOKEN_CACHE_FILE.exists():
        return ""

    try:
        cache = json.loads(TOKEN_CACHE_FILE.read_text())
        token = cache.get(note_id, "")
        if token:
            logger.info("Auto-resolved xsec_token for %s from cache", note_id)
        return token
    except (OSError, json.JSONDecodeError):
        return ""