Bilibili Cli

name: Bug Report
description: Report a bug or unexpected behavior
labels: ["bug"]
body:
  - type: input
    id: version
    attributes:
      label: Version
      description: "Run `bilibili --version` or `pip show bilibili-cli | grep Version`"
      placeholder: "e.g. 1.0.0"
    validations:
      required: true
  - type: dropdown
    id: os
    attributes:
      label: Operating System
      options:
        - macOS
        - Linux
        - Windows
        - Other
    validations:
      required: true
  - type: textarea
    id: description
    attributes:
      label: What happened?
      description: Describe the bug clearly.
    validations:
      required: true
  - type: textarea
    id: expected
    attributes:
      label: Expected behavior
      description: What did you expect to happen?
    validations:
      required: true
  - type: textarea
    id: reproduce
    attributes:
      label: Steps to reproduce
      description: "Commands or steps to reproduce the issue. Use `bilibili -v <command>` for debug output."
      render: bash
  - type: textarea
    id: logs
    attributes:
      label: Error output / logs
      description: Paste any error messages or verbose output here.
      render: text

name: Feature Request
description: Suggest a new feature or improvement
labels: ["enhancement"]
body:
  - type: input
    id: version
    attributes:
      label: Current version
      description: "Run `bilibili --version` or `pip show bilibili-cli | grep Version`"
      placeholder: "e.g. 1.0.0"
    validations:
      required: false
  - type: textarea
    id: description
    attributes:
      label: Describe the feature
      description: What would you like to see added or changed?
    validations:
      required: true
  - type: textarea
    id: use_case
    attributes:
      label: Use case
      description: Why do you need this feature? How would you use it?

name: CI

on:
  workflow_call:
  pull_request:
  push:
    branches:
      - main

jobs:
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"

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

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

      - name: Ruff
        run: uv run ruff check .

  typecheck:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"

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

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

      - name: Mypy
        run: uv run mypy bili_cli

  test:
    needs: [lint, typecheck]
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.10", "3.12"]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}

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

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

      - name: Run tests
        run: uv run pytest tests/ -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: Install uv
        uses: astral-sh/setup-uv@v5

      - name: Build
        run: uv build

      - 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/
ENV/

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

# OS
.DS_Store
Thumbs.db

# Environment
.env
.env.local

# uv

"""bilibili-cli — browse Bilibili from the terminal."""

try:
    from importlib.metadata import version

    __version__ = version("bilibili-cli")
except Exception:
    __version__ = "0.0.0"

"""Authentication for Bilibili.

Strategy:
1. Try loading saved credential from ~/.bilibili-cli/credential.json
2. Try extracting cookies from local browsers via browser-cookie3
3. Fallback: QR code login via bilibili-api-python + terminal display
"""

from __future__ import annotations

import asyncio
import json
import logging
import shutil
import subprocess
import sys
import time
from pathlib import Path
from typing import Literal

import qrcode
from bilibili_api.login_v2 import QrCodeLogin, QrCodeLoginEvents
from bilibili_api.utils.network import Credential

logger = logging.getLogger(__name__)

CONFIG_DIR = Path.home() / ".bilibili-cli"
CREDENTIAL_FILE = CONFIG_DIR / "credential.json"

# Required cookies for a valid Bilibili session
REQUIRED_COOKIES = {"SESSDATA"}

# Extra cookie fields that help bypass Bilibili's 412 anti-scraping checks
EXTRA_COOKIE_FIELDS = ("buvid3", "buvid4", "dedeuserid")

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


AuthMode = Literal["optional", "read", "write"]


def get_credential(mode: AuthMode = "read") -> Credential | None:
    """Try auth methods in order and return credential according to mode.

    - optional: only load saved credential (no network validation, no browser scan)
    - read: prefer validated credential; if validation is indeterminate (network),
      return saved/browser credential as best effort
    - write: same as read, but require bili_jct capability
    """
    require_write = mode == "write"

    # 1. Saved credential file
    cred = _load_saved_credential()
    if cred:
        # Check TTL — try to refresh from browser if stale
        if _is_credential_stale():
            logger.info("Credential older than %d days, attempting browser refresh", CREDENTIAL_TTL_DAYS)
            fresh = _extract_browser_credential()
            if fresh:
                validation = _validate_credential(fresh, require_write=require_write)
                if validation is True:
                    logger.info("Refreshed credential from browser")
                    save_credential(fresh)
                    return fresh
            # Refresh failed — validate existing credential
            logger.warning(
                "Credential is %d+ days old; browser refresh failed. Validating existing credential...",
                CREDENTIAL_TTL_DAYS,
            )

        if mode == "optional":
            return cred
        validation = _validate_credential(cred, require_write=require_write)
        if validation is True:
            logger.info("Loaded valid credential from %s", CREDENTIAL_FILE)
            return cred
        if validation is None:
            logger.warning("Credential validation failed due to network; using saved credential as best effort")
            return cred
        if validation is False:
            logger.warning("Saved credential is expired, clearing")
            clear_credential()

    if mode == "optional":
        return None

    # 2. Browser cookie extraction
    cred = _extract_browser_credential()
    if cred:
        validation = _validate_credential(cred, require_write=require_write)
        if validation is True:
            logger.info("Extracted valid credential from local browser")
            save_credential(cred)
            return cred
        if validation is None:
            logger.warning("Skipping browser credential validation due to network; using best effort")
            return cred
        if validation is False:
            logger.warning("Browser cookies are expired/invalid")

    return None


def _is_credential_stale() -> bool:
    """Check if saved credential file is older than TTL."""
    if not CREDENTIAL_FILE.exists():
        return False
    try:
        data = json.loads(CREDENTIAL_FILE.read_text())
        saved_at = data.get("saved_at", 0)
        if not saved_at:
            # Legacy file without saved_at — treat as stale to add the field
            return True
        return (time.time() - saved_at) > _CREDENTIAL_TTL_SECONDS
    except (json.JSONDecodeError, OSError):
        return False


def _validate_credential(cred: Credential, require_write: bool = False) -> bool | None:
    """Check if a credential is valid.

    Returns:
      - True: credential validated by API
      - False: credential confirmed invalid or missing required fields
      - None: validation is indeterminate due to network/runtime issues
    """
    from bilibili_api import user
    from bilibili_api.exceptions import NetworkException

    if not getattr(cred, "sessdata", ""):
        return False
    if require_write and not getattr(cred, "bili_jct", ""):
        return False

    async def _check():
        try:
            await user.get_self_info(cred)
            return True
        except NetworkException:
            return None
        except Exception:
            return False

    try:
        return asyncio.run(_check())
    except Exception:
        return None


def _load_saved_credential() -> Credential | None:
    """Load credential from saved file."""
    if not CREDENTIAL_FILE.exists():
        return None

    try:
        data = json.loads(CREDENTIAL_FILE.read_text())
        sessdata = data.get("sessdata", "")
        if not sessdata:
            return None
        return Credential(
            sessdata=sessdata,
            bili_jct=data.get("bili_jct", ""),
            ac_time_value=data.get("ac_time_value", ""),
            buvid3=data.get("buvid3", ""),
            buvid4=data.get("buvid4", ""),
            dedeuserid=data.get("dedeuserid", ""),
        )
    except (json.JSONDecodeError, KeyError) as e:
        logger.warning("Failed to load saved credential: %s", e)
        return None


def _extract_browser_credential() -> Credential | None:
    """Extract Bilibili 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).
    """
    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=".bilibili.com")
        cookies = {c.name: c.value for c in cj if "bilibili.com" in (c.domain or "")}
        if "SESSDATA" 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

        output = result.stdout.strip()
        if not output:
            logger.debug("Cookie extraction returned empty output")
            return None

        data = json.loads(output)

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

        cookies = data["cookies"]
        browser_name = data["browser"]
        if not REQUIRED_COOKIES.issubset(cookies):
            logger.debug("Browser cookies missing required keys: %s", REQUIRED_COOKIES)
            return None
        logger.info(
            "Found valid cookies in %s (%d cookies)", browser_name, len(cookies)
        )

        return Credential(
            sessdata=cookies.get("SESSDATA", ""),
            bili_jct=cookies.get("bili_jct", ""),
            ac_time_value=cookies.get("ac_time_value", ""),
            buvid3=cookies.get("buvid3", ""),
            buvid4=cookies.get("buvid4", ""),
            dedeuserid=cookies.get("DedeUserID", ""),
        )

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


