#!/usr/bin/env python3
"""Standard public client for server-backed online skills."""

from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import json
import os
import platform
import re
import socket
import stat
import subprocess
import sys
import tempfile
import urllib.error
import urllib.parse
import urllib.request
import uuid
from pathlib import Path
from typing import Any

from _protected_rules import find_protected_identity_leaks


DEFAULT_API_BASE = "https://superskill.online"
FALLBACK_API_BASES = [
    "http://43.156.61.150",
    "http://superskill.online",
]
PACKAGE_ROOT = Path(__file__).resolve().parents[1]
TRANSIENT_HTTP_STATUS = {408, 429, 500, 502, 503, 504}
AUTH_HTTP_STATUS = {401, 403}
AUTH_FAILURE_MESSAGE = "验证没有通过：当前激活码或手机号未授权，或本机设备与首次绑定设备不一致。请联系服务提供方处理。"
NEED_CREDENTIAL_MESSAGE = "请输入激活码或手机号码识别授权。首次验证后，新对话会自动读取最近一次凭证。"
VERSION_FILE = "online-version.json"
DEFAULT_CACHE_DAYS = 10
AUTH_CACHE_DIR_ENV = "ONLINE_SKILL_AUTH_CACHE_DIR"
CREDENTIAL_ENV = "ONLINE_SKILL_CREDENTIAL"
TEST_MODE_ENV = "ONLINE_SKILL_TEST_MODE"
TEST_RUN_ID_ENV = "ONLINE_SKILL_E2E_RUN_ID"
TEST_CREDENTIAL_ENV = "ONLINE_SKILL_E2E_CREDENTIAL"
DEFAULT_CACHE_DIR = Path.home() / ".codex" / "online-skill-auth"
# Compatibility aliases for callers that imported these names. Runtime code
# resolves the directory for every command so environment isolation works even
# when the module was imported before the environment was set.
CACHE_DIR = DEFAULT_CACHE_DIR
MACHINE_ID_FILE = DEFAULT_CACHE_DIR / "machine-id.json"
CLIENT_VERSION = "20260721"
TRUE_ENV_VALUES = {"1", "true", "yes", "on"}
SAFE_TEST_RUN_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
SAFE_SKILL_ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,127}$")
_RUNTIME_CACHE_DIR_OVERRIDE: Path | None = None
_RUNTIME_TEST_MODE_OVERRIDE = False


class AuthFailure(RuntimeError):
    def __init__(
        self,
        message: str,
        *,
        code: str = "AUTHORIZATION_FAILED",
        details: dict[str, Any] | None = None,
        status: int | None = None,
    ) -> None:
        super().__init__(message)
        self.code = code
        self.details = details or {}
        self.status = status


class ProtectedResponseLeak(RuntimeError):
    pass


def resolve_auth_cache_dir(explicit: str | Path | None = None) -> Path:
    value = explicit
    if value is None:
        value = (
            _RUNTIME_CACHE_DIR_OVERRIDE
            or os.environ.get(AUTH_CACHE_DIR_ENV, "").strip()
            or DEFAULT_CACHE_DIR
        )
    return Path(value).expanduser().resolve()


def args_auth_cache_dir(args: argparse.Namespace) -> Path:
    return resolve_auth_cache_dir(getattr(args, "auth_cache_dir", None))


def activate_args_auth_cache_dir(args: argparse.Namespace) -> Path:
    global _RUNTIME_CACHE_DIR_OVERRIDE, _RUNTIME_TEST_MODE_OVERRIDE
    _RUNTIME_CACHE_DIR_OVERRIDE = args_auth_cache_dir(args)
    _RUNTIME_TEST_MODE_OVERRIDE = bool(getattr(args, "test_mode", False))
    return _RUNTIME_CACHE_DIR_OVERRIDE


def test_mode_enabled(explicit: bool = False) -> bool:
    if explicit or _RUNTIME_TEST_MODE_OVERRIDE:
        return True
    return os.environ.get(TEST_MODE_ENV, "").strip().lower() in TRUE_ENV_VALUES


def args_test_mode(args: argparse.Namespace) -> bool:
    return test_mode_enabled(bool(getattr(args, "test_mode", False)))


def current_test_run_id() -> str | None:
    value = os.environ.get(TEST_RUN_ID_ENV, "").strip()
    if not value:
        return None
    if not SAFE_TEST_RUN_ID_RE.fullmatch(value):
        raise SystemExit(f"{TEST_RUN_ID_ENV} contains an invalid run id")
    return value


def now_utc() -> dt.datetime:
    return dt.datetime.now(dt.timezone.utc)


def parse_time(value: Any) -> dt.datetime | None:
    if not value:
        return None
    text = str(value).strip()
    if not text:
        return None
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    try:
        parsed = dt.datetime.fromisoformat(text)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=dt.timezone.utc)
    return parsed.astimezone(dt.timezone.utc)


def format_time(value: dt.datetime | None) -> str | None:
    if value is None:
        return None
    return value.astimezone(dt.timezone.utc).isoformat().replace("+00:00", "Z")