def save_credential(credential: Credential):
    """Save credential to config file with timestamp for TTL tracking."""
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)

    data = {
        "sessdata": credential.sessdata,
        "bili_jct": credential.bili_jct,
        "ac_time_value": credential.ac_time_value or "",
        "buvid3": credential.buvid3 or "",
        "buvid4": credential.buvid4 or "",
        "dedeuserid": credential.dedeuserid or "",
        "saved_at": time.time(),
    }
    CREDENTIAL_FILE.write_text(json.dumps(data, indent=2, ensure_ascii=False))
    CREDENTIAL_FILE.chmod(0o600)  # Owner-only read/write
    logger.info("Credential saved to %s", CREDENTIAL_FILE)


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


def _supports_unicode_half_blocks() -> bool:
    """Return True when stdout encoding can represent half-block glyphs."""
    encoding = getattr(sys.stdout, "encoding", None)
    if not encoding:
        return False
    try:
        "▀▄█".encode(encoding)
    except (LookupError, UnicodeEncodeError):
        return False
    return True


def _render_compact_qr(data: str) -> str | None:
    """Render a compact QR code using Unicode half-block characters.

    Uses ▀, ▄, █, and space to encode two vertical modules per character row,
    reducing the QR code height by half compared to full-block rendering.
    Each module is 1 character wide (vs 2 in qrcode-terminal), so total area
    is ~25% of the original.
    """
    qr = qrcode.QRCode(error_correction=qrcode.constants.ERROR_CORRECT_L)
    qr.add_data(data)
    qr.make(fit=True)
    matrix = qr.get_matrix()

    # 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 and warn if too narrow
    term_cols = shutil.get_terminal_size(fallback=(80, 24)).columns
    qr_width = len(matrix[0])
    if qr_width > term_cols:
        logger.warning(
            "Terminal width (%d) too narrow for compact QR (%d), falling back",
            term_cols,
            qr_width,
        )
        return None

    lines: list[str] = []
    # Process two rows at a time using half-block characters
    # top=black, bottom=black → █ (full block)
    # top=black, bottom=white → ▀ (upper half)
    # top=white, bottom=black → ▄ (lower half)
    # top=white, bottom=white → ' ' (space)
    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 _get_qr_terminal_output(login: QrCodeLogin) -> str:
    """Choose compact QR rendering when possible, otherwise use default output."""
    default_qr = login.get_qrcode_terminal()

    qr_link = getattr(login, "_QrCodeLogin__qr_link", None)
    if not qr_link:
        logger.warning("QR link unavailable from QrCodeLogin internals, using default renderer")
        return default_qr

    if not _supports_unicode_half_blocks():
        logger.warning("stdout encoding cannot render Unicode QR blocks, using default renderer")
        return default_qr

    compact_qr = _render_compact_qr(qr_link)
    if compact_qr is None:
        return default_qr
    return compact_qr


async def qr_login() -> Credential:
    """QR code login via terminal.

    Displays a QR code in the terminal, polls until login completes,
    then saves and returns the credential.
    """
    login = QrCodeLogin()
    await login.generate_qrcode()

    # Display QR code in terminal
    print("\n📱 请使用 Bilibili App 扫描以下二维码登录:\n")
    print(_get_qr_terminal_output(login))
    print("\n⭐ 扫码后请在手机上确认登录...")

    # Poll login state
    while True:
        state = await login.check_state()

        if state == QrCodeLoginEvents.DONE:
            credential = login.get_credential()
            save_credential(credential)
            print("\n✅ 登录成功！凭证已保存")
            return credential

        elif state == QrCodeLoginEvents.TIMEOUT:
            raise RuntimeError("二维码已过期，请重试")

        elif state == QrCodeLoginEvents.CONF:
            print("  📲 已扫码，请在手机上确认...")

        await asyncio.sleep(2)

"""CLI entry point for bilibili-cli.

Usage:
    bili login / logout / status / whoami
    bili video <BV号或URL> [--subtitle] [--ai] [--comments] [--related] [--yaml|--json]
    bili user <UID或用户名>          bili user-videos <UID> [--max N]
    bili search <关键词> [--type user|video] [--yaml|--json]
    bili hot / rank / feed / my-dynamics / following / history / watch-later / favorites
    bili dynamic-post <TEXT> / dynamic-delete <动态ID>
    bili like / coin / triple <BV号> / unfollow <UID>
    bili audio <BV号> [--segment N] [--no-split] [-o DIR]
"""

from __future__ import annotations

import click

from . import __version__
from .commands import account, audio, collections, common, discovery, interactions, user_search, video


# Keep helper names for backward compatibility with tests/importers.
def _format_duration(seconds: int) -> str:
    return common.format_duration(seconds)


def _format_count(n: int) -> str:
    return common.format_count(n)


@click.group()
@click.version_option(version=__version__, prog_name="bili")
@click.option("-v", "--verbose", is_flag=True, help="Enable debug logging.")
def cli(verbose: bool):
    """bili — Bilibili CLI tool 📺"""
    common.setup_logging(verbose)


# Register commands.
cli.add_command(account.login)
cli.add_command(account.logout)
cli.add_command(account.status)
cli.add_command(account.whoami)

cli.add_command(video.video)

cli.add_command(user_search.user)
cli.add_command(user_search.user_videos)
cli.add_command(user_search.search)

cli.add_command(collections.favorites)
cli.add_command(collections.following)
cli.add_command(collections.history)
cli.add_command(collections.watch_later)
cli.add_command(collections.feed)
cli.add_command(collections.my_dynamics)
cli.add_command(collections.dynamic_post)
cli.add_command(collections.dynamic_delete)

cli.add_command(discovery.hot_cmd)
cli.add_command(discovery.rank_cmd)

cli.add_command(interactions.like)
cli.add_command(interactions.coin)
cli.add_command(interactions.triple)
cli.add_command(interactions.unfollow)

cli.add_command(audio.audio)


if __name__ == "__main__":
    cli()

"""Bilibili API client — thin async wrappers around bilibili-api-python.

All public functions are async and accept an optional Credential for
authenticated operations (subtitles, favorites, etc.).
"""

from __future__ import annotations

import asyncio
import logging
import os
import re
from typing import Any

import aiohttp
from bilibili_api import comment, dynamic, favorite_list, homepage, hot, rank, search, user, video
from bilibili_api.exceptions import (
    ApiException,
    CredentialNoBiliJctException,
    CredentialNoSessdataException,
    NetworkException,
    ResponseCodeException,
    ResponseException,
)
from bilibili_api.utils.network import Credential

from .exceptions import AuthenticationError, BiliError, InvalidBvidError, NetworkError, NotFoundError, RateLimitError

logger = logging.getLogger(__name__)

_USER_AGENT = (
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
    "AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/133.0.0.0 Safari/537.36"
)


# ---------------------------------------------------------------------------
# BV ID helpers
# ---------------------------------------------------------------------------

_BVID_RE = re.compile(r"\bBV[0-9A-Za-z]{10}\b")


def extract_bvid(url_or_bvid: str) -> str:
    """Extract BV ID from a Bilibili URL or return as-is if already a BV ID."""
    match = _BVID_RE.search(url_or_bvid)
    if match:
        return match.group(0)
    raise InvalidBvidError(f"无法提取 BV 号: {url_or_bvid}")


def _map_api_error(action: str, exc: Exception) -> BiliError:
    """Map third-party API exceptions into stable local exception types."""
    if isinstance(exc, BiliError):
        return exc

    if isinstance(exc, (CredentialNoSessdataException, CredentialNoBiliJctException)):
        return AuthenticationError(f"{action}: {exc}")

    if isinstance(exc, ResponseCodeException):
        code = getattr(exc, "code", None)
        # Auth failures
        if code in {-101, -111}:
            return AuthenticationError(f"{action}: {exc}")
        # Not found
        if code in {-404, 62002, 62004}:
            return NotFoundError(f"{action}: {exc}")
        # Rate limit / anti-scraping
        if code in {-412, 412}:
            return RateLimitError(f"{action}: {exc}")
        return BiliError(f"{action}: [{code}] {exc}")

    if isinstance(exc, (NetworkException, ResponseException, aiohttp.ClientError, asyncio.TimeoutError)):
        return NetworkError(f"{action}: {exc}")

    if isinstance(exc, ApiException):
        return BiliError(f"{action}: {exc}")

    return BiliError(f"{action}: {exc}")


async def _call_api(action: str, awaitable):
    """Run an awaitable and normalize API/network/auth errors."""
    try:
        return await awaitable
    except Exception as exc:
        raise _map_api_error(action, exc) from exc


# ---------------------------------------------------------------------------
# Video
# ---------------------------------------------------------------------------


async def get_video_info(bvid: str, credential: Credential | None = None) -> dict[str, Any]:
    """Fetch video metadata (title, duration, stats, owner, etc.)."""
    v = video.Video(bvid=bvid, credential=credential)
    return await _call_api("获取视频信息", v.get_info())


def format_subtitle_timeline(
    raw: list[dict[str, Any]] | None,
    output_format: str = "timeline",
) -> str:
    """Format subtitle items with timestamps."""
    if not raw:
        return ""

    if output_format == "srt":
        lines: list[str] = []
        for index, item in enumerate(raw, 1):
            lines.append(str(index))
            lines.append(
                f"{_format_subtitle_srt_time(item.get('from', 0.0))} --> "
                f"{_format_subtitle_srt_time(item.get('to', 0.0))}"
            )
            lines.append(item.get("content", ""))
            lines.append("")
        return "\n".join(lines)

    return "\n".join(
        (
            f"[{_format_subtitle_time(item.get('from', 0.0))} --> "
            f"{_format_subtitle_time(item.get('to', 0.0))}] "
            f"{item.get('content', '')}"
        )
        for item in raw
    )