def read_json(path: Path) -> dict[str, Any]:
    try:
        data = json.loads(path.read_text("utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}
    return data if isinstance(data, dict) else {}


def write_json_private(path: Path, data: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8")
    descriptor, temporary = tempfile.mkstemp(
        prefix=f".{path.name}.",
        suffix=".tmp",
        dir=path.parent,
    )
    try:
        os.fchmod(descriptor, stat.S_IRUSR | stat.S_IWUSR)
        with os.fdopen(descriptor, "wb") as handle:
            descriptor = -1
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
        path.chmod(stat.S_IRUSR | stat.S_IWUSR)
    finally:
        if descriptor >= 0:
            os.close(descriptor)
        try:
            Path(temporary).unlink()
        except FileNotFoundError:
            pass


def read_skill_id(skill_root: Path, explicit: str | None) -> str:
    if explicit:
        skill_id = str(explicit).strip()
        if skill_id:
            return skill_id
    manifest = read_json(skill_root / VERSION_FILE)
    skill_id = str(manifest.get("skill_id") or "").strip()
    if not skill_id:
        raise SystemExit(
            "skill installation path error: online-version.json is missing or does not contain skill_id. "
            "Run the client from the installed skill package or pass the correct --skill-root/--skill-id."
        )
    return skill_id


def api_bases() -> list[str]:
    override = os.environ.get("WOO_API_BASE", "").strip().rstrip("/")
    if override:
        return [override]
    return [DEFAULT_API_BASE, *FALLBACK_API_BASES]


def normalize_credential(value: str | None) -> str:
    text = str(value or "").strip()
    return re.sub(r"[^A-Za-z0-9]", "", text).upper()


def credential_type(credential: str) -> str:
    if credential.isdigit() and 7 <= len(credential) <= 15:
        return "phone"
    return "activation_code"


def machine_id_path() -> Path:
    if test_mode_enabled():
        return resolve_auth_cache_dir() / "machine-id.json"
    return MACHINE_ID_FILE


def valid_machine_id(data: dict[str, Any]) -> str | None:
    if data.get("schema") != "online-skill-machine-id/v1":
        return None
    value = str(data.get("machine_id") or "").strip()
    try:
        parsed = uuid.UUID(value)
    except (ValueError, AttributeError):
        return None
    return str(parsed)


def read_existing_machine_id(path: Path) -> str | None:
    if not path.exists():
        return None
    value = valid_machine_id(read_json(path))
    if not value:
        raise SystemExit(
            f"machine identity file is invalid: {path}. "
            "Do not delete or regenerate it automatically; restore the backup or contact the service provider."
        )
    return value


def create_machine_id_exclusive(path: Path, value: str) -> bool:
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = json.dumps(
        {
            "schema": "online-skill-machine-id/v1",
            "machine_id": value,
            "created_at": format_time(now_utc()),
        },
        ensure_ascii=False,
        indent=2,
    ).encode("utf-8")
    try:
        descriptor = os.open(
            path,
            os.O_WRONLY | os.O_CREAT | os.O_EXCL,
            stat.S_IRUSR | stat.S_IWUSR,
        )
    except FileExistsError:
        return False
    try:
        with os.fdopen(descriptor, "wb") as handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
    except Exception:
        try:
            path.unlink()
        except OSError:
            pass
        raise
    return True


def stable_machine_id() -> str:
    machine_id_file = machine_id_path()
    existing = read_existing_machine_id(machine_id_file)
    if existing:
        return existing
    value = str(uuid.uuid4())
    if create_machine_id_exclusive(machine_id_file, value):
        return value
    existing = read_existing_machine_id(machine_id_file)
    if existing:
        return existing
    raise SystemExit("machine identity creation failed")


def legacy_fingerprint_v20260705() -> str:
    raw = "|".join(
        [
            platform.system(),
            platform.release(),
            platform.machine(),
            socket.gethostname(),
            str(Path.home()),
        ]
    )
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


def legacy_fingerprint_v1() -> str:
    parts = [
        platform.system(),
        platform.release(),
        platform.machine(),
        socket.gethostname(),
        str(Path.home()),
    ]
    if platform.system() == "Darwin":
        try:
            output = subprocess.check_output(
                ["ioreg", "-rd1", "-c", "IOPlatformExpertDevice"],
                text=True,
                stderr=subprocess.DEVNULL,
                timeout=2,
            )
            for line in output.splitlines():
                if "IOPlatformUUID" in line:
                    parts.append(line.split("=", 1)[1].strip().strip('"'))
                    break
        except Exception:
            pass
    return hashlib.sha256("|".join(part for part in parts if part).encode("utf-8")).hexdigest()


def legacy_device_fingerprints() -> list[str]:
    return list(dict.fromkeys([legacy_fingerprint_v20260705(), legacy_fingerprint_v1()]))


def device_fingerprint() -> str:
    return stable_machine_id()


def short_hash(value: str) -> str:
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


def cache_key(skill_id: str, credential: str, device_hash: str) -> str:
    return short_hash(f"{skill_id}\0{credential}\0{device_hash}")[:32]


def cache_path(
    skill_id: str,
    credential: str,
    device_hash: str,
    auth_cache_dir: str | Path | None = None,
) -> Path:
    return (
        resolve_auth_cache_dir(auth_cache_dir)
        / skill_id
        / f"{cache_key(skill_id, credential, device_hash)}.json"
    )


def last_credential_path(
    skill_id: str,
    auth_cache_dir: str | Path | None = None,
) -> Path:
    return resolve_auth_cache_dir(auth_cache_dir) / skill_id / "last-credential.json"


def store_last_credential(
    skill_id: str,
    credential: str,
    auth_cache_dir: str | Path | None = None,
) -> None:
    write_json_private(
        last_credential_path(skill_id, auth_cache_dir),
        {
            "skill_id": skill_id,
            "credential": credential,
            "credential_hash": short_hash(credential),
            "updated_at": format_time(now_utc()),
        },
    )


def read_last_credential(
    skill_id: str,
    auth_cache_dir: str | Path | None = None,
) -> str | None:
    data = read_json(last_credential_path(skill_id, auth_cache_dir))
    if str(data.get("skill_id") or "") != skill_id:
        return None
    credential = normalize_credential(data.get("credential"))
    if not credential or data.get("credential_hash") != short_hash(credential):
        return None
    return credential or None


def provided_credential(args: argparse.Namespace) -> str | None:
    credential = normalize_credential(
        getattr(args, "credential", None) or getattr(args, "phone", None)
    )
    if not credential:
        credential = normalize_credential(os.environ.get(CREDENTIAL_ENV))
    if not credential and args_test_mode(args):
        credential = normalize_credential(os.environ.get(TEST_CREDENTIAL_ENV))
    return credential or None


def effective_credential(args: argparse.Namespace, skill_id: str) -> str | None:
    credential = provided_credential(args)
    if credential:
        return credential
    return read_last_credential(skill_id, args_auth_cache_dir(args))


def auth_headers(cache: dict[str, Any] | None = None) -> dict[str, str]:
    headers = {"Content-Type": "application/json"}
    if not cache:
        return headers
    token = (
        cache.get("access_token")
        or cache.get("auth_token")
        or cache.get("cache_token")
        or cache.get("signed_cache")
    )
    if token:
        headers["Authorization"] = f"Bearer {token}"
    return headers


def request_json(base: str, endpoint: str, payload: dict[str, Any] | None, timeout: int, headers: dict[str, str]) -> dict[str, Any]:
    url = f"{base.rstrip('/')}{endpoint}"
    body = None
    method = "GET"
    if payload is not None:
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        method = "POST"
    request = urllib.request.Request(url, data=body, headers=headers, method=method)
    with urllib.request.urlopen(request, timeout=timeout) as response:
        text = response.read().decode("utf-8")
    if not text.strip():
        return {}
    data = json.loads(text)
    if isinstance(data, dict):
        return data
    if isinstance(data, list):
        return {"results": data}
    return {"data": data}


AUTH_ERROR_MESSAGES = {
    "AUTHORIZATION_FAILED": AUTH_FAILURE_MESSAGE,
    "SKILL_MISMATCH": AUTH_FAILURE_MESSAGE,
    "CREDENTIAL_NOT_FOUND": "当前激活码或手机号码没有这个 Skill 的授权记录。",
    "LICENSE_DISABLED": "该授权已被禁用。",
    "LICENSE_EXPIRED": "该授权已过期，请联系管理员续期。",
    "DEVICE_NOT_BOUND": "该授权尚未绑定设备，客户端将尝试完成首次绑定。",
    "DEVICE_MISMATCH": "验证没有通过，当前设备未授权，请联系管理员处理。",
    "DEVICE_LIMIT": "该授权已绑定设备数量达到上限，请联系管理员换绑。",
}


def safe_skill_id(value: Any) -> str | None:
    text = str(value or "").strip()
    return text if SAFE_SKILL_ID_RE.fullmatch(text) else None


def safe_public_url(value: Any, *, required_path_prefix: str | None = None) -> str | None:
    text = str(value or "").strip()
    if not text:
        return None
    parsed = urllib.parse.urlparse(text)
    if (
        parsed.scheme != "https"
        or parsed.hostname != "superskill.online"
        or parsed.username
        or parsed.password
        or parsed.port not in (None, 443)
        or parsed.query
        or parsed.fragment
    ):
        return None
    if required_path_prefix and not parsed.path.startswith(required_path_prefix):
        return None
    return text


def sanitize_auth_details(value: Any) -> dict[str, Any]:
    if not isinstance(value, dict):
        return {}
    detail: dict[str, Any] = {}
    requested = safe_skill_id(value.get("requested_skill_id"))
    if requested:
        detail["requested_skill_id"] = requested
    raw_authorized_ids = value.get("authorized_skill_ids")
    if not isinstance(raw_authorized_ids, list):
        raw_authorized_ids = []
    authorized_ids = sorted(
        {
            skill_id
            for skill_id in (safe_skill_id(item) for item in raw_authorized_ids)
            if skill_id
        }
    )
    if authorized_ids:
        detail["authorized_skill_ids"] = authorized_ids
    safe_targets: list[dict[str, Any]] = []
    targets = value.get("authorized_skills")
    if isinstance(targets, list):
        for item in targets:
            if not isinstance(item, dict):
                continue
            skill_id = safe_skill_id(item.get("skill_id"))
            if not skill_id or skill_id not in authorized_ids:
                continue
            target: dict[str, Any] = {"skill_id": skill_id}
            guide_url = safe_public_url(
                item.get("guide_url"),
                required_path_prefix=f"/guides/{skill_id}/",
            )
            version_url = safe_public_url(
                item.get("version_url"),
                required_path_prefix=f"/v1/skills/{skill_id}/version",
            )
            package_url = safe_public_url(
                item.get("public_package_url"),
                required_path_prefix="/downloads/",
            )
            if guide_url:
                target["guide_url"] = guide_url
            if version_url:
                target["version_url"] = version_url
            if package_url:
                target["public_package_url"] = package_url
            for key in (
                "public_skill_version",
                "public_package_filename",
                "public_package_sha256",
            ):
                text = str(item.get(key) or "").strip()
                if text and len(text) <= 256:
                    target[key] = text
            safe_targets.append(target)
    if safe_targets:
        detail["authorized_skills"] = safe_targets
    for key in ("bound_device_count", "max_device_count"):
        value_int = value.get(key)
        if isinstance(value_int, int) and value_int >= 0:
            detail[key] = value_int
    return detail


def auth_failure_from_http(status: int, raw_detail: str) -> AuthFailure:
    structured: dict[str, Any] = {}
    try:
        decoded = json.loads(raw_detail)
    except json.JSONDecodeError:
        decoded = {}
    if isinstance(decoded, dict):
        nested = decoded.get("detail")
        if isinstance(nested, dict):
            structured = nested
        elif isinstance(decoded.get("error"), dict):
            structured = decoded["error"]
        else:
            structured = decoded
    code = str(structured.get("error_code") or "").strip().upper()
    if code not in AUTH_ERROR_MESSAGES:
        code = "AUTHORIZATION_FAILED"
    details = sanitize_auth_details(structured)
    if code == "SKILL_MISMATCH":
        requested = details.get("requested_skill_id") or "当前 Skill"
        authorized = details.get("authorized_skill_ids") or []
        if authorized:
            message = (
                f"这个激活码或手机号码没有 {requested} 权益。"
                f"它对应的 Skill 是：{'、'.join(authorized)}。"
                "请下载安装正确的 Skill；当前授权没有失效，无需换绑。"
            )
        else:
            code = "AUTHORIZATION_FAILED"
            message = AUTH_FAILURE_MESSAGE
    else:
        message = AUTH_ERROR_MESSAGES[code]
    return AuthFailure(message, code=code, details=details, status=status)


def public_auth_failure(skill_id: str, exc: AuthFailure) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "ok": False,
        "skill_id": skill_id,
        "error_code": exc.code,
        "reason": str(exc),
    }
    if exc.code == "SKILL_MISMATCH":
        for key in ("authorized_skill_ids", "authorized_skills"):
            if key in exc.details:
                payload[key] = exc.details[key]
    for key in ("bound_device_count", "max_device_count"):
        if key in exc.details:
            payload[key] = exc.details[key]
    return payload


def call_api(
    endpoint: str,
    *,
    payload: dict[str, Any] | None = None,
    timeout: int,
    headers: dict[str, str] | None = None,
) -> dict[str, Any]:
    errors: list[str] = []
    headers = headers or {"Content-Type": "application/json"}
    for base in api_bases():
        try:
            data = request_json(base, endpoint, payload, timeout, headers)
            data.setdefault("api_base", base)
            return data
        except urllib.error.HTTPError as exc:
            raw_detail = exc.read(65_536).decode("utf-8", errors="ignore")
            detail = raw_detail[:500]
            if exc.code in AUTH_HTTP_STATUS:
                raise auth_failure_from_http(exc.code, raw_detail)
            if exc.code in TRANSIENT_HTTP_STATUS:
                errors.append(f"{base}: HTTP {exc.code}: {detail}")
                continue
            if exc.code == 404:
                raise SystemExit(
                    "skill configuration error: the server does not recognize this skill_id. "
                    "Check the installed skill root and online-version.json; this is not an authorization-service outage."
                )
            raise SystemExit(f"request failed with HTTP {exc.code}: {detail}")
        except (urllib.error.URLError, TimeoutError, OSError, json.JSONDecodeError) as exc:
            errors.append(f"{base}: {exc}")
    raise SystemExit("request failed: " + " | ".join(errors))


def cache_expiry_from_response(data: dict[str, Any]) -> tuple[dt.datetime, str | None, str | None]:
    base_expiry = now_utc() + dt.timedelta(days=DEFAULT_CACHE_DAYS)
    license_expires_at = data.get("license_expires_at")
    server_cache_expires_at = data.get("auth_cache_expires_at") or data.get("cache_expires_at")
    license_dt = parse_time(license_expires_at)
    server_cache_dt = parse_time(server_cache_expires_at)
    candidates = [base_expiry]
    if license_dt:
        candidates.append(license_dt)
    if server_cache_dt:
        candidates.append(server_cache_dt)
    auth_cache_dt = min(candidates)
    return auth_cache_dt, format_time(license_dt), format_time(auth_cache_dt)


SUCCESS_STATUS_VALUES = {"ok", "success", "succeeded", "authorized", "valid", "active", "bound"}


def explicit_success_value(value: Any) -> bool:
    if value is True or value == 1:
        return True
    if value is False or value is None or value == 0:
        return False
    if isinstance(value, str):
        return value.strip().lower() in SUCCESS_STATUS_VALUES | {"true", "yes"}
    return False


def has_explicit_status(data: dict[str, Any]) -> bool:
    return any(key in data for key in ("ok", "success", "authorized", "valid", "status"))


def is_success_response(data: dict[str, Any]) -> bool:
    indicators = [data[key] for key in ("ok", "success", "authorized", "valid") if key in data]
    if indicators:
        return all(explicit_success_value(value) for value in indicators)
    if "status" in data:
        return str(data.get("status") or "").strip().lower() in SUCCESS_STATUS_VALUES
    return False


def has_protected_result_payload(data: dict[str, Any]) -> bool:
    return any(key in data for key in ("results", "items", "matches", "data", "refs", "reference", "content"))


def bind_credential(
    skill_id: str,
    credential: str,
    timeout: int,
    *,
    auth_cache_dir: str | Path | None = None,
    write_last_credential_entry: bool | None = None,
) -> dict[str, Any]:
    device_hash = device_fingerprint()
    kind = credential_type(credential)
    payload = {
        "skill_id": skill_id,
        "credential": credential,
        "credential_type": kind,
        "phone": credential,
        "activation_code": credential,
        "code": credential,
        "invite_code": credential,
        "device_fingerprint": device_hash,
        "device_hash": device_hash,
        "machine_id": stable_machine_id(),
        "legacy_device_fingerprints": legacy_device_fingerprints(),
        "client_version": CLIENT_VERSION,
    }
    test_run_id = current_test_run_id() if test_mode_enabled() else None
    if test_run_id:
        payload["test_run_id"] = test_run_id
    endpoint = f"/v1/skills/{urllib.parse.quote(skill_id)}/device/bind"
    data = call_api(endpoint, payload=payload, timeout=timeout, headers=auth_headers())
    if not is_success_response(data):
        raise AuthFailure(AUTH_FAILURE_MESSAGE)
    auth_cache_dt, license_expiry, cache_expiry = cache_expiry_from_response(data)
    cache = {
        "schema": "online-skill-auth-cache/v1",
        "skill_id": skill_id,
        "credential_hash": short_hash(credential),
        "credential_type": data.get("credential_type") or kind,
        "device_fingerprint_hash": device_hash,
        "machine_id_hash": short_hash(stable_machine_id()),
        "created_at": format_time(now_utc()),
        "license_expires_at": license_expiry,
        "auth_cache_expires_at": cache_expiry,
        "api_base": data.get("api_base"),
        "access_token": data.get("access_token"),
        "auth_token": data.get("auth_token"),
        "cache_token": data.get("cache_token"),
        "signed_cache": data.get("signed_cache") or data.get("auth_cache"),
        "server_response": public_response_for_endpoint(data, endpoint),
    }
    write_json_private(
        cache_path(skill_id, credential, device_hash, auth_cache_dir),
        cache,
    )
    if test_mode_enabled():
        write_last_credential_entry = False
    elif write_last_credential_entry is None:
        write_last_credential_entry = True
    if write_last_credential_entry:
        store_last_credential(skill_id, credential, auth_cache_dir)
    return cache


def load_valid_cache(
    skill_id: str,
    credential: str,
    auth_cache_dir: str | Path | None = None,
) -> dict[str, Any] | None:
    state, cache = inspect_cache(skill_id, credential, auth_cache_dir)
    return cache if state == "valid" else None


def inspect_cache(
    skill_id: str,
    credential: str,
    auth_cache_dir: str | Path | None = None,
) -> tuple[str, dict[str, Any] | None]:
    device_hash = device_fingerprint()
    path = cache_path(skill_id, credential, device_hash, auth_cache_dir)
    if not path.exists():
        return "missing", None
    cache = read_json(path)
    if not cache:
        return "invalid", None
    if cache.get("schema") != "online-skill-auth-cache/v1":
        return "invalid", None
    if cache.get("skill_id") != skill_id:
        return "invalid", None
    if cache.get("credential_hash") != short_hash(credential):
        return "invalid", None
    if cache.get("device_fingerprint_hash") != device_hash:
        return "invalid", None
    if cache.get("machine_id_hash") != short_hash(stable_machine_id()):
        return "invalid", None
    expires_at = parse_time(cache.get("auth_cache_expires_at"))
    if not expires_at or expires_at <= now_utc():
        return "expired", None
    return "valid", cache


def get_or_bind_cache(args: argparse.Namespace, skill_id: str, *, allow_bind: bool) -> tuple[str, dict[str, Any] | None]:
    credential = effective_credential(args, skill_id)
    if not credential:
        return "", None
    cache_dir = args_auth_cache_dir(args)
    cache = load_valid_cache(skill_id, credential, cache_dir)
    if cache:
        if provided_credential(args) and not args_test_mode(args):
            store_last_credential(skill_id, credential, cache_dir)
        return credential, cache
    if not allow_bind:
        return credential, cache
    cache = bind_credential(
        skill_id,
        credential,
        args.timeout,
        auth_cache_dir=cache_dir,
        write_last_credential_entry=not args_test_mode(args),
    )
    return credential, cache


def auth_payload(skill_id: str, credential: str, cache: dict[str, Any], extra: dict[str, Any] | None = None) -> dict[str, Any]:
    device_hash = device_fingerprint()
    payload = {
        "skill_id": skill_id,
        "credential": credential,
        "credential_type": cache.get("credential_type") or credential_type(credential),
        "phone": credential,
        "activation_code": credential,
        "code": credential,
        "invite_code": credential,
        "device_fingerprint": device_hash,
        "device_hash": device_hash,
        "machine_id": stable_machine_id(),
        "legacy_device_fingerprints": legacy_device_fingerprints(),
        "client_version": CLIENT_VERSION,
        "auth_cache": cache.get("signed_cache") or cache.get("cache_token"),
    }
    if extra:
        payload.update(extra)
    test_run_id = current_test_run_id() if test_mode_enabled() else None
    if test_run_id:
        payload["test_run_id"] = test_run_id
    return payload


SENSITIVE_RESPONSE_FIELDS = {
    "auth_cache",
    "signed_cache",
    "cache_token",
    "auth_token",
    "access_token",
    "authorization",
    "cache_ttl_seconds",
    "auth_cache_expires_at",
    "cache_expires_at",
    "database_key",
    "decrypted_payload",
    "credential",
    "phone",
    "activation_code",
    "code",
    "invite_code",
    "device_fingerprint",
    "device_hash",
    "machine_id",
    "legacy_device_fingerprints",
}


def redact_protected_response(value: Any) -> Any:
    """Remove internal authorization/cache material from visible and cached data."""
    if isinstance(value, dict):
        return {
            key: redact_protected_response(child)
            for key, child in value.items()
            if str(key).lower() not in SENSITIVE_RESPONSE_FIELDS
        }
    if isinstance(value, list):
        return [redact_protected_response(child) for child in value]
    return value


SOURCE_IDENTITY_FIELDS = {
    "source_path",
    "rel_path",
    "sample_paths",
    "cache_file",
    "file_name",
    "filename",
    "path",
    "local_path",
    "source_url",
    "source_page",
    "referer",
    "manifest",
    "catalog",
    "raw_manifest",
    "source_id",
    "film_id",
    "script_name",
    "book_name",
    "article_title",
    "work_title",
    "source_title",
    "original_title",
    "source",
    "source_name",
    "original_source",
}
SOURCE_TITLE_FIELDS = {"title", "work", "book", "article", "script"}
LOCAL_PATH_LOCATOR_RE = re.compile(
    r"(?:file://|(?:^|(?<=[\s`\"'(]))~/|"
    r"/(?:Users|Volumes|private|tmp|opt|var|home|mnt|srv|root|Library|Applications)(?:/|\b)|"
    r"\b[A-Za-z]:[\\/]|(?<!:)(?:\\\\|//)[^\\/\s]+[\\/][^\\/\s]+)",
    re.IGNORECASE,
)
SOURCE_URL_FRAGMENT_RE = re.compile(
    r"(?:https?|s3|cos)://[^\s`\"'<>)]+",
    re.IGNORECASE,
)


def redact_source_identity(value: Any) -> Any:
    """Remove source identity fields and machine/URL fragments recursively."""
    if isinstance(value, dict):
        public: dict[str, Any] = {}
        for key, child in value.items():
            key_lower = str(key).lower()
            if key_lower in SOURCE_IDENTITY_FIELDS | SOURCE_TITLE_FIELDS:
                continue
            public[key] = redact_source_identity(child)
        return public
    if isinstance(value, list):
        return [redact_source_identity(child) for child in value]
    if isinstance(value, str):
        if LOCAL_PATH_LOCATOR_RE.search(value):
            return "[redacted source identity]"
        return SOURCE_URL_FRAGMENT_RE.sub("[redacted source identity]", value)
    return value


FULL_TEXT_BODY_FIELDS = {
    "snippet",
    "content",
    "text",
    "raw_text",
    "original_text",
    "source_text",
    "full_text",
}
FULL_TEXT_RESULT_FIELDS = {"result", "results", "items", "matches", "data"}
FULL_TEXT_SAFE_METADATA_FIELDS = {
    "ok",
    "success",
    "authorized",
    "valid",
    "status",
    "count",
    "total",
    "total_count",
    "result_count",
    "api_base",
    "skill_id",
    "license_expires_at",
}
PUBLIC_METADATA_FIELDS_EXCLUDED_FROM_LEAK_SCAN = {
    "api_base",
    "public_package_url",
}
FULL_TEXT_CONFIRMATION = (
    "云端全文索引已命中查询词；公开客户端不返回原句、角色名、片名或来源标识。"
)


def anonymous_full_text_item(item: dict[str, Any]) -> dict[str, Any]:
    seed = json.dumps(item, ensure_ascii=False, sort_keys=True)
    anonymous: dict[str, Any] = {
        "id": "fulltext_" + hashlib.sha256(seed.encode("utf-8")).hexdigest()[:16],
        "category": "匿名全文机制命中",
        "snippet": FULL_TEXT_CONFIRMATION,
        "next_action": "Use this only as anonymous mechanism evidence; do not request or expose source identity or original wording.",
    }
    char_count = item.get("char_count")
    if isinstance(char_count, (int, float)) and not isinstance(char_count, bool):
        anonymous["char_count"] = char_count
    return anonymous


def anonymize_full_text_shape(value: Any) -> Any:
    """Anonymize one value stored under results/items/matches/data/result."""
    if isinstance(value, list):
        return [anonymize_full_text_shape(child) for child in value]
    if isinstance(value, dict):
        if any(str(key).lower() in FULL_TEXT_RESULT_FIELDS for key in value):
            return anonymize_full_text_response(value)
        return anonymous_full_text_item(value)
    return FULL_TEXT_CONFIRMATION


def anonymize_full_text_response(value: Any) -> Any:
    """Recursively anonymize scalar and nested full-text response shapes."""
    if isinstance(value, dict):
        public: dict[str, Any] = {}
        for key, child in value.items():
            key_lower = str(key).lower()
            if key_lower in FULL_TEXT_BODY_FIELDS:
                public[key] = FULL_TEXT_CONFIRMATION
            elif key_lower in FULL_TEXT_RESULT_FIELDS:
                public[key] = anonymize_full_text_shape(child)
            elif key_lower in FULL_TEXT_SAFE_METADATA_FIELDS:
                public[key] = child
            else:
                public[key] = anonymize_full_text_response(child)
        return public
    if isinstance(value, list):
        return [anonymize_full_text_shape(child) for child in value]
    return FULL_TEXT_CONFIRMATION


def response_for_leak_scan(value: Any) -> Any:
    if isinstance(value, dict):
        return {
            key: response_for_leak_scan(child)
            for key, child in value.items()
            if str(key).lower() not in PUBLIC_METADATA_FIELDS_EXCLUDED_FROM_LEAK_SCAN
        }
    if isinstance(value, list):
        return [response_for_leak_scan(child) for child in value]
    return value


def public_response_for_endpoint(value: Any, endpoint: str) -> Any:
    """Create public output and fail closed if protected identity remains."""
    public = redact_source_identity(redact_protected_response(value))
    if "/full-text/search" in endpoint:
        public = anonymize_full_text_response(public)
    leaks = find_protected_identity_leaks(response_for_leak_scan(public))
    if leaks:
        raise ProtectedResponseLeak(
            "protected response failed recursive source-identity validation"
        )
    return public


def print_json(data: Any, *, exit_code: int = 0) -> int:
    print(json.dumps(data, ensure_ascii=False, indent=2))
    return exit_code


def public_authorization_fields(cache: dict[str, Any], credential: str) -> dict[str, Any]:
    server_response = cache.get("server_response") if isinstance(cache.get("server_response"), dict) else {}
    return {
        "credential_type": cache.get("credential_type") or credential_type(credential),
        "credential_hash": short_hash(credential),
        "license_activated_at": server_response.get("license_activated_at"),
        "license_expires_at": cache.get("license_expires_at"),
        "license_duration_label": server_response.get("license_duration_label"),
        "license_status": server_response.get("license_status"),
        "license_is_permanent": server_response.get("license_is_permanent"),
        "api_base": cache.get("api_base"),
    }


def command_cache_status(args: argparse.Namespace) -> int:
    activate_args_auth_cache_dir(args)
    skill_root = args.skill_root.expanduser().resolve()
    skill_id = read_skill_id(skill_root, args.skill_id)
    credential = effective_credential(args, skill_id)
    if not credential:
        return print_json(
            {
                "ok": False,
                "auth_state": "credential_missing",
                "reason": NEED_CREDENTIAL_MESSAGE,
                "skill_id": skill_id,
            },
            exit_code=1,
        )
    state, cache = inspect_cache(skill_id, credential, args_auth_cache_dir(args))
    if state != "valid" or not cache:
        auth_state = {
            "missing": "local_cache_missing_refreshable",
            "expired": "local_cache_expired_refreshable",
            "invalid": "local_cache_invalid_refreshable",
        }[state]
        return print_json(
            {
                "ok": False,
                "auth_state": auth_state,
                "skill_id": skill_id,
                "credential_type": credential_type(credential),
                "credential_hash": short_hash(credential),
                "reason": (
                    "本地授权缓存需要联网刷新；这不等于云端授权失效。"
                    "请运行 ensure-auth 自动续验。"
                ),
            },
            exit_code=1,
        )
    if provided_credential(args) and not args_test_mode(args):
        store_last_credential(skill_id, credential, args_auth_cache_dir(args))
    payload = {"ok": True, "auth_state": "local_cache_valid", "skill_id": skill_id}
    payload.update(public_authorization_fields(cache, credential))
    return print_json(payload)


def command_ensure_auth(args: argparse.Namespace) -> int:
    activate_args_auth_cache_dir(args)
    skill_root = args.skill_root.expanduser().resolve()
    skill_id = read_skill_id(skill_root, args.skill_id)
    credential = effective_credential(args, skill_id)
    if not credential:
        return print_json(
            {
                "ok": False,
                "auth_state": "credential_missing",
                "reason": NEED_CREDENTIAL_MESSAGE,
                "skill_id": skill_id,
            },
            exit_code=1,
        )
    cache_dir = args_auth_cache_dir(args)
    previous_state, cache = inspect_cache(skill_id, credential, cache_dir)
    if previous_state == "valid" and cache:
        if provided_credential(args) and not args_test_mode(args):
            store_last_credential(skill_id, credential, cache_dir)
        payload = {
            "ok": True,
            "auth_state": "local_cache_valid",
            "skill_id": skill_id,
        }
        payload.update(public_authorization_fields(cache, credential))
        return print_json(payload)
    try:
        cache = bind_credential(
            skill_id,
            credential,
            args.timeout,
            auth_cache_dir=cache_dir,
            write_last_credential_entry=not args_test_mode(args),
        )
    except AuthFailure as exc:
        payload = public_auth_failure(skill_id, exc)
        payload["auth_state"] = "cloud_verification_failed"
        payload["previous_cache_state"] = previous_state
        return print_json(payload, exit_code=1)
    except ProtectedResponseLeak as exc:
        return print_json(
            {
                "ok": False,
                "auth_state": "cloud_verification_failed",
                "skill_id": skill_id,
                "reason": str(exc),
            },
            exit_code=1,
        )
    payload = {
        "ok": True,
        "auth_state": "cloud_reverified",
        "previous_cache_state": previous_state,
        "skill_id": skill_id,
    }
    payload.update(public_authorization_fields(cache, credential))
    return print_json(payload)


def command_bind(args: argparse.Namespace) -> int:
    activate_args_auth_cache_dir(args)
    skill_root = args.skill_root.expanduser().resolve()
    skill_id = read_skill_id(skill_root, args.skill_id)
    credential = provided_credential(args)
    if not credential:
        print_json({"ok": False, "reason": NEED_CREDENTIAL_MESSAGE, "skill_id": skill_id})
        return 1
    try:
        cache = bind_credential(
            skill_id,
            credential,
            args.timeout,
            auth_cache_dir=args_auth_cache_dir(args),
            write_last_credential_entry=not args_test_mode(args),
        )
    except AuthFailure as exc:
        print_json(public_auth_failure(skill_id, exc))
        return 1
    except ProtectedResponseLeak as exc:
        print_json({"ok": False, "reason": str(exc), "skill_id": skill_id})
        return 1
    payload = {"ok": True, "skill_id": skill_id}
    payload.update(public_authorization_fields(cache, credential))
    return print_json(payload)


def protected_call(args: argparse.Namespace, endpoint: str, extra: dict[str, Any]) -> int:
    activate_args_auth_cache_dir(args)
    skill_root = args.skill_root.expanduser().resolve()
    skill_id = read_skill_id(skill_root, args.skill_id)
    try:
        credential, cache = get_or_bind_cache(args, skill_id, allow_bind=True)
        if not credential or not cache:
            return print_json(
                {"ok": False, "reason": NEED_CREDENTIAL_MESSAGE, "skill_id": skill_id},
                exit_code=1,
            )
        payload = auth_payload(skill_id, credential, cache, extra)
        data = call_api(endpoint, payload=payload, timeout=args.timeout, headers=auth_headers(cache))
    except AuthFailure as exc:
        return print_json(public_auth_failure(skill_id, exc), exit_code=1)
    except ProtectedResponseLeak as exc:
        return print_json({"ok": False, "reason": str(exc), "skill_id": skill_id}, exit_code=1)
    if not is_success_response(data):
        if has_explicit_status(data) or not has_protected_result_payload(data):
            return print_json({"ok": False, "reason": AUTH_FAILURE_MESSAGE, "skill_id": skill_id}, exit_code=1)
        data["ok"] = True
    try:
        public_data = public_response_for_endpoint(data, endpoint)
    except ProtectedResponseLeak:
        return print_json(
            {
                "ok": False,
                "reason": "protected response failed privacy validation",
                "skill_id": skill_id,
            },
            exit_code=2,
        )
    public_data.setdefault("skill_id", skill_id)
    public_data.setdefault("license_expires_at", cache.get("license_expires_at"))
    return print_json(public_data)


def command_search(args: argparse.Namespace) -> int:
    skill_id = read_skill_id(args.skill_root.expanduser().resolve(), args.skill_id)
    endpoint = (
        f"/v1/skills/{urllib.parse.quote(skill_id)}/full-text/search"
        if args.mode == "full_text"
        else f"/v1/skills/{urllib.parse.quote(skill_id)}/search"
    )
    extra = {
        "query": args.query,
        "top_k": args.top_k,
        "top-k": args.top_k,
        "mode": args.mode,
    }
    return protected_call(args, endpoint, extra)


def command_list_refs(args: argparse.Namespace) -> int:
    skill_id = read_skill_id(args.skill_root.expanduser().resolve(), args.skill_id)
    endpoint = f"/v1/skills/{urllib.parse.quote(skill_id)}/refs"
    return protected_call(args, endpoint, {"action": "list"})


def command_read_ref(args: argparse.Namespace) -> int:
    skill_id = read_skill_id(args.skill_root.expanduser().resolve(), args.skill_id)
    ref_name = urllib.parse.quote(args.name.strip(), safe="")
    endpoint = f"/v1/skills/{urllib.parse.quote(skill_id)}/refs/{ref_name}"
    return protected_call(args, endpoint, {"name": args.name, "ref": args.name})


def add_common_auth_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--skill-root", type=Path, default=PACKAGE_ROOT)
    parser.add_argument("--skill-id")
    parser.add_argument("--credential", help="Activation code or phone credential")
    parser.add_argument("--phone", help="Backward-compatible alias for --credential")
    parser.add_argument(
        "--auth-cache-dir",
        type=Path,
        help=f"Isolated authorization cache directory (or set {AUTH_CACHE_DIR_ENV})",
    )
    parser.add_argument(
        "--test-mode",
        action="store_true",
        help="Use E2E isolation and never write last-credential.json",
    )
    parser.add_argument("--timeout", type=int, default=30)


def main() -> int:
    parser = argparse.ArgumentParser(description="Call protected online-skill endpoints")
    subparsers = parser.add_subparsers(dest="command", required=True)

    cache_status = subparsers.add_parser("cache-status", help="Show local authorization cache status")
    add_common_auth_args(cache_status)
    cache_status.set_defaults(func=command_cache_status)

    ensure_auth = subparsers.add_parser(
        "ensure-auth",
        help="Use a valid local cache or refresh it from the cloud",
    )
    add_common_auth_args(ensure_auth)
    ensure_auth.set_defaults(func=command_ensure_auth)

    bind = subparsers.add_parser("bind", help="Bind or refresh authorization for this device")
    add_common_auth_args(bind)
    bind.set_defaults(func=command_bind)

    search = subparsers.add_parser("search", help="Search protected cloud data")
    search.add_argument("query")
    search.add_argument("--top-k", type=int, default=6)
    search.add_argument("--mode", default="search")
    add_common_auth_args(search)
    search.set_defaults(func=command_search)

    full_text = subparsers.add_parser("full-text-search", help="Search protected full-text data")
    full_text.add_argument("query")
    full_text.add_argument("--top-k", type=int, default=6)
    full_text.set_defaults(mode="full_text")
    add_common_auth_args(full_text)
    full_text.set_defaults(func=command_search)

    list_refs = subparsers.add_parser("list-refs", help="List protected remote references")
    add_common_auth_args(list_refs)
    list_refs.set_defaults(func=command_list_refs)

    read_ref = subparsers.add_parser("read-ref", help="Read one protected remote reference")
    read_ref.add_argument("name")
    add_common_auth_args(read_ref)
    read_ref.set_defaults(func=command_read_ref)

    args = parser.parse_args()
    return args.func(args)


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except KeyboardInterrupt:
        raise SystemExit(130)