def _format_subtitle_time(seconds: float) -> str:
    """Format seconds as MM:SS.mmm."""
    minutes = int(seconds // 60)
    secs = seconds % 60
    return f"{minutes:02d}:{secs:06.3f}"


def _format_subtitle_srt_time(seconds: float) -> str:
    """Format seconds as HH:MM:SS,mmm for SRT output."""
    hours = int(seconds // 3600)
    minutes = int((seconds % 3600) // 60)
    secs = seconds % 60
    return f"{hours:02d}:{minutes:02d}:{secs:06.3f}".replace(".", ",")


async def get_video_subtitle(
    bvid: str, credential: Credential | None = None
) -> tuple[str, list]:
    """Fetch video subtitle content.

    Returns (plain_text, raw_subtitle_items).
    An empty tuple element means no subtitle available.
    """
    v = video.Video(bvid=bvid, credential=credential)

    # Get cid from first page
    pages = await _call_api("获取视频分P信息", v.get_pages())
    if not pages:
        logger.warning("No pages found for %s", bvid)
        return "", []

    cid = pages[0].get("cid")
    if not cid:
        logger.warning("No cid found for %s", bvid)
        return "", []

    # Get subtitle list from player info
    player_info = await _call_api("获取播放器信息", v.get_player_info(cid=cid))
    subtitle_info = player_info.get("subtitle", {})

    if not subtitle_info or not subtitle_info.get("subtitles"):
        return "", []

    subtitle_list = subtitle_info["subtitles"]

    # Prefer Chinese subtitles
    subtitle_url = None
    for sub in subtitle_list:
        if "zh" in sub.get("lan", "").lower():
            subtitle_url = sub.get("subtitle_url", "")
            break

    if not subtitle_url and subtitle_list:
        subtitle_url = subtitle_list[0].get("subtitle_url", "")

    if not subtitle_url:
        return "", []

    # Ensure absolute URL
    if subtitle_url.startswith("//"):
        subtitle_url = "https:" + subtitle_url

    # Download subtitle JSON
    try:
        timeout = aiohttp.ClientTimeout(total=10)
        async with aiohttp.ClientSession(timeout=timeout) as session:
            async with session.get(subtitle_url) as resp:
                resp.raise_for_status()
                subtitle_data = await resp.json(content_type=None)
    except (aiohttp.ClientError, asyncio.TimeoutError, ValueError) as e:
        raise NetworkError(f"下载字幕失败: {e}") from e

    if "body" in subtitle_data:
        raw = subtitle_data["body"]
        texts = [item.get("content", "") for item in raw]
        return "\n".join(texts), raw

    return "", []


# ---------------------------------------------------------------------------
# User
# ---------------------------------------------------------------------------


async def get_user_info(uid: int, credential: Credential | None = None) -> dict[str, Any]:
    """Fetch user profile information."""
    u = user.User(uid=uid, credential=credential)
    return await _call_api("获取用户信息", u.get_user_info())


async def get_user_relation_info(uid: int, credential: Credential | None = None) -> dict[str, Any]:
    """Fetch user relation stats (follower count, following count)."""
    u = user.User(uid=uid, credential=credential)
    return await _call_api("获取用户关系信息", u.get_relation_info())


async def get_user_videos(
    uid: int, count: int = 10, credential: Credential | None = None
) -> list[dict[str, Any]]:
    """Fetch a user's latest videos.

    Returns list of video dicts (bvid, title, play, length, etc.).
    """
    u = user.User(uid=uid, credential=credential)

    results: list[dict[str, Any]] = []
    page = 1
    per_page = min(count, 50)

    while len(results) < count:
        try:
            data = await _call_api("获取用户视频列表", u.get_videos(ps=per_page, pn=page))
        except BiliError as e:
            if page == 1:
                raise
            logger.warning("Failed to get videos page %d: %s", page, e)
            break

        vlist = data.get("list", {}).get("vlist", [])
        if not vlist:
            break

        for v in vlist:
            results.append(v)
            if len(results) >= count:
                break

        page += 1
        if page > 20:
            break

    return results


# ---------------------------------------------------------------------------
# Search
# ---------------------------------------------------------------------------


async def search_user(keyword: str, page: int = 1) -> list[dict[str, Any]]:
    """Search for users by keyword.

    Returns list of user result dicts.
    """
    res = await _call_api("搜索用户", search.search_by_type(
        keyword=keyword,
        search_type=search.SearchObjectType.USER,
        page=page,
    ))
    return res.get("result", [])


# ---------------------------------------------------------------------------
# Favorites
# ---------------------------------------------------------------------------


async def get_self_info(credential: Credential) -> dict[str, Any]:
    """Get logged-in user's own info."""
    return await _call_api("获取当前登录用户信息", user.get_self_info(credential))


async def get_favorite_list(credential: Credential) -> list[dict[str, Any]]:
    """List all favorite folders for the logged-in user."""
    me = await get_self_info(credential)
    uid = me.get("mid")
    if uid is None:
        raise BiliError("获取收藏夹列表: 当前用户信息缺少 mid")

    fav_data = await _call_api(
        "获取收藏夹列表",
        favorite_list.get_video_favorite_list(uid=uid, credential=credential),
    )
    return fav_data.get("list", [])


async def get_favorite_videos(
    fav_id: int, credential: Credential, page: int = 1
) -> dict[str, Any]:
    """Get videos in a specific favorite folder.

    Returns the raw response dict with 'medias', 'has_more', etc.
    """
    return await _call_api(
        "获取收藏夹内容",
        favorite_list.get_video_favorite_list_content(
            media_id=fav_id, page=page, credential=credential
        ),
    )


# ---------------------------------------------------------------------------
# Hot & Rank
# ---------------------------------------------------------------------------


async def get_hot_videos(pn: int = 1, ps: int = 20) -> dict[str, Any]:
    """Fetch popular/hot videos."""
    return await _call_api("获取热门视频", hot.get_hot_videos(pn=pn, ps=ps))


async def get_rank_videos(day: int = 3) -> dict[str, Any]:
    """Fetch ranking videos (default: 3-day rank)."""
    day_type = rank.RankDayType.THREE_DAY if day == 3 else rank.RankDayType.WEEK
    return await _call_api("获取排行榜", rank.get_rank(day=day_type))


# ---------------------------------------------------------------------------
# Video extras
# ---------------------------------------------------------------------------


async def _get_video_comments_direct(
    aid: int,
    bvid: str,
    page: int,
    credential: Credential | None = None,
) -> dict[str, Any]:
    """Fallback direct API call for video comments when SDK returns empty."""
    api_url = "https://api.bilibili.com/x/v2/reply"
    params = {
        "oid": aid,
        "type": 1,
        "pn": page,
        "ps": 20,
        "sort": 2,  # hot/popular
    }
    headers = {
        "User-Agent": _USER_AGENT,
        "Origin": "https://www.bilibili.com",
        "Referer": f"https://www.bilibili.com/video/{bvid}/",
        "Accept": "application/json, text/plain, */*",
        "Accept-Language": "zh-CN,zh;q=0.9",
        "sec-ch-ua": '"Chromium";v="133", "Not(A:Brand";v="99", "Google Chrome";v="133"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"macOS"',
    }
    if credential and credential.sessdata:
        cookies = [f"SESSDATA={credential.sessdata}"]
        if credential.bili_jct:
            cookies.append(f"bili_jct={credential.bili_jct}")
        headers["Cookie"] = "; ".join(cookies)

    timeout = aiohttp.ClientTimeout(total=30)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get(api_url, params=params, headers=headers) as resp:
            resp.raise_for_status()
            payload = await resp.json()
    if payload.get("code") != 0:
        raise BiliError(
            f"获取视频评论: [{payload.get('code')}] {payload.get('message', 'Unknown error')}"
        )
    data = payload.get("data")
    return data if isinstance(data, dict) else {}


async def get_video_comments(
    bvid: str, page: int = 1, credential: Credential | None = None
) -> dict[str, Any]:
    """Fetch video comments with SDK-first + direct-API fallback strategy."""
    v = video.Video(bvid=bvid, credential=credential)
    info = await _call_api("获取视频信息", v.get_info())
    aid = info.get("aid")
    if aid is None:
        raise BiliError("获取视频评论: 视频信息缺少 aid")

    sdk_result: dict[str, Any] | None = None
    try:
        sdk_result = await _call_api(
            "获取视频评论",
            comment.get_comments(
                oid=aid,
                type_=comment.CommentResourceType.VIDEO,
                page_index=page,
                order=comment.OrderType.LIKE,
                credential=credential,
            ),
        )
    except BiliError as exc:
        logger.warning("SDK comment fetch failed, fallback to direct API: %s", exc)

    if isinstance(sdk_result, dict) and sdk_result.get("replies"):
        return sdk_result

    try:
        direct_result = await _call_api(
            "获取视频评论",
            _get_video_comments_direct(aid=aid, bvid=bvid, page=page, credential=credential),
        )
        if isinstance(direct_result, dict):
            return direct_result
    except BiliError as exc:
        if isinstance(sdk_result, dict) and sdk_result.get("replies"):
            logger.warning("Direct comment fallback failed, return non-empty SDK result: %s", exc)
            return sdk_result
        logger.warning("Direct comment fallback failed after SDK empty/error: %s", exc)
        raise

    if isinstance(sdk_result, dict):
        return sdk_result
    return {}


async def get_video_ai_conclusion(
    bvid: str, credential: Credential | None = None
) -> dict[str, Any]:
    """Fetch AI-generated video summary."""
    v = video.Video(bvid=bvid, credential=credential)
    pages = await _call_api("获取视频分P信息", v.get_pages())
    if not pages:
        return {}
    cid = pages[0].get("cid")
    if not cid:
        return {}
    return await _call_api("获取 AI 总结", v.get_ai_conclusion(cid=cid))


async def get_related_videos(
    bvid: str, credential: Credential | None = None
) -> list[dict[str, Any]]:
    """Fetch related/recommended videos."""
    v = video.Video(bvid=bvid, credential=credential)
    data = await _call_api("获取相关推荐", v.get_related())
    if isinstance(data, list):
        return data
    return []


# ---------------------------------------------------------------------------
# Search (video)
# ---------------------------------------------------------------------------


async def search_video(keyword: str, page: int = 1) -> list[dict[str, Any]]:
    """Search for videos by keyword."""
    res = await _call_api("搜索视频", search.search_by_type(
        keyword=keyword,
        search_type=search.SearchObjectType.VIDEO,
        page=page,
    ))
    return res.get("result", [])


# ---------------------------------------------------------------------------
# Following & Toview
# ---------------------------------------------------------------------------


async def get_followings(
    uid: int, pn: int = 1, ps: int = 20, credential: Credential | None = None
) -> dict[str, Any]:
    """Fetch user's following list."""
    u = user.User(uid=uid, credential=credential)
    return await _call_api("获取关注列表", u.get_followings(pn=pn, ps=ps))


async def modify_user_relation(
    uid: int,
    relation: user.RelationType,
    credential: Credential,
) -> dict[str, Any]:
    """Modify relation to a user (subscribe/unsubscribe/block...)."""
    u = user.User(uid=uid, credential=credential)
    return await _call_api("修改用户关系", u.modify_relation(relation=relation))


async def unfollow_user(uid: int, credential: Credential) -> dict[str, Any]:
    """Unfollow a user by UID."""
    return await modify_user_relation(
        uid=uid,
        relation=user.RelationType.UNSUBSCRIBE,
        credential=credential,
    )


async def get_watch_history(
    page: int = 1, count: int = 30, credential: Credential | None = None
) -> dict[str, Any]:
    """Fetch watch history (观看历史)."""
    if credential is None:
        raise AuthenticationError("credential is required for watch history")
    per_page = max(1, min(count, 100))
    return await _call_api(
        "获取观看历史",
        user.get_self_history(page_num=page, per_page_item=per_page, credential=credential),
    )


async def get_toview(credential: Credential) -> dict[str, Any]:
    """Fetch watch-later (稍后再看) list."""
    data = await _call_api("获取稍后再看列表", homepage.get_favorite_list_and_toview(credential))
    if not isinstance(data, list):
        logger.warning("Unexpected toview payload type: %s", type(data).__name__)
        return {"list": [], "count": 0}
    # data is a list; the item with name="稍后再看" contains toview videos
    for item in data:
        if item.get("name") == "稍后再看" or item.get("id") == 2:
            resp = item.get("mediaListResponse", {})
            return {
                "list": resp.get("list", []),
                "count": resp.get("count", 0),
            }
    return {"list": [], "count": 0}


# ---------------------------------------------------------------------------
# Dynamic Feed
# ---------------------------------------------------------------------------


async def get_dynamic_feed(
    offset: str | int | None = "", credential: Credential | None = None
) -> dict[str, Any]:
    """Fetch dynamic feed (动态时间线)."""
    if credential is None:
        raise AuthenticationError("credential is required for dynamic feed")
    if offset in ("", None):
        parsed_offset = None
    elif isinstance(offset, int):
        parsed_offset = offset
    elif isinstance(offset, str):
        try:
            parsed_offset = int(offset)
        except ValueError as e:
            raise BiliError(f"获取动态时间线: offset 非法: {offset}") from e
    else:
        raise BiliError(f"获取动态时间线: offset 类型不支持: {type(offset).__name__}")

    return await _call_api(
        "获取动态时间线",
        dynamic.get_dynamic_page_info(
            credential=credential,
            pn=1,
            offset=parsed_offset,
        ),
    )


async def post_text_dynamic(text: str, credential: Credential) -> dict[str, Any]:
    """Publish a plain-text dynamic."""
    content = text.strip()
    if not content:
        raise BiliError("发布动态: 文本不能为空")
    info = dynamic.BuildDynamic.empty().add_text(content)
    return await _call_api(
        "发布动态",
        dynamic.send_dynamic(info=info, credential=credential),
    )


async def get_user_dynamics(
    uid: int,
    offset: int = 0,
    need_top: bool = False,
    credential: Credential | None = None,
) -> dict[str, Any]:
    """Fetch a user's own dynamics timeline."""
    u = user.User(uid=uid, credential=credential)
    return await _call_api(
        "获取用户动态",
        u.get_dynamics(offset=offset, need_top=need_top),
    )


async def delete_dynamic(dynamic_id: int, credential: Credential) -> dict[str, Any]:
    """Delete a dynamic by dynamic id."""
    d = dynamic.Dynamic(dynamic_id=dynamic_id, credential=credential)
    return await _call_api("删除动态", d.delete())


# ---------------------------------------------------------------------------
# Interactions (like, coin, triple)
# ---------------------------------------------------------------------------


async def like_video(bvid: str, credential: Credential, undo: bool = False) -> dict[str, Any]:
    """Like or unlike a video."""
    v = video.Video(bvid=bvid, credential=credential)
    return await _call_api("点赞视频", v.like(status=not undo))


async def coin_video(bvid: str, credential: Credential, num: int = 1) -> dict[str, Any]:
    """Give coins to a video (1 or 2)."""
    v = video.Video(bvid=bvid, credential=credential)
    return await _call_api("投币", v.pay_coin(num=num))


async def triple_video(bvid: str, credential: Credential) -> dict[str, Any]:
    """Triple (like + coin + favorite) a video."""
    v = video.Video(bvid=bvid, credential=credential)
    return await _call_api("一键三连", v.triple())


# ---------------------------------------------------------------------------
# Audio extraction
# ---------------------------------------------------------------------------

_DOWNLOAD_HEADERS = {
    "User-Agent": _USER_AGENT,
    "Referer": "https://www.bilibili.com",
}


async def get_audio_url(bvid: str, credential: Credential | None = None) -> str:
    """Get the best audio stream URL for a video (DASH preferred)."""
    from bilibili_api.video import AudioQuality, VideoDownloadURLDataDetecter

    v = video.Video(bvid=bvid, credential=credential)
    download_data = await _call_api("获取下载地址", v.get_download_url(page_index=0))
    detector = VideoDownloadURLDataDetecter(download_data)
    streams = detector.detect_best_streams(
        audio_max_quality=AudioQuality._64K,
        no_dolby_audio=True,
        no_hires=True,
    )

    if detector.check_flv_mp4_stream():
        if streams and streams[0] and hasattr(streams[0], "url"):
            return streams[0].url
    else:
        # DASH: audio is at index 1
        if len(streams) >= 2 and streams[1] is not None and hasattr(streams[1], "url"):
            return streams[1].url
        # Fallback: find any stream with audio_quality
        for s in streams:
            if s is not None and hasattr(s, "audio_quality"):
                return s.url

    raise BiliError("无法获取音频流（可能是会员专属视频）")


async def download_audio(audio_url: str, output_path: str) -> int:
    """Download audio stream to a file. Returns bytes written."""
    timeout = aiohttp.ClientTimeout(total=300)
    max_retries = 3

    for attempt in range(max_retries):
        try:
            async with aiohttp.ClientSession(timeout=timeout) as session:
                async with session.get(audio_url, headers=_DOWNLOAD_HEADERS) as resp:
                    if resp.status == 200:
                        os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
                        total_bytes = 0
                        with open(output_path, "wb") as f:
                            async for chunk in resp.content.iter_chunked(256 * 1024):
                                if not chunk:
                                    continue
                                f.write(chunk)
                                total_bytes += len(chunk)
                        return total_bytes
                    if attempt < max_retries - 1:
                        logger.warning("Download HTTP %d, retrying...", resp.status)
                        await asyncio.sleep(2)
                    else:
                        raise NetworkError(f"音频下载失败: HTTP {resp.status}")
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt < max_retries - 1:
                logger.warning("Download error: %s, retrying...", e)
                await asyncio.sleep(2)
            else:
                raise NetworkError(f"音频下载失败: {e}") from e

    raise NetworkError("音频下载失败: 重试次数用尽")


def split_audio(input_path: str, output_dir: str, segment_seconds: int = 25) -> list[str]:
    """Split audio into WAV segments using PyAV.

    Returns list of segment file paths.
    Each segment is 16kHz mono PCM s16le WAV — ready for ASR APIs.
    """
    try:
        import av as pyav
    except ImportError:
        raise BiliError(
            "音频切分需要 PyAV 库。请安装: pip install av\n"
            "或: uv add av"
        ) from None

    if segment_seconds <= 0:
        raise BiliError("segment_seconds 必须大于 0")

    os.makedirs(output_dir, exist_ok=True)

    def _write_segment(frames: list, seg_idx: int) -> str:
        seg_path = os.path.join(output_dir, f"seg_{seg_idx:03d}.wav")
        out = None
        try:
            out = pyav.open(seg_path, "w", format="wav")
            out_stream = out.add_stream("pcm_s16le", rate=16000, layout="mono")
            resampler = pyav.AudioResampler(format="s16", layout="mono", rate=16000)
            for fr in frames:
                fr.pts = None
                for resampled in resampler.resample(fr):
                    for pkt in out_stream.encode(resampled):
                        out.mux(pkt)
            for pkt in out_stream.encode():
                out.mux(pkt)
        finally:
            if out is not None:
                out.close()
        return seg_path

    input_container = None
    try:
        input_container = pyav.open(input_path)
        if not input_container.streams.audio:
            raise BiliError("音频解码失败: 无音频流")

        chunk_paths = []
        current_samples = 0
        segment_frames: list = []
        seg_idx = 0
        samples_per_segment = None
        decoded_any = False

        for frame in input_container.decode(audio=0):
            decoded_any = True
            if samples_per_segment is None:
                frame_rate = frame.sample_rate or 16000
                samples_per_segment = segment_seconds * frame_rate

            segment_frames.append(frame)
            current_samples += frame.samples or 0

            if samples_per_segment and current_samples >= samples_per_segment:
                chunk_paths.append(_write_segment(segment_frames, seg_idx))
                seg_idx += 1
                current_samples = 0
                segment_frames = []

        if not decoded_any:
            raise BiliError("音频解码失败: 无帧数据")

        if segment_frames:
            chunk_paths.append(_write_segment(segment_frames, seg_idx))

        return chunk_paths
    finally:
        if input_container is not None:
            input_container.close()

"""CLI command modules."""

"""Account and authentication related commands."""

from __future__ import annotations

import sys

import click
from rich.panel import Panel

from .. import payloads
from . import common


@click.command()
def login():
    """扫码登录 Bilibili。"""
    try:
        common.run(common.qr_login())
    except RuntimeError as e:
        common.exit_error(str(e))
    except Exception as e:
        common.exit_error(f"登录失败: {e}")


@click.command()
def logout():
    """注销并清除保存的凭证。"""
    common.clear_credential()
    common.console.print("[green]✅ 已注销，凭证已清除[/green]")


@click.command()
@common.structured_output_options
def status(as_json: bool, as_yaml: bool):
    """检查登录状态。"""
    output_format = common.resolve_output_format(as_json=as_json, as_yaml=as_yaml)
    cred = common.get_credential(mode="read")
    if not cred:
        payload = common.error_payload("not_authenticated", "未登录。使用 bili login 登录。")
        if common.emit_structured(payload, output_format):
            raise SystemExit(1) from None
        common.print_login_required("未登录。使用 [bold]bili login[/bold] 登录。")
        sys.exit(1)

    from .. import client

    try:
        info = common.run(client.get_self_info(cred))
    except Exception as exc:
        payload = common.error_payload("api_error", f"检查登录状态失败: {exc}")
        if common.emit_structured(payload, output_format):
            raise SystemExit(1) from None
        common.exit_error(f"检查登录状态失败: {exc}")

    payload = common.success_payload(
        {
            "authenticated": True,
            "user": payloads.normalize_user(info),
        }
    )
    def render() -> None:
        name = info.get("name", "unknown")
        uid = info.get("mid", "unknown")
        common.console.print(f"[green]✅ 已登录：[bold]{name}[/bold]  (UID: {uid})[/green]")

    if common.emit_or_print(payload, output_format, render):
        return


@click.command()
@common.structured_output_options
def whoami(as_json: bool, as_yaml: bool):
    """查看当前登录用户的详细信息。"""
    from .. import client

    output_format = common.resolve_output_format(as_json=as_json, as_yaml=as_yaml)

    cred = common.get_credential(mode="read")
    if not cred:
        payload = common.error_payload("not_authenticated", "未登录。使用 bili login 登录。")
        if common.emit_structured(payload, output_format):
            raise SystemExit(1) from None
        common.print_login_required("未登录。使用 [bold]bili login[/bold] 登录。")
        sys.exit(1)

    try:
        info = common.run(client.get_self_info(cred))
        uid = info.get("mid", "unknown")
        relation = common.run(client.get_user_relation_info(uid, credential=cred))
    except Exception as exc:
        payload = common.error_payload("api_error", f"获取用户信息失败: {exc}")
        if common.emit_structured(payload, output_format):
            raise SystemExit(1) from None
        common.exit_error(f"获取用户信息失败: {exc}")

    payload = common.success_payload(
        {"user": payloads.normalize_user(info), "relation": payloads.normalize_relation(relation)}
    )

    def render() -> None:
        name = info.get("name", "unknown")
        level = info.get("level", "?")
        coins = info.get("coins", 0)
        follower = relation.get("follower", 0)
        following = relation.get("following", 0)

        vip = info.get("vip", {})
        vip_label = ""
        if vip.get("status") == 1:
            vip_type = "大会员" if vip.get("type") == 2 else "小会员"
            vip_label = f"  |  🏅 {vip_type}"

        sign = info.get("sign", "").strip()

        lines = [
            f"👤 [bold]{name}[/bold]  (UID: {uid})",
            f"⭐ Level {level}  |  🪙 硬币 {coins}{vip_label}",
            f"👥 粉丝 {common.format_count(follower)}  |  🔔 关注 {common.format_count(following)}",
        ]
        if sign:
            lines.append(f"📝 {sign}")

        common.console.print(Panel(
            "\n".join(lines),
            title="个人信息",
            border_style="green",
        ))

    if common.emit_or_print(payload, output_format, render):
        return

"""Audio extraction command — download and split video audio for ASR."""

from __future__ import annotations

import os
import re
import tempfile

import click

from .common import console, exit_error, extract_bvid_or_exit, get_credential, run_or_exit

DEFAULT_TMP_DIR = os.path.join(tempfile.gettempdir(), "bilibili-cli")


def _sanitize_filename(title: str) -> str:
    """Remove or replace characters that are unsafe in file paths."""
    title = re.sub(r'[<>:"/\\|?*]', "_", title)
    title = title.strip(". ")
    return title[:120] or "audio"


@click.command()
@click.argument("bv_or_url")
@click.option("--segment", "-s", default=25, type=click.IntRange(5, 300),
              help="每段时长（秒），默认 25。")
@click.option("--no-split", is_flag=True, help="不切分，直接保存完整音频文件。")
@click.option("--output", "-o", default=None, type=click.Path(),
              help=f"输出目录（默认 {DEFAULT_TMP_DIR}/{{title}}/）。")
def audio(bv_or_url: str, segment: int, no_split: bool, output: str | None):
    """下载视频音频并切分为 ASR-ready WAV 片段。

    默认输出到 /tmp/bilibili-cli/{title}/ 目录，
    每段 25 秒，16kHz mono PCM WAV 格式，可直接用于语音转文字 API。

    \b
    示例:
      bili audio BV1ABcsztEcY                  # 下载并切分
      bili audio BV1ABcsztEcY --segment 60     # 每段 60 秒
      bili audio BV1ABcsztEcY --no-split       # 保存完整 m4a
      bili audio BV1ABcsztEcY -o ~/data/       # 自定义输出目录
    """
    from .. import client

    bvid = extract_bvid_or_exit(bv_or_url)

    # 1. Get video info for title
    cred = get_credential(mode="optional")
    info = run_or_exit(client.get_video_info(bvid, credential=cred), "获取视频信息")
    title = info.get("title", bvid)
    duration = info.get("duration", 0)
    safe_title = _sanitize_filename(title)

    console.print(f"[bold]🎵 {title}[/bold]  ({_format_time(duration)})")

    # 2. Get audio stream URL
    console.print("[dim]获取音频流地址...[/dim]")
    audio_url = run_or_exit(client.get_audio_url(bvid, credential=cred), "获取音频流")

    # 3. Determine output directory
    if output:
        out_dir = os.path.expanduser(output)
    else:
        out_dir = os.path.join(DEFAULT_TMP_DIR, safe_title)

    if no_split:
        # Download full audio without splitting
        out_file = os.path.join(out_dir, f"{safe_title}.m4a")
        console.print("[dim]下载音频中...[/dim]")
        nbytes = run_or_exit(client.download_audio(audio_url, out_file), "下载音频")
        size_mb = nbytes / (1024 * 1024)
        console.print(f"[green]✅ 音频已保存: {out_file} ({size_mb:.1f} MB)[/green]")
    else:
        # Download to temp file, then split
        os.makedirs(out_dir, exist_ok=True)
        tmp_path = os.path.join(out_dir, "_raw.m4s")

        console.print("[dim]下载音频中...[/dim]")
        nbytes = run_or_exit(client.download_audio(audio_url, tmp_path), "下载音频")
        size_mb = nbytes / (1024 * 1024)
        console.print(f"[dim]下载完成 ({size_mb:.1f} MB)，切分中...[/dim]")

        try:
            segments = client.split_audio(tmp_path, out_dir, segment_seconds=segment)
        except Exception as e:
            exit_error(f"音频切分失败: {e}")
        finally:
            # Clean up raw download
            if os.path.exists(tmp_path):
                os.unlink(tmp_path)

        console.print(f"[green]✅ 切分完成: {len(segments)} 段 (每段 ~{segment}s)[/green]")
        console.print(f"[green]   输出目录: {out_dir}[/green]")
        for _i, seg in enumerate(segments):
            basename = os.path.basename(seg)
            size_kb = os.path.getsize(seg) / 1024
            console.print(f"[dim]   {basename}  ({size_kb:.0f} KB)[/dim]")


def _format_time(seconds: int) -> str:
    """Format duration for display."""
    if seconds >= 3600:
        h, rem = divmod(seconds, 3600)
        m, s = divmod(rem, 60)
        return f"{h}:{m:02d}:{s:02d}"
    m, s = divmod(seconds, 60)
    return f"{m:02d}:{s:02d}"

"""Collection and timeline related commands."""

from __future__ import annotations

import json
from datetime import datetime
from pathlib import Path
from typing import Any

import click
from rich.table import Table

from .. import payloads
from . import common


def _decode_json(value: object) -> dict[str, Any]:
    if isinstance(value, dict):
        return value
    if not isinstance(value, str):
        return {}
    try:
        parsed = json.loads(value)
    except json.JSONDecodeError:
        return {}
    return parsed if isinstance(parsed, dict) else {}


def _extract_dynamic_id(item: dict[str, Any]) -> int:
    desc = item.get("desc", {}) if isinstance(item, dict) else {}
    candidates = [
        desc.get("dynamic_id"),
        desc.get("dynamic_id_str"),
        item.get("id_str"),
        item.get("id"),
    ]
    for candidate in candidates:
        if isinstance(candidate, int):
            return candidate
        if isinstance(candidate, str):
            try:
                return int(candidate)
            except ValueError:
                continue
    return 0


def _extract_dynamic_timestamp(item: dict[str, Any]) -> int:
    desc = item.get("desc", {}) if isinstance(item, dict) else {}
    ts = desc.get("timestamp")
    if isinstance(ts, int):
        return ts
    if isinstance(ts, str):
        try:
            return int(ts)
        except ValueError:
            return 0
    return 0


def _extract_dynamic_text(item: dict[str, Any]) -> str:
    parts: list[str] = []

    modules = item.get("modules", {}) if isinstance(item, dict) else {}
    if isinstance(modules, dict):
        dynamic_mod = modules.get("module_dynamic", {})
        if isinstance(dynamic_mod, dict):
            desc = dynamic_mod.get("desc", {})
            if isinstance(desc, dict) and isinstance(desc.get("text"), str):
                parts.append(desc["text"])

    card = _decode_json(item.get("card"))
    for key in ("title", "description", "dynamic", "summary"):
        value = card.get(key)
        if isinstance(value, str) and value.strip():
            parts.append(value.strip())
    card_item = card.get("item")
    if isinstance(card_item, dict):
        for key in ("title", "description", "content"):
            value = card_item.get(key)
            if isinstance(value, str) and value.strip():
                parts.append(value.strip())

    if not parts:
        desc = item.get("desc", {}) if isinstance(item, dict) else {}
        if isinstance(desc, dict):
            for key in ("description", "dynamic_id_str"):
                value = desc.get(key)
                if isinstance(value, str) and value.strip():
                    parts.append(value.strip())

    return " ".join(parts).strip()


@click.command()
@click.argument("fav_id", required=False, type=int)
@click.option("--page", "-p", default=1, type=click.IntRange(1), help="页码 (默认 1，最小 1)。")
@common.structured_output_options
def favorites(fav_id: int | None, page: int, as_json: bool, as_yaml: bool):
    """浏览收藏夹。

    不带参数列出所有收藏夹，带 FAV_ID 查看收藏夹内的视频。
    """
    from .. import client

    output_format = common.resolve_output_format(as_json=as_json, as_yaml=as_yaml)

    cred = common.require_login(message="需要登录才能查看收藏夹。使用 [bold]bili login[/bold] 登录。")

    if fav_id is None:
        fav_list = common.run_or_exit(client.get_favorite_list(cred), "获取收藏夹列表失败")

        if common.emit_structured([payloads.normalize_favorite_folder(item) for item in fav_list], output_format):
            return

        if not fav_list:
            common.console.print("[yellow]未找到收藏夹[/yellow]")
            return

        table = Table(title="📂 收藏夹列表", border_style="blue")
        table.add_column("ID", style="cyan", width=12)
        table.add_column("名称", width=20)
        table.add_column("视频数", width=10, justify="right")

        for f in fav_list:
            table.add_row(
                str(f.get("id", "")),
                f.get("title", ""),
                str(f.get("media_count", 0)),
            )

        common.console.print(table)
        common.console.print("\n[dim]使用 [bold]bili favorites <ID>[/bold] 查看收藏夹内容[/dim]")

    else:
        data = common.run_or_exit(
            client.get_favorite_videos(fav_id, cred, page=page),
            "获取收藏夹内容失败",
        )

        if common.emit_structured(
            {
                "folder_id": fav_id,
                "page": page,
                "has_more": bool(data.get("has_more", False)),
                "items": [payloads.normalize_favorite_media(item) for item in (data.get("medias") or [])],
            },
            output_format,
        ):
            return

        medias = data.get("medias") or []
        if not medias:
            common.console.print("[yellow]收藏夹为空或不存在[/yellow]")
            return

        table = Table(title=f"📂 收藏夹 #{fav_id}  (第 {page} 页)", border_style="blue")
        table.add_column("#", style="dim", width=4)
        table.add_column("BV号", style="cyan", width=14)
        table.add_column("标题", max_width=40)
        table.add_column("UP主", width=12)
        table.add_column("时长", width=8)

        for i, m in enumerate(medias, 1 + (page - 1) * 20):
            upper = m.get("upper", {})
            table.add_row(
                str(i),
                m.get("bvid", ""),
                (m.get("title", "") or "")[:40],
                (upper.get("name", "") or "")[:12],
                common.format_duration(m.get("duration", 0)),
            )

        common.console.print(table)

        has_more = data.get("has_more", False)
        if has_more:
            common.console.print(f"\n[dim]还有更多内容，使用 [bold]bili favorites {fav_id} --page {page + 1}[/bold] 查看下一页[/dim]")


@click.command()
@click.option("--page", "-p", default=1, type=click.IntRange(1), help="页码 (默认 1，最小 1)。")
@common.structured_output_options
def following(page: int, as_json: bool, as_yaml: bool):
    """查看关注列表。"""
    from .. import client

    output_format = common.resolve_output_format(as_json=as_json, as_yaml=as_yaml)

    cred = common.require_login()

    me = common.run_or_exit(client.get_self_info(cred), "获取关注列表失败")
    uid = me["mid"]
    data = common.run_or_exit(
        client.get_followings(uid, pn=page, credential=cred),
        "获取关注列表失败",
    )

    if common.emit_structured(
        {
            "page": page,
            "total": data.get("total", 0),
            "items": [payloads.normalize_following_user(item) for item in (data.get("list") or [])],
        },
        output_format,
    ):
        return

    flist = data.get("list") or []
    if not flist:
        common.console.print("[yellow]关注列表为空[/yellow]")
        return

    total = data.get("total", "?")
    table = Table(title=f"🔔 关注列表  (共 {total}, 第 {page} 页)", border_style="blue")
    table.add_column("#", style="dim", width=4)
    table.add_column("UID", style="cyan", width=12)
    table.add_column("用户名", width=16)
    table.add_column("签名", max_width=40)

    for i, u in enumerate(flist, 1 + (page - 1) * 20):
        table.add_row(
            str(i),
            str(u.get("mid", "")),
            u.get("uname", ""),
            (u.get("sign", "") or "")[:40],
        )

    common.console.print(table)
    common.console.print(f"\n[dim]使用 [bold]bili following --page {page + 1}[/bold] 查看下一页[/dim]")


@click.command()
@click.option("--page", "-p", default=1, type=click.IntRange(1), help="页码 (默认 1，最小 1)。")
@click.option("--max", "-n", "count", default=30, type=click.IntRange(1, 100), help="显示数量 (默认 30，1-100)。")
@common.structured_output_options
def history(page: int, count: int, as_json: bool, as_yaml: bool):
    """查看观看历史。"""
    from .. import client

    output_format = common.resolve_output_format(as_json=as_json, as_yaml=as_yaml)

    cred = common.require_login()
    data = common.run_or_exit(
        client.get_watch_history(page=page, count=count, credential=cred),
        "获取观看历史失败",
    )

    if isinstance(data, list):
        vlist = data
    else:
        vlist = data.get("list") or data.get("items") or data.get("data") or []

    history_items = vlist if isinstance(vlist, list) else []
    if common.emit_structured(
        {
            "page": page,
            "count": min(count, len(history_items)) if history_items else 0,
            "items": [payloads.normalize_history_item(item) for item in history_items[:count] if isinstance(item, dict)],
        },
        output_format,
    ):
        return

    if not isinstance(vlist, list) or not vlist:
        common.console.print("[yellow]观看历史为空[/yellow]")
        return

    table = Table(title=f"🕘 观看历史  (第 {page} 页)", border_style="blue")
    table.add_column("#", style="dim", width=4)
    table.add_column("标识", style="cyan", width=14)
    table.add_column("标题", max_width=36)
    table.add_column("UP主", width=12)
    table.add_column("观看时间", width=12)

    for i, v in enumerate(vlist[:count], 1):
        history_info = v.get("history", {}) if isinstance(v, dict) else {}
        owner = v.get("owner", {}) if isinstance(v, dict) else {}
        view_at = history_info.get("view_at") or (v.get("view_at", 0) if isinstance(v, dict) else 0)
        if isinstance(view_at, int) and view_at > 0:
            view_time = datetime.fromtimestamp(view_at).strftime("%m-%d %H:%M")
        else:
            view_time = "-"
        table.add_row(
            str(i),
            history_info.get("bvid") or v.get("bvid", "") or str(history_info.get("oid", "")),
            (v.get("title", "") or v.get("name", ""))[:36],
            (owner.get("name", "") or v.get("author_name", "") or v.get("author", ""))[:12],
            view_time,
        )

    common.console.print(table)


@click.command(name="watch-later")
@common.structured_output_options
def watch_later(as_json: bool, as_yaml: bool):
    """查看稍后再看列表。"""
    from .. import client

    output_format = common.resolve_output_format(as_json=as_json, as_yaml=as_yaml)

    cred = common.require_login()

    data = common.run_or_exit(client.get_toview(cred), "获取稍后再看失败")

    if common.emit_structured(
        {
            "count": data.get("count", 0),
            "items": [payloads.normalize_watch_later_item(item) for item in (data.get("list") or [])],
        },
        output_format,
    ):
        return

    vlist = data.get("list") or []
    if not vlist:
        common.console.print("[yellow]稍后再看列表为空[/yellow]")
        return

    total = data.get("count", len(vlist))
    table = Table(title=f"⏰ 稍后再看  (共 {total} 个)", border_style="blue")
    table.add_column("#", style="dim", width=4)
    table.add_column("BV号", style="cyan", width=14)
    table.add_column("标题", max_width=36)
    table.add_column("UP主", width=12)
    table.add_column("时长", width=8)

    for i, v in enumerate(vlist[:30], 1):
        owner = v.get("owner", {})
        table.add_row(
            str(i),
            v.get("bvid", ""),
            v.get("title", "")[:36],
            owner.get("name", "")[:12],
            common.format_duration(v.get("duration", 0)),
        )

    common.console.print(table)


@click.command()
@click.option("--offset", default="", help="分页游标；留空为最新。可使用上一页返回的 next_offset/offset。")
@common.structured_output_options
def feed(offset: str, as_json: bool, as_yaml: bool):
    """查看动态时间线。"""
    from .. import client

    output_format = common.resolve_output_format(as_json=as_json, as_yaml=as_yaml)

    cred = common.require_login()

    data = common.run_or_exit(
        client.get_dynamic_feed(offset=offset, credential=cred),
        "获取动态失败",
    )

    items = data.get("items") or []
    if common.emit_structured(
        {
            "items": [payloads.normalize_dynamic_item(item) for item in items if isinstance(item, dict)],
            "next_offset": data.get("next_offset") or data.get("offset") or "",
        },
        output_format,
    ):
        return

    if not items:
        common.console.print("[yellow]暂无动态[/yellow]")
        return

    common.console.print("[bold]📰 动态时间线[/bold]\n")

    for item in items[:15]:
        modules = item.get("modules", {})
        author = modules.get("module_author", {})
        dyn_main = modules.get("module_dynamic", {})
        stat = modules.get("module_stat", {})

        name = author.get("name", "")
        pub_time = author.get("pub_time", "")

        desc = dyn_main.get("desc", {})
        text = desc.get("text", "") if desc else ""

        major = dyn_main.get("major", {})
        title = ""
        if major:
            archive = major.get("archive", {})
            if archive:
                title = archive.get("title", "")
            article = major.get("article", {})
            if article:
                title = article.get("title", "")

        comment_info = stat.get("comment", {})
        like_info = stat.get("like", {})
        comment_count = comment_info.get("count", 0) if comment_info else 0
        like_count = like_info.get("count", 0) if like_info else 0

        common.console.print(f"  [cyan]{name}[/cyan]  [dim]{pub_time}[/dim]")
        if title:
            common.console.print(f"  📺 {title}")
        if text:
            common.console.print(f"  {text[:100]}")
        if comment_count or like_count:
            common.console.print(f"  [dim]👍 {like_count}  💬 {comment_count}[/dim]")
        common.console.print()

    next_offset = data.get("next_offset") or data.get("offset")
    if next_offset not in ("", None):
        common.console.print(f"[dim]下一页：bili feed --offset {next_offset}[/dim]")


@click.command(name="my-dynamics")
@click.option("--offset", default=0, type=click.IntRange(0), help="分页偏移量；默认 0。")
@click.option("--top/--no-top", "need_top", default=False, help="是否包含置顶动态。")
@click.option("--max", "-n", "count", default=20, type=click.IntRange(1, 50), help="显示条数 (1-50)。")
@common.structured_output_options
def my_dynamics(offset: int, need_top: bool, count: int, as_json: bool, as_yaml: bool):
    """查看我发布的动态。"""
    from .. import client

    output_format = common.resolve_output_format(as_json=as_json, as_yaml=as_yaml)

    cred = common.require_login()

    me = common.run_or_exit(client.get_self_info(cred), "获取我的动态失败")
    uid = me.get("mid")
    if not isinstance(uid, int):
        common.exit_error("获取我的动态失败: 当前用户信息缺少 mid")

    data = common.run_or_exit(
        client.get_user_dynamics(uid=uid, offset=offset, need_top=need_top, credential=cred),
        "获取我的动态失败",
    )

    cards = data.get("cards") or []
    if common.emit_structured(
        {
            "offset": offset,
            "next_offset": data.get("next_offset") or data.get("offset") or "",
            "items": [payloads.normalize_dynamic_item(card) for card in cards[:count] if isinstance(card, dict)],
        },
        output_format,
    ):
        return

    if not isinstance(cards, list) or not cards:
        common.console.print("[yellow]暂无我发布的动态[/yellow]")
        return

    table = Table(title=f"📝 我的动态  (offset={offset})", border_style="blue")
    table.add_column("#", style="dim", width=4)
    table.add_column("动态ID", style="cyan", width=16)
    table.add_column("发布时间", width=12)
    table.add_column("内容", max_width=60)

    for idx, card in enumerate(cards[:count], 1):
        if not isinstance(card, dict):
            continue
        dynamic_id = _extract_dynamic_id(card)
        ts = _extract_dynamic_timestamp(card)
        pub_time = datetime.fromtimestamp(ts).strftime("%m-%d %H:%M") if ts > 0 else "-"
        text = _extract_dynamic_text(card)[:60] or "-"
        table.add_row(str(idx), str(dynamic_id), pub_time, text)

    common.console.print(table)

    next_offset = data.get("next_offset") or data.get("offset")
    if next_offset not in ("", None, offset):
        common.console.print(f"\n[dim]下一页：bili my-dynamics --offset {next_offset}[/dim]")


@click.command(name="dynamic-post")
@click.argument("text", required=False)
@click.option(
    "--from-file",
    "from_file",
    type=click.Path(exists=True, dir_okay=False, path_type=Path),
    default=None,
    help="从文件读取动态文本。",
)
@common.structured_output_options
def dynamic_post(text: str | None, from_file: Path | None, as_json: bool, as_yaml: bool):
    """发布一条纯文本动态。"""
    from .. import client

    output_format = common.resolve_output_format(as_json=as_json, as_yaml=as_yaml)

    cred = common.require_login(require_write=True)
    raw_text = text or ""
    if from_file is not None:
        raw_text = from_file.read_text(encoding="utf-8")

    content = raw_text.strip()
    if not content:
        common.exit_error("请提供动态文本。可用参数：TEXT 或 --from-file FILE")

    data = common.run_or_exit(
        client.post_text_dynamic(content, credential=cred),
        "发布动态失败",
    )

    dynamic_id = data.get("dynamic_id") or data.get("dynamic_id_str") or data.get("dyn_id")
    if common.emit_structured(
        payloads.action_result("dynamic_post", dynamic_id=str(dynamic_id or ""), text=content),
        output_format,
    ):
        return

    if dynamic_id:
        common.console.print(f"[green]✅ 已发布动态: {dynamic_id}[/green]")
    else:
        common.console.print("[green]✅ 已发布动态[/green]")


@click.command(name="dynamic-delete")
@click.argument("dynamic_id", type=int)
@click.option("--yes", is_flag=True, help="跳过确认，直接删除。")
@common.structured_output_options
def dynamic_delete(dynamic_id: int, yes: bool, as_json: bool, as_yaml: bool):
    """删除一条动态。"""
    from .. import client

    output_format = common.resolve_output_format(as_json=as_json, as_yaml=as_yaml)

    cred = common.require_login(require_write=True)

    if not yes:
        confirmed = click.confirm(f"确认删除动态 {dynamic_id} 吗？", default=False)
        if not confirmed:
            common.console.print("[yellow]已取消删除[/yellow]")
            return

    common.run_or_exit(
        client.delete_dynamic(dynamic_id=dynamic_id, credential=cred),
        "删除动态失败",
    )

    if common.emit_structured(
        payloads.action_result("dynamic_delete", dynamic_id=str(dynamic_id)),
        output_format,
    ):
        return

    common.console.print(f"[green]🗑️ 已删除动态: {dynamic_id}[/green]")