Byted Market Insight Agent

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

   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION

   1. Definitions.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

      You may add Your own license statement for Your modifications and
      may provide additional grant of rights to use, copy, modify, merge,
      publish, distribute, sublicense, and/or sell copies of the
      Contribution, either before or after.

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

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

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

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

   9. Accepting Warranty or Additional Liability. While redistributing
      the Work or Derivative Works thereof, You may choose to offer,
      and charge a fee for, acceptance of support, warranty, indemnity,
      or other liability obligations and/or rights consistent with this
      License. However, in accepting such obligations, You may offer only
      conditions that are (a) consistent with the terms of this License, and
      (b) include a complete copy of this License. Upon Your request, the
      Licensor may provide such Contributor access to the License terms.

   END OF TERMS AND CONDITIONS

   Copyright 2024 ByteDance, Inc.

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

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

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

#!/usr/bin/env python3
# Copyright 2024 ByteDance, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""市场洞察 Agent 统一鉴权与 Provider 选择模块。

职责：
- 读取环境变量与本地持久化状态/凭证；
- 在 Gateway 与 SDK 之间做出决策，默认优先 Gateway；
- 实现会话级粘性（尽量复用上一次成功的 Provider）；
- 在需要 SDK 时自动安装 `volcengine-python-sdk>=5.0.22`；
- 将状态持久化到 `persist/state.json`，凭证持久化到 `persist/auth.json`；
- 暴露“最小化询问”占位提示，由 Agent 层据此调用 ask_user 获取凭证。

注意：本模块不做任何真实网络调用，只负责决策与配置。
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import sys
import time
from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Dict, Optional, Set

# ------------------------ 路径与常量 ------------------------

ROOT_DIR = Path(__file__).resolve().parent.parent
PERSIST_DIR = ROOT_DIR / "persist"
STATE_FILE = PERSIST_DIR / "state.json"
AUTH_FILE = PERSIST_DIR / "auth.json"

# SDK 依赖要求
SDK_PACKAGE_NAME = "volcengine-python-sdk"
SDK_MIN_VERSION_SPEC = "volcengine-python-sdk>=5.0.22"

# 提示 Agent 层如何最小化询问用户
MINIMAL_ASK_HINT = (
    "当前缺少可用的市场洞察访问凭证。\n"
    "Agent 层应调用 ask_user，引导用户择一提供：\n"
    "1) API Gateway 地址 + API Key (ARK_SKILL_API_BASE / ARK_SKILL_API_KEY)，或\n"
    "2) 官方 AK/SK (VOLCSTACK_ACCESS_KEY_ID / VOLCSTACK_SECRET_ACCESS_KEY)。\n"
    "一套凭证即可，Skill 会自动选择最合适的链路，并在本地持久化，后续会话复用。"
)


# ------------------------ 自定义异常 ------------------------


class AuthError(RuntimeError):
    """鉴权失败：凭证缺失或无效。"""


class NetworkError(RuntimeError):
    """网络或网关错误（含 429 限流），已经过内部重试。"""


class MissingCredentialsError(RuntimeError):
    """完全缺少可用凭证时抛出，由 Agent 层触发最小化询问。"""

    def __init__(self, message: str | None = None) -> None:
        super().__init__(message or MINIMAL_ASK_HINT)


class ProviderNotAvailableError(RuntimeError):
    """指定 Provider 不可用（如被标记为降级或缺少依赖）。"""


class SdkNotInstalledError(RuntimeError):
    """SDK 未安装或无法导入。"""


# ------------------------ 状态与凭证结构 ------------------------


@dataclass
class ProviderState:
    """Provider 运行状态，持久化到 state.json。

    字段说明：
    - provider: 上一次成功使用的 Provider 名称（"gateway" / "sdk"）。
    - degraded: 某个 Provider 是否处于降级状态（例如近期连续鉴权失败）。
    - sdk_installed: 是否已经在当前环境成功安装并导入过 SDK。
    - sdk_install_failed: SDK 安装是否已失败（失败后不再反复尝试）。
    - last_success_at: 最近一次成功调用的时间戳（秒）。
    - last_error: 最近一次错误的摘要信息（供调试使用，不含敏感信息）。
    """

    provider: Optional[str] = None
    degraded: Dict[str, bool] = field(
        default_factory=lambda: {"gateway": False, "sdk": False}
    )
    sdk_installed: bool = False
    sdk_install_failed: bool = False
    last_success_at: Optional[float] = None
    last_error: Optional[Dict[str, str]] = None


@dataclass
class AuthConfig:
    """统一的鉴权配置视图，综合环境变量与本地持久化 auth.json。

    gateway_* 字段用于 Gateway 调用；sdk_* 字段用于 SDK 调用。
    环境变量始终优先于本地 auth.json。
    """

    gateway_api_base: Optional[str] = None
    gateway_api_key: Optional[str] = None
    sdk_access_key_id: Optional[str] = None
    sdk_secret_access_key: Optional[str] = None
    sdk_region: str = "cn-beijing"


def _mask_value(value: str) -> str:
    """对环境变量值做简单脱敏处理：保留前 4 位和后 2 位，中间使用 ***。

    说明：对于长度较短的值，仍会进行脱敏，避免直接暴露完整内容。
    """
    v = value.strip()
    if not v:
        return "***"
    if len(v) <= 6:
        # 短值仅保留首尾一位
        if len(v) <= 2:
            return "***"
        return f"{v[0]}***{v[-1]}"
    return f"{v[:4]}***{v[-2:]}"


# 常见 rc 文件中允许解析的目标变量集合
RC_ENV_TARGET_VARS = {
    "ARK_SKILL_API_BASE",
    "ARK_SKILL_API_KEY",
    "VOLCSTACK_ACCESS_KEY_ID",
    "VOLCSTACK_SECRET_ACCESS_KEY",
    "VOLCSTACK_REGION",
}

# 缓存 rc 解析结果，避免多次重复读取
_RC_ENV_CACHE: Dict[str, str] = {}
_RC_ENV_SOURCE: Dict[str, str] = {}


def load_rc_env() -> Dict[str, str]:
    """从常见 rc 文件中解析 export 行，返回可用的环境变量值（不含脱敏）。

    注意：
    - 不执行 shell，不展开变量；
    - 仅解析形如 `export VAR=VALUE` 的简单单行；
    - VALUE 中若包含 `$`、`` ` ``、`$(` 等符号，则视为不可解析并跳过；
    - 仅收集 RC_ENV_TARGET_VARS 中定义的变量；
    - 若同一变量在多个文件中出现，只保留最先解析到的一份。
    """
    if _RC_ENV_CACHE:
        return dict(_RC_ENV_CACHE)

    home = Path.home()
    rc_files = [
        ("~/.bashrc", home / ".bashrc"),
        ("~/.bash_profile", home / ".bash_profile"),
        ("~/.zshrc", home / ".zshrc"),
        ("~/.profile", home / ".profile"),
    ]

    pattern = re.compile(r"^\s*export\s+([A-Za-z_][A-Za-z0-9_]*)=(.*)$")

    for display_name, path in rc_files:
        if not path.exists() or not path.is_file():
            continue

        try:
            content = path.read_text(encoding="utf-8", errors="ignore")
        except Exception:
            # 单个文件读取失败不影响整体结果
            continue

        for line in content.splitlines():
            match = pattern.match(line)
            if not match:
                continue

            var_name, raw_value = match.group(1), match.group(2).strip()
            if var_name not in RC_ENV_TARGET_VARS:
                continue
            if var_name in _RC_ENV_CACHE:
                # 已经解析过该变量，保持最先出现的定义
                continue
            if not raw_value:
                continue

            # 对包含明显 shell 展开/命令替换的复杂表达式不做解析
            if any(sym in raw_value for sym in ("$(", "`")):
                continue

            # 去掉成对包裹的引号
            if (
                (raw_value.startswith('"') and raw_value.endswith('"'))
                or (raw_value.startswith("'") and raw_value.endswith("'"))
            ):
                raw_value = raw_value[1:-1].strip()

            if not raw_value:
                continue

            _RC_ENV_CACHE[var_name] = raw_value
            _RC_ENV_SOURCE[var_name] = display_name

    return dict(_RC_ENV_CACHE)


def discover_env_candidates() -> list[dict]:
    """只读扫描常见 shell rc 文件中的环境变量定义（脱敏后用于提示）。

    仅在当前进程缺少 ARK_* 或 VOLCSTACK_* 中任意一组凭证时才有意义：
    - 扫描 ~/.bashrc、~/.bash_profile、~/.zshrc、~/.profile（若存在）；
    - 仅匹配形如 `export VAR=VALUE` 的简单单行；
    - 支持的变量包括 RC_ENV_TARGET_VARS 中的定义；
    - 返回示例结构：
      {"file": "~/.bashrc", "vars": {"ARK_SKILL_API_BASE": "xxxx***yy", ...}} 列表。
    """
    # 当前进程如果已经具备两组凭证，则无需扫描 rc 文件
    has_ark = bool(os.getenv("ARK_SKILL_API_BASE") and os.getenv("ARK_SKILL_API_KEY"))
    has_volc = bool(
        os.getenv("VOLCSTACK_ACCESS_KEY_ID")
        and os.getenv("VOLCSTACK_SECRET_ACCESS_KEY")
    )
    if has_ark and has_volc:
        return []

    rc_env = load_rc_env()
    if not rc_env:
        return []

    # 根据变量来源文件聚合，并对值做脱敏
    file_to_vars: Dict[str, Dict[str, str]] = {}
    for var_name, raw_value in rc_env.items():
        file_label = _RC_ENV_SOURCE.get(var_name, "")
        if not file_label:
            continue
        masked = _mask_value(raw_value)
        bucket = file_to_vars.setdefault(file_label, {})
        bucket[var_name] = masked

    results: list[dict] = []
    for file_label, vars_map in file_to_vars.items():
        if vars_map:
            results.append({"file": file_label, "vars": vars_map})

    return results


# ------------------------ 文件读写工具 ------------------------


def _safe_read_json(path: Path) -> Dict:
    if not path.exists():
        return {}
    try:
        text = path.read_text(encoding="utf-8")
        if not text.strip():
            return {}
        return json.loads(text)
    except Exception:
        # 损坏或不可解析时忽略，避免阻塞运行
        return {}


def _safe_write_json(path: Path, data: Dict) -> None:
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        # 避免 NaN/Infinity 等不可序列化值
        payload = json.dumps(data, ensure_ascii=False)
        path.write_text(payload, encoding="utf-8")
    except Exception:
        # 状态写入失败不应影响主流程
        return


def load_state() -> ProviderState:
    data = _safe_read_json(STATE_FILE)
    if not data:
        return ProviderState()

    try:
        return ProviderState(
            provider=data.get("provider"),
            degraded=data.get("degraded") or {"gateway": False, "sdk": False},
            sdk_installed=bool(data.get("sdk_installed")),
            sdk_install_failed=bool(data.get("sdk_install_failed")),
            last_success_at=data.get("last_success_at"),
            last_error=data.get("last_error"),
        )
    except Exception:
        return ProviderState()


def save_state(state: ProviderState) -> None:
    _safe_write_json(STATE_FILE, asdict(state))


def load_auth() -> AuthConfig:
    """合并环境变量、本地 auth.json 与 rc 自动填充，优先级为：
    os.environ > auth.json > rc 文件。
    """

    raw = _safe_read_json(AUTH_FILE)
    gateway_raw = raw.get("gateway") if isinstance(raw.get("gateway"), dict) else {}
    sdk_raw = raw.get("sdk") if isinstance(raw.get("sdk"), dict) else {}

    # 1) 环境变量优先，其次为 auth.json
    gateway_api_base = os.getenv("ARK_SKILL_API_BASE") or gateway_raw.get("api_base")
    gateway_api_key = os.getenv("ARK_SKILL_API_KEY") or gateway_raw.get("api_key")

    sdk_access_key_id = os.getenv("VOLCSTACK_ACCESS_KEY_ID") or sdk_raw.get(
        "access_key_id"
    )
    sdk_secret_access_key = os.getenv("VOLCSTACK_SECRET_ACCESS_KEY") or sdk_raw.get(
        "secret_access_key"
    )
    sdk_region_env = os.getenv("VOLCSTACK_REGION")
    sdk_region_auth = sdk_raw.get("region")
    sdk_region = sdk_region_env or sdk_region_auth or "cn-beijing"

    auth = AuthConfig(
        gateway_api_base=gateway_api_base,
        gateway_api_key=gateway_api_key,
        sdk_access_key_id=sdk_access_key_id,
        sdk_secret_access_key=sdk_secret_access_key,
        sdk_region=sdk_region,
    )

    # 2) 若某一 Provider 的成对变量缺失，则尝试从 rc 文件自动填充（最低优先级）
    need_gateway = not (auth.gateway_api_base and auth.gateway_api_key)
    need_sdk = not (auth.sdk_access_key_id and auth.sdk_secret_access_key)

    if need_gateway or need_sdk:
        rc_env = load_rc_env()

        # Gateway: 需要同时存在 BASE 与 KEY
        if need_gateway:
            rc_base = rc_env.get("ARK_SKILL_API_BASE")
            rc_key = rc_env.get("ARK_SKILL_API_KEY")
            if rc_base and rc_key:
                auth.gateway_api_base = rc_base
                auth.gateway_api_key = rc_key
                file_label = (
                    _RC_ENV_SOURCE.get("ARK_SKILL_API_BASE")
                    or _RC_ENV_SOURCE.get("ARK_SKILL_API_KEY")
                    or "rc 文件"
                )
                masked_base = _mask_value(rc_base)
                masked_key = _mask_value(rc_key)
                print(
                    "[INFO] 从 {file} 检测到 Gateway 凭证定义（已自动采用，未持久化）："
                    "ARK_SKILL_API_BASE={base}, ARK_SKILL_API_KEY={key}".format(
                        file=file_label,
                        base=masked_base,
                        key=masked_key,
                    )
                )

        # SDK: 需要同时存在 AK 与 SK，可选 region
        if need_sdk:
            rc_ak = rc_env.get("VOLCSTACK_ACCESS_KEY_ID")
            rc_sk = rc_env.get("VOLCSTACK_SECRET_ACCESS_KEY")
            if rc_ak and rc_sk:
                auth.sdk_access_key_id = rc_ak
                auth.sdk_secret_access_key = rc_sk
                # 仅当环境变量/持久化中未显式配置 region 时，才使用 rc 中的 region
                rc_region = rc_env.get("VOLCSTACK_REGION")
                if rc_region and not (sdk_region_env or sdk_region_auth):
                    auth.sdk_region = rc_region

                file_label = (
                    _RC_ENV_SOURCE.get("VOLCSTACK_ACCESS_KEY_ID")
                    or _RC_ENV_SOURCE.get("VOLCSTACK_SECRET_ACCESS_KEY")
                    or "rc 文件"
                )
                masked_ak = _mask_value(rc_ak)
                masked_sk = _mask_value(rc_sk)
                print(
                    "[INFO] 从 {file} 检测到 SDK 凭证定义（已自动采用，未持久化）："
                    "VOLCSTACK_ACCESS_KEY_ID={ak}, VOLCSTACK_SECRET_ACCESS_KEY={sk}".format(
                        file=file_label,
                        ak=masked_ak,
                        sk=masked_sk,
                    )
                )

    return auth


def save_auth_gateway(api_base: str, api_key: str) -> None:
    raw = _safe_read_json(AUTH_FILE)
    raw["gateway"] = {
        "api_base": api_base,
        "api_key": api_key,
    }
    _safe_write_json(AUTH_FILE, raw)


def save_auth_sdk(access_key_id: str, secret_access_key: str, region: str) -> None:
    raw = _safe_read_json(AUTH_FILE)
    raw["sdk"] = {
        "access_key_id": access_key_id,
        "secret_access_key": secret_access_key,
        "region": region,
    }
    _safe_write_json(AUTH_FILE, raw)


# ------------------------ SDK 安装与检测 ------------------------


def _try_import_sdk() -> bool:
    """尝试导入 SDK 相关模块。成功返回 True，失败返回 False。"""

    try:
        import importlib

        importlib.import_module("volcenginesdkinsight")
        importlib.import_module("volcenginesdkcore")
        return True
    except Exception:
        return False


def ensure_sdk_available(state: Optional[ProviderState] = None) -> bool:
    """确保 SDK 可用：

    - 若已标记安装失败，直接返回 False；
    - 若已安装且可导入，返回 True；
    - 否则尝试通过 pip 安装一次 `volcengine-python-sdk>=5.0.22`；
    - 安装成功则返回 True，失败时标记 sdk_install_failed，返回 False。
    """

    st = state or load_state()

    if st.sdk_install_failed:
        return False

    if _try_import_sdk():
        if not st.sdk_installed:
            st.sdk_installed = True
            save_state(st)
        return True

    # 尝试自动安装
    auto_pip_flag = os.getenv("MARKET_INSIGHT_AUTO_PIP", "1").lower()
    allow_auto_pip = auto_pip_flag in {"1", "true", "yes", "y"}

    if not allow_auto_pip:
        return False

    print(
        "[INFO] 检测到 SDK 未安装，正在自动安装 "
        f"{SDK_MIN_VERSION_SPEC}（仅在缺少时执行一次）..."
    )

    cmd = [sys.executable, "-m", "pip", "install", SDK_MIN_VERSION_SPEC]

    try:
        proc = subprocess.run(
            cmd,
            check=False,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )
    except Exception as exc:  # noqa: BLE001
        print(f"[WARN] 自动安装 SDK 失败: {exc}")
        st.sdk_install_failed = True
        save_state(st)
        return False

    if proc.returncode != 0:
        print("[WARN] 自动安装 SDK 失败，后续将不再自动重试。")
        st.sdk_install_failed = True
        save_state(st)
        return False

    if _try_import_sdk():
        st.sdk_installed = True
        st.sdk_install_failed = False
        save_state(st)
        print("[INFO] SDK 安装并导入成功。")
        return True

    print("[WARN] SDK 安装后仍无法导入，将标记为安装失败。")
    st.sdk_install_failed = True
    save_state(st)
    return False


# ------------------------ Provider 封装 ------------------------


class BaseProvider:
    """统一的 Provider 抽象，具体实现由 Gateway / SDK 封装。"""

    name: str

    def list_custom_subs_task(self, **params):  # type: ignore[no-untyped-def]
        raise NotImplementedError

    def pull_post(self, **params):  # type: ignore[no-untyped-def]
        raise NotImplementedError

    def query_clue_info(self, **params):  # type: ignore[no-untyped-def]
        raise NotImplementedError


class GatewayProviderWrapper(BaseProvider):
    def __init__(self, api_base: str, api_key: str) -> None:
        self.name = "gateway"
        # 避免重复斜杠
        self.api_base = api_base.rstrip("/")
        self.api_key = api_key

        # 延迟导入，避免初始化阶段的循环依赖
        from providers import gateway_provider  # type: ignore[import]

        self._impl = gateway_provider

    def list_custom_subs_task(self, **params):  # type: ignore[no-untyped-def]
        return self._impl.list_custom_subs_task(
            api_base=self.api_base,
            api_key=self.api_key,
            **params,
        )

    def pull_post(self, **params):  # type: ignore[no-untyped-def]
        return self._impl.pull_post(
            api_base=self.api_base,
            api_key=self.api_key,
            **params,
        )

    def query_clue_info(self, **params):  # type: ignore[no-untyped-def]
        return self._impl.query_clue_info(
            api_base=self.api_base,
            api_key=self.api_key,
            **params,
        )


class SdkProviderWrapper(BaseProvider):
    def __init__(self, access_key_id: str, secret_access_key: str, region: str) -> None:
        self.name = "sdk"
        self.access_key_id = access_key_id
        self.secret_access_key = secret_access_key
        self.region = region

        if not ensure_sdk_available():
            raise SdkNotInstalledError(
                "volcengine-python-sdk 未安装或无法导入，且自动安装失败。"
            )

        from providers import sdk_provider  # type: ignore[import]

        self._impl = sdk_provider

    def list_custom_subs_task(self, **params):  # type: ignore[no-untyped-def]
        return self._impl.list_custom_subs_task(
            access_key_id=self.access_key_id,
            secret_access_key=self.secret_access_key,
            region=self.region,
            **params,
        )

    def pull_post(self, **params):  # type: ignore[no-untyped-def]
        return self._impl.pull_post(
            access_key_id=self.access_key_id,
            secret_access_key=self.secret_access_key,
            region=self.region,
            **params,
        )

    def query_clue_info(self, **params):  # type: ignore[no-untyped-def]
        return self._impl.query_clue_info(
            access_key_id=self.access_key_id,
            secret_access_key=self.secret_access_key,
            region=self.region,
            **params,
        )


# ------------------------ Provider 选择逻辑 ------------------------


def _pick_sticky_provider(
    state: ProviderState,
    auth: AuthConfig,
    exclude: Set[str],
) -> Optional[BaseProvider]:
    """若存在上一次成功 Provider 且未被排除，则优先尝试。"""

    name = state.provider
    if not name or name in exclude:
        return None

    if name == "gateway":
        if (
            auth.gateway_api_base
            and auth.gateway_api_key
            and not state.degraded.get("gateway", False)
        ):
            return GatewayProviderWrapper(auth.gateway_api_base, auth.gateway_api_key)

    if name == "sdk":
        if (
            auth.sdk_access_key_id
            and auth.sdk_secret_access_key
            and not state.degraded.get("sdk", False)
        ):
            return SdkProviderWrapper(
                auth.sdk_access_key_id,
                auth.sdk_secret_access_key,
                auth.sdk_region,
            )

    return None


def _pick_fresh_provider(
    state: ProviderState,
    auth: AuthConfig,
    exclude: Set[str],
) -> Optional[BaseProvider]:
    """按优先级选择 Provider：Gateway → SDK。"""

    # 默认优先 Gateway
    if (
        "gateway" not in exclude
        and auth.gateway_api_base
        and auth.gateway_api_key
        and not state.degraded.get("gateway", False)
    ):
        return GatewayProviderWrapper(auth.gateway_api_base, auth.gateway_api_key)

    # 其次 SDK
    if (
        "sdk" not in exclude
        and auth.sdk_access_key_id
        and auth.sdk_secret_access_key
        and not state.degraded.get("sdk", False)
    ):
        return SdkProviderWrapper(
            auth.sdk_access_key_id,
            auth.sdk_secret_access_key,
            auth.sdk_region,
        )

    return None


def get_provider(exclude: Optional[Set[str]] = None) -> BaseProvider:
    """决策当前应使用的 Provider。

    优先级：
    1. 若 state.provider 存在且未被排除，且对应凭证与状态可用，则优先使用（会话级粘性）；
    2. 否则按默认顺序尝试：Gateway → SDK；
    3. 若两者均不可用，则抛出 MissingCredentialsError，交由 Agent 层进行最小化询问。
    """

    exclude = exclude or set()
    state = load_state()
    auth = load_auth()

    # 先尝试粘性 Provider
    try:
        sticky = _pick_sticky_provider(state, auth, exclude)
        if sticky is not None:
            print(f"[INFO] 使用会话粘性 Provider: {sticky.name}")
            return sticky
    except (SdkNotInstalledError, ProviderNotAvailableError):
        # 如果粘性 Provider 初始化失败，继续尝试后续 Provider
        pass

    # 再按默认顺序选择
    try:
        fresh = _pick_fresh_provider(state, auth, exclude)
        if fresh is not None:
            print(f"[INFO] 使用 Provider: {fresh.name}")
            return fresh
    except SdkNotInstalledError:
        # SDK 初始化失败时，不再尝试 SDK
        state.degraded["sdk"] = True
        save_state(state)

    # 均不可用：缺少凭证或已全部降级
    candidates = discover_env_candidates()
    if candidates:
        # 在异常信息中追加只读环境变量位置提示，帮助用户排查 rc 文件
        lines = [
            "另：在以下文件中发现可能的环境变量定义，但当前进程未生效："
        ]
        for item in candidates:
            file_label = item.get("file") or ""
            vars_map = item.get("vars") or {}
            if not vars_map:
                continue
            pairs = ", ".join(f"{k}={v}" for k, v in vars_map.items())
            lines.append(f"- {file_label}: {pairs}")
        lines.append("请确认是否 export 以及启动环境是否 source 相应 rc 文件。")
        extra_msg = "\n".join(lines)
        raise MissingCredentialsError(f"{MINIMAL_ASK_HINT}\n\n{extra_msg}")
    raise MissingCredentialsError()


# ------------------------ 状态更新工具 ------------------------


def record_success(provider_name: str) -> None:
    """记录 Provider 调用成功，更新粘性状态。"""

    state = load_state()
    state.provider = provider_name
    state.degraded[provider_name] = False
    state.last_success_at = time.time()
    state.last_error = None
    save_state(state)


def mark_degraded(provider_name: str, error: Exception) -> None:
    """将指定 Provider 标记为降级，并记录错误摘要。"""

    state = load_state()
    state.degraded[provider_name] = True
    state.last_error = {
        "provider": provider_name,
        "type": error.__class__.__name__,
        "message": str(error)[:200],  # 避免异常文本过长
    }
    save_state(state)


__all__ = [
    "AuthError",
    "NetworkError",
    "MissingCredentialsError",
    "ProviderNotAvailableError",
    "SdkNotInstalledError",
    "ProviderState",
    "AuthConfig",
    "MINIMAL_ASK_HINT",
    "load_rc_env",
    "discover_env_candidates",
    "load_state",
    "save_state",
    "load_auth",
    "save_auth_gateway",
    "save_auth_sdk",
    "ensure_sdk_available",
    "BaseProvider",
    "GatewayProviderWrapper",
    "SdkProviderWrapper",
    "get_provider",
    "record_success",
    "mark_degraded",
]

#!/usr/bin/env python3
# Copyright 2024 ByteDance, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""市场洞察 Agent 统一客户端入口。

对外暴露三个函数：
- list_custom_subs_task
- pull_post
- query_clue_info

内部通过 auth_resolver 自动选择 Gateway 或 SDK Provider：
- 默认优先 Gateway，按会话状态实现粘性；
- 在需要 SDK 时自动安装依赖；
- 统一返回结构；
- 提供 CLI 入口便于本地调试。
"""

from __future__ import annotations

import argparse
import json
import sys
from typing import Any, Dict, Optional, Set

from auth_resolver import (
    AuthError,
    NetworkError,
    MissingCredentialsError,
    MINIMAL_ASK_HINT,
    get_provider,
    mark_degraded,
    record_success,
)


# ------------------------ 内部调度封装 ------------------------


def _invoke_with_fallback(method_name: str, **params: Any) -> Dict[str, Any]:
    """统一的调度封装：

    - 首次按照会话粘性 + 默认优先级选择 Provider；
    - 若发生 AuthError，则将当前 Provider 标记为降级并尝试另一条链路；
    - 若所有 Provider 均不可用，则抛出 MissingCredentialsError，交由 Agent 层进行最小化询问；
    - NetworkError 直接向上抛出（内部已做退避重试）。
    """

    exclude: Set[str] = set()
    last_error: Optional[Exception] = None

    # 最多尝试两条 Provider 链路（gateway + sdk）
    for _ in range(2):
        provider = get_provider(exclude)
        try:
            func = getattr(provider, method_name)
        except AttributeError as exc:  # noqa: BLE001
            raise RuntimeError(
                f"Provider {provider.name} 不支持方法 {method_name}"
            ) from exc

        try:
            result = func(**params)
            record_success(provider.name)
            return result
        except AuthError as exc:
            # 鉴权错误：标记当前 Provider 降级，尝试其他链路
            mark_degraded(provider.name, exc)
            exclude.add(provider.name)
            last_error = exc
        except NetworkError as exc:
            # 网络错误：不切换链路，直接抛出
            last_error = exc
            raise

    # 所有 Provider 均不可用：触发最小化询问
    if isinstance(last_error, AuthError) or last_error is None:
        raise MissingCredentialsError()

    # 兜底：返回最后一个错误
    raise last_error


# ------------------------ 对外函数（Python 调用） ------------------------


def list_custom_subs_task(
    *,
    Status: int = 2,
    TaskName: Optional[str] = None,
    PageNum: int = 1,
    PageSize: int = 30,
) -> Dict[str, Any]:
    """查询订阅/监控任务列表（ListCustomSubsTask）。

    返回结构：{"InsightSaasTaskList": [...], "Total": int}
    """

    return _invoke_with_fallback(
        "list_custom_subs_task",
        Status=Status,
        TaskName=TaskName,
        PageNum=PageNum,
        PageSize=PageSize,
    )


def pull_post(
    *,
    TaskID: int,
    StartTime: str,
    EndTime: str,
    Size: int = 50,
    PageToken: Optional[str] = None,
) -> Dict[str, Any]:
    """拉取监测任务的 AI 精筛数据（PullPost）。

    返回结构：{"ItemDocs": [...], "HasMore": bool, "NextPageToken": str | None}
    """

    return _invoke_with_fallback(
        "pull_post",
        TaskID=TaskID,
        StartTime=StartTime,
        EndTime=EndTime,
        Size=Size,
        PageToken=PageToken,
    )


def query_clue_info(
    *,
    StartTime: str,
    EndTime: str,
    MaxResults: int = 10,
    NextToken: Optional[str] = None,
) -> Dict[str, Any]:
    """查询商机信息（QueryClueInfo）。

    返回结构：{"ClueList": [...], "NextToken": str | None, "ResultCnt": int}
    其中 ClueText 字段若存在，将被解析为结构化对象。
    """

    return _invoke_with_fallback(
        "query_clue_info",
        StartTime=StartTime,
        EndTime=EndTime,
        MaxResults=MaxResults,
        NextToken=NextToken,
    )


# ------------------------ CLI 入口 ------------------------


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="市场洞察 Agent 统一客户端（Gateway + SDK 无感切换）",
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    # list_custom_subs_task
    p_list = subparsers.add_parser(
        "list_custom_subs_task",
        help="查询订阅/监控任务列表 (ListCustomSubsTask)",
    )
    p_list.add_argument("--Status", type=int, default=2, choices=[1, 2], help="任务状态过滤：1=运行中, 2=全部（默认 2）")
    p_list.add_argument("--TaskName", type=str, default=None, help="按任务名称模糊搜索（可选）")
    p_list.add_argument("--PageNum", type=int, default=1, help="页码，从 1 开始（默认 1）")
    p_list.add_argument("--PageSize", type=int, default=30, help="每页条数（默认 30）")

    # pull_post
    p_pull = subparsers.add_parser(
        "pull_post",
        help="拉取监测任务 AI 精筛数据 (PullPost)",
    )
    p_pull.add_argument("--TaskID", type=int, required=True, help="监测任务 ID")
    p_pull.add_argument("--StartTime", type=str, required=True, help='数据起始时间，格式 "YYYY-MM-DD HH:MM:SS"')
    p_pull.add_argument("--EndTime", type=str, required=True, help='数据结束时间，格式 "YYYY-MM-DD HH:MM:SS"')
    p_pull.add_argument("--Size", type=int, default=50, help="每页条数（默认 50）")
    p_pull.add_argument("--PageToken", type=str, default=None, help="分页游标（可选）")

    # query_clue_info
    p_clue = subparsers.add_parser(
        "query_clue_info",
        help="查询商机信息 (QueryClueInfo)",
    )
    p_clue.add_argument("--StartTime", type=str, required=True, help='数据起始时间，格式 "YYYY-MM-DD HH:MM:SS"')
    p_clue.add_argument("--EndTime", type=str, required=True, help='数据结束时间，格式 "YYYY-MM-DD HH:MM:SS"')
    p_clue.add_argument("--MaxResults", type=int, default=10, help="每页最大返回条数（默认 10）")
    p_clue.add_argument("--NextToken", type=str, default=None, help="分页游标（可选）")

    return parser


def _run_cli(argv: Optional[list[str]] = None) -> int:
    parser = _build_parser()
    args = parser.parse_args(argv)

    try:
        if args.command == "list_custom_subs_task":
            result = list_custom_subs_task(
                Status=args.Status,
                TaskName=args.TaskName,
                PageNum=args.PageNum,
                PageSize=args.PageSize,
            )
        elif args.command == "pull_post":
            result = pull_post(
                TaskID=args.TaskID,
                StartTime=args.StartTime,
                EndTime=args.EndTime,
                Size=args.Size,
                PageToken=args.PageToken,
            )
        elif args.command == "query_clue_info":
            result = query_clue_info(
                StartTime=args.StartTime,
                EndTime=args.EndTime,
                MaxResults=args.MaxResults,
                NextToken=args.NextToken,
            )
        else:  # 理论上不会触达
            parser.print_help()
            return 1

        print(json.dumps(result, ensure_ascii=False, indent=2))
        return 0

    except MissingCredentialsError as exc:
        # 提示 Agent 层应进行最小化询问
        print(str(exc or MINIMAL_ASK_HINT), file=sys.stderr)
        return 2
    except AuthError as exc:
        print(f"鉴权错误：{exc}", file=sys.stderr)
        return 3
    except NetworkError as exc:
        print(f"网络/网关错误：{exc}", file=sys.stderr)
        return 4
    except KeyboardInterrupt:
        print("已中断", file=sys.stderr)
        return 130


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(_run_cli())

#!/usr/bin/env python3
# Copyright 2024 ByteDance, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Gateway Provider 实现。

使用 urllib 通过 API Gateway 调用市场洞察三大接口：
- ListCustomSubsTask（GET + JSON Body）
- PullPost（POST + JSON Body）
- QueryClueInfo（GET + JSON Body）

本模块：
- 仅依赖标准库；
- 统一解析网关返回的 Result；
- 对 QueryClueInfo 的 ClueText 做 json.loads 解析；
- 对网络错误和 429 做退避重试；
- 不打印密钥或完整网关地址，仅输出非敏感 Debug 摘要。
"""

from __future__ import annotations

import json
import time
import urllib.error
import urllib.request
from typing import Any, Dict, Optional

from auth_resolver import AuthError, NetworkError

API_VERSION = "2025-09-05"
DEFAULT_TIMEOUT = 30
MAX_RETRIES = 3
BACKOFF_BASE_SECONDS = 0.5


def _sanitize_host(url: str) -> str:
    """去掉 scheme 和查询参数，只保留主机部分并做简单截断。"""

    # 形如 https://host/path?query
    without_scheme = url.split("://", 1)[-1]
    host = without_scheme.split("/", 1)[0]
    if len(host) > 40:
        return host[:37] + "..."
    return host


def _debug_request_summary(action: str, method: str, url: str, payload: Dict[str, Any]) -> None:
    host = _sanitize_host(url)
    page_num = payload.get("PageNum")
    page_size = payload.get("PageSize") or payload.get("Size") or payload.get("MaxResults")
    print(
        f"[DEBUG][gateway] Action={action} method={method} host={host} "
        f"PageNum={page_num} PageSize/Size/MaxResults={page_size}"
    )


def _debug_response_summary(action: str, body: bytes) -> None:
    try:
        text = body.decode("utf-8", errors="replace")
    except Exception:  # noqa: BLE001
        return
    snippet = text[:500]
    print(
        f"[DEBUG][gateway] Action={action} 响应体前 500 字符:" f" {snippet!r}"
    )


def _build_url(api_base: str, action: str) -> str:
    return f"{api_base.rstrip('/')}/?Action={action}&Version={API_VERSION}"


def _do_request(
    *,
    method: str,
    url: str,
    api_key: str,
    payload: Dict[str, Any],
    action: str,
) -> Dict[str, Any]:
    """执行单次 HTTP 请求并解析 JSON 响应，包含重试与错误分类。"""

    data = json.dumps(payload).encode("utf-8")
    headers = {
        "Content-Type": "application/json; charset=UTF-8",
        "Authorization": f"Bearer {api_key}",
        "ServiceName": "insight",
    }

    for attempt in range(MAX_RETRIES):
        try:
            _debug_request_summary(action, method, url, payload)

            if method == "GET":
                req = urllib.request.Request(url, data=data, headers=headers)
                # 有 body 时 urllib 默认 POST，这里强制改为 GET
                req.get_method = lambda: "GET"  # type: ignore[assignment]
            else:
                req = urllib.request.Request(
                    url, data=data, headers=headers, method="POST"
                )

            with urllib.request.urlopen(req, timeout=DEFAULT_TIMEOUT) as resp:
                body = resp.read()
                _debug_response_summary(action, body)
                try:
                    parsed = json.loads(body.decode("utf-8"))
                except Exception as exc:  # noqa: BLE001
                    raise NetworkError(
                        f"Gateway 响应不是合法 JSON（Action={action}）: {exc}"
                    ) from exc

                if not isinstance(parsed, dict):
                    raise NetworkError(
                        f"Gateway 响应 JSON 顶层不是对象（Action={action}）"
                    )
                return parsed

        except urllib.error.HTTPError as e:  # HTTP 层错误
            status = e.code
            # 读出错误体但不打印，以避免泄漏敏感信息
            try:
                _ = e.read()
            except Exception:  # noqa: BLE001
                _ = b""

            if status in (401, 403):
                raise AuthError(f"Gateway 鉴权失败（HTTP {status}）") from e

            if status == 429 or 500 <= status < 600:
                # 限流或服务器错误：退避重试
                if attempt < MAX_RETRIES - 1:
                    delay = BACKOFF_BASE_SECONDS * (2 ** attempt)
                    print(
                        f"[WARN][gateway] Action={action} HTTP {status}，" f"{delay:.2f}s 后重试..."
                    )
                    time.sleep(delay)
                    continue
                raise NetworkError(
                    f"Gateway 请求失败（HTTP {status}，已重试 {MAX_RETRIES} 次）"
                ) from e

            # 其它 HTTP 视为一次性错误，不再重试
            raise NetworkError(f"Gateway 请求异常（HTTP {status}）") from e

        except urllib.error.URLError as e:
            # 网络错误（DNS/连接超时等）
            if attempt < MAX_RETRIES - 1:
                delay = BACKOFF_BASE_SECONDS * (2 ** attempt)
                print(
                    f"[WARN][gateway] Action={action} 网络错误 {e.reason}，" f"{delay:.2f}s 后重试..."
                )
                time.sleep(delay)
                continue
            raise NetworkError(
                f"Gateway 网络错误（Action={action}）：{e.reason}"
            ) from e

    # 理论上不会到达这里
    raise NetworkError(f"Gateway 请求失败（Action={action}）：重试耗尽")


# ------------------------ 统一封装的三个接口 ------------------------


def list_custom_subs_task(
    *,
    api_base: str,
    api_key: str,
    Status: int = 2,
    TaskName: Optional[str] = None,
    PageNum: int = 1,
    PageSize: int = 30,
) -> Dict[str, Any]:
    """调用 ListCustomSubsTask，返回统一结构：

    {"InsightSaasTaskList": [...], "Total": int}
    """

    url = _build_url(api_base, "ListCustomSubsTask")

    payload: Dict[str, Any] = {
        "Status": Status,
        "PageNum": PageNum,
        "PageSize": PageSize,
    }
    if TaskName:
        payload["TaskName"] = TaskName

    raw = _do_request(method="GET", url=url, api_key=api_key, payload=payload, action="ListCustomSubsTask")

    result = raw.get("Result") or {}
    if not isinstance(result, dict):
        raise NetworkError("ListCustomSubsTask 响应缺少 Result 字段或类型错误")

    tasks = result.get("InsightSaasTaskList") or []
    total_raw = result.get("Total", 0)
    try:
        total = int(total_raw)
    except Exception:  # noqa: BLE001
        total = 0

    # 业务为空：返回空列表和 0 即可
    return {
        "InsightSaasTaskList": tasks,
        "Total": total,
    }


def pull_post(
    *,
    api_base: str,
    api_key: str,
    TaskID: int,
    StartTime: str,
    EndTime: str,
    Size: int = 50,
    PageToken: Optional[str] = None,
) -> Dict[str, Any]:
    """调用 PullPost，返回统一结构：

    {"ItemDocs": [...], "HasMore": bool, "NextPageToken": str | None}
    """

    url = _build_url(api_base, "PullPost")

    payload: Dict[str, Any] = {
        "TaskID": TaskID,
        "StartTime": StartTime,
        "EndTime": EndTime,
        "Size": Size,
    }
    if PageToken:
        payload["PageToken"] = PageToken

    raw = _do_request(method="POST", url=url, api_key=api_key, payload=payload, action="PullPost")

    result = raw.get("Result") or {}
    if not isinstance(result, dict):
        raise NetworkError("PullPost 响应缺少 Result 字段或类型错误")

    docs = result.get("ItemDocs") or []
    has_more = bool(result.get("HasMore", False))
    next_token = result.get("NextPageToken")

    return {
        "ItemDocs": docs,
        "HasMore": has_more,
        "NextPageToken": next_token,
    }


def query_clue_info(
    *,
    api_base: str,
    api_key: str,
    StartTime: str,
    EndTime: str,
    MaxResults: int = 10,
    NextToken: Optional[str] = None,
) -> Dict[str, Any]:
    """调用 QueryClueInfo，返回统一结构：

    {"ClueList": [...], "NextToken": str | None, "ResultCnt": int}

    其中每个元素的 ClueText 字段若为 JSON 字符串，将被解析为对象。
    """

    url = _build_url(api_base, "QueryClueInfo")

    payload: Dict[str, Any] = {
        "StartTime": StartTime,
        "EndTime": EndTime,
        "MaxResults": MaxResults,
    }
    if NextToken:
        payload["NextToken"] = NextToken

    raw = _do_request(method="GET", url=url, api_key=api_key, payload=payload, action="QueryClueInfo")

    result = raw.get("Result") or {}
    if not isinstance(result, dict):
        raise NetworkError("QueryClueInfo 响应缺少 Result 字段或类型错误")

    clues = result.get("ClueList") or []
    parsed_clues = []
    for item in clues:
        if isinstance(item, dict):
            parsed = dict(item)
        else:
            # 非 dict 的异常数据直接跳过
            continue

        raw_text = parsed.get("ClueText")
        if isinstance(raw_text, str) and raw_text:
            try:
                parsed["ClueText"] = json.loads(raw_text)
            except Exception:  # noqa: BLE001
                # 保留原始字符串，避免因单条解析失败影响整体
                pass

        parsed_clues.append(parsed)

    result_cnt_raw = result.get("ResultCnt")
    try:
        result_cnt = int(result_cnt_raw) if result_cnt_raw is not None else len(parsed_clues)
    except Exception:  # noqa: BLE001
        result_cnt = len(parsed_clues)

    next_token = result.get("NextToken")

    return {
        "ClueList": parsed_clues,
        "NextToken": next_token,
        "ResultCnt": result_cnt,
    }


__all__ = [
    "list_custom_subs_task",
    "pull_post",
    "query_clue_info",
]

#!/usr/bin/env python3
# Copyright 2024 ByteDance, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""SDK Provider 实现。

通过官方 `volcengine-python-sdk` 调用市场洞察三大接口：
- ListCustomSubsTask
- PullPost
- QueryClueInfo

本模块：
- 依赖 `volcengine-python-sdk>=5.0.22`；
- 不负责自动安装 SDK（由 auth_resolver.ensure_sdk_available 统一处理）；
- 将统一 PascalCase 参数映射到 SDK 的 snake_case 请求体；
- 返回与 Gateway Provider 一致的顶层结构；
- 对 QueryClueInfo 的 ClueText 做 json.loads 解析，并提供 `ClueText` 字段。
"""

from __future__ import annotations

from typing import Any, Dict, Optional

from auth_resolver import AuthError, NetworkError


def _init_sdk_config(access_key_id: str, secret_access_key: str, region: str) -> None:
    """初始化 SDK 全局配置（每次调用按给定 AK/SK/Region 覆盖）。"""

    try:
        import volcenginesdkcore  # type: ignore[import]
    except ImportError as exc:  # noqa: BLE001
        raise NetworkError(f"导入 volcenginesdkcore 失败: {exc}") from exc

    configuration = volcenginesdkcore.Configuration()
    configuration.ak = access_key_id
    configuration.sk = secret_access_key
    configuration.region = region

    # 重试与超时配置，与便携版示例保持一致
    configuration.auto_retry = True
    configuration.num_max_retries = 5
    configuration.retry_error_codes = {"Throttling", "RequestLimitExceeded"}
    configuration.connect_timeout = 10
    configuration.read_timeout = 30

    volcenginesdkcore.Configuration.set_default(configuration)


def _get_insight_api():
    try:
        from volcenginesdkinsight import INSIGHTApi  # type: ignore[import]
    except ImportError as exc:  # noqa: BLE001
        raise NetworkError(f"导入 volcenginesdkinsight 失败: {exc}") from exc

    return INSIGHTApi()


def _model_to_plain(value: Any) -> Any:
    """将 SDK 返回的模型对象递归转换为基础类型（dict/list/标量）。"""

    if isinstance(value, list):
        return [_model_to_plain(v) for v in value]
    if isinstance(value, dict):
        return {k: _model_to_plain(v) for k, v in value.items()}

    # 优先使用 SDK 模型自带的 to_dict
    if hasattr(value, "to_dict") and callable(getattr(value, "to_dict")):
        try:
            return _model_to_plain(value.to_dict())
        except Exception:  # noqa: BLE001
            return str(value)

    # 回退到 __dict__
    if hasattr(value, "__dict__") and not isinstance(value, type):
        try:
            return _model_to_plain(vars(value))
        except Exception:  # noqa: BLE001
            return str(value)

    return value


def _handle_api_exception(action: str, exc: Exception) -> None:
    """统一将 SDK 异常转换为 AuthError 或 NetworkError。"""

    try:
        from volcenginesdkcore.rest import ApiException  # type: ignore[import]
    except Exception:  # noqa: BLE001
        # 未能导入 ApiException，则一律视为网络错误
        raise NetworkError(f"{action} 调用异常: {exc}") from exc

    if isinstance(exc, ApiException):
        status = getattr(exc, "status", None)
        if status in (401, 403):
            raise AuthError(f"SDK {action} 鉴权失败（HTTP {status}）") from exc
        if status == 429 or (isinstance(status, int) and status >= 500):
            raise NetworkError(f"SDK {action} 请求失败（HTTP {status}）") from exc
        raise NetworkError(f"SDK {action} 调用失败: {exc}") from exc

    raise NetworkError(f"SDK {action} 调用异常: {exc}") from exc


# ------------------------ 统一封装的三个接口 ------------------------


def list_custom_subs_task(
    *,
    access_key_id: str,
    secret_access_key: str,
    region: str,
    Status: int = 2,
    TaskName: Optional[str] = None,
    PageNum: int = 1,
    PageSize: int = 30,
) -> Dict[str, Any]:
    """通过 SDK 调用 ListCustomSubsTask，返回统一结构：

    {"InsightSaasTaskList": [...], "Total": int}
    """

    _init_sdk_config(access_key_id, secret_access_key, region)

    try:
        from volcenginesdkinsight import (  # type: ignore[import]
            INSIGHTApi,
            ListCustomSubsTaskRequest,
        )
    except ImportError as exc:  # noqa: BLE001
        raise NetworkError(f"导入 volcenginesdkinsight 失败: {exc}") from exc

    api = INSIGHTApi()

    # SDK 使用 snake_case 参数
    request = ListCustomSubsTaskRequest(
        status=Status,
        page_num=PageNum,
        page_size=PageSize,
    )
    if TaskName:
        request.task_name = TaskName

    try:
        response = api.list_custom_subs_task(request)
    except Exception as exc:  # noqa: BLE001
        _handle_api_exception("ListCustomSubsTask", exc)
        raise  # for type checker

    tasks = getattr(response, "insight_saas_task_list", None) or []
    total_raw = getattr(response, "total", 0)
    try:
        total = int(total_raw)
    except Exception:  # noqa: BLE001
        total = 0

    tasks_plain = _model_to_plain(tasks)

    return {
        "InsightSaasTaskList": tasks_plain,
        "Total": total,
    }


def pull_post(
    *,
    access_key_id: str,
    secret_access_key: str,
    region: str,
    TaskID: int,
    StartTime: str,
    EndTime: str,
    Size: int = 50,
    PageToken: Optional[str] = None,
) -> Dict[str, Any]:
    """通过 SDK 调用 PullPost，返回统一结构：

    {"ItemDocs": [...], "HasMore": bool, "NextPageToken": str | None}
    """

    _init_sdk_config(access_key_id, secret_access_key, region)

    try:
        from volcenginesdkinsight import (  # type: ignore[import]
            INSIGHTApi,
            PullPostRequest,
        )
    except ImportError as exc:  # noqa: BLE001
        raise NetworkError(f"导入 volcenginesdkinsight 失败: {exc}") from exc

    api = INSIGHTApi()

    request = PullPostRequest(
        task_id=TaskID,
        start_time=StartTime,
        end_time=EndTime,
        size=Size,
    )
    if PageToken:
        request.page_token = PageToken

    try:
        response = api.pull_post(request)
    except Exception as exc:  # noqa: BLE001
        _handle_api_exception("PullPost", exc)
        raise

    docs = getattr(response, "item_docs", None) or []
    has_more = bool(getattr(response, "has_more", False))
    next_page_token = getattr(response, "next_page_token", None)

    docs_plain = _model_to_plain(docs)

    return {
        "ItemDocs": docs_plain,
        "HasMore": has_more,
        "NextPageToken": next_page_token,
    }


def query_clue_info(
    *,
    access_key_id: str,
    secret_access_key: str,
    region: str,
    StartTime: str,
    EndTime: str,
    MaxResults: int = 10,
    NextToken: Optional[str] = None,
) -> Dict[str, Any]:
    """通过 SDK 调用 QueryClueInfo，返回统一结构：

    {"ClueList": [...], "NextToken": str | None, "ResultCnt": int}

    其中每个元素的 ClueText 字段为解析后的对象（若解析成功），并保留原始字段。"""

    _init_sdk_config(access_key_id, secret_access_key, region)

    try:
        from volcenginesdkinsight import (  # type: ignore[import]
            INSIGHTApi,
            QueryClueInfoRequest,
        )
    except ImportError as exc:  # noqa: BLE001
        raise NetworkError(f"导入 volcenginesdkinsight 失败: {exc}") from exc

    api = INSIGHTApi()

    request = QueryClueInfoRequest(
        start_time=StartTime,
        end_time=EndTime,
    )
    if MaxResults:
        request.max_results = MaxResults
    if NextToken:
        request.next_token = NextToken

    try:
        response = api.query_clue_info(request)
    except Exception as exc:  # noqa: BLE001
        _handle_api_exception("QueryClueInfo", exc)
        raise

    clues = getattr(response, "clue_list", None) or []
    next_token = getattr(response, "next_token", None)
    result_cnt_raw = getattr(response, "result_cnt", None)

    clues_plain = _model_to_plain(clues)  # 期望为 List[dict]
    parsed_clues = []
    for item in clues_plain:
        if not isinstance(item, dict):
            continue
        parsed = dict(item)
        raw_text = parsed.get("ClueText") or parsed.get("clue_text")
        if isinstance(raw_text, str) and raw_text:
            try:
                import json

                parsed["ClueText"] = json.loads(raw_text)
            except Exception:  # noqa: BLE001
                # 保留原始字符串
                parsed.setdefault("ClueText", raw_text)
        parsed_clues.append(parsed)

    try:
        result_cnt = int(result_cnt_raw) if result_cnt_raw is not None else len(parsed_clues)
    except Exception:  # noqa: BLE001
        result_cnt = len(parsed_clues)

    return {
        "ClueList": parsed_clues,
        "NextToken": next_token,
        "ResultCnt": result_cnt,
    }


__all__ = [
    "list_custom_subs_task",
    "pull_post",
    "query_clue_info",
]

#!/usr/bin/env python3
# Copyright 2024 ByteDance, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""byted-market-insight-agent 静态自检脚本。

不进行任何外网调用，仅做以下检查：
- 关键文件与目录是否存在；
- Python 版本是否满足推荐要求；
- 关键环境变量是否已设置（仅提示，不强制）；
- 官方 SDK 是否可导入（可选依赖）。
"""

from __future__ import annotations

import importlib.util
import os
import sys
from pathlib import Path


ROOT_DIR = Path(__file__).resolve().parent.parent


def check_python_version() -> bool:
    major, minor = sys.version_info[:2]
    ok = (major, minor) >= (3, 8)
    print(f"[CHECK] Python 版本: {major}.{minor} (要求 >= 3.8) -> {'OK' if ok else 'WARN'}")
    return ok


def check_files() -> bool:
    required = [
        ROOT_DIR / "SKILL.md",
        ROOT_DIR / "scripts" / "client.py",
        ROOT_DIR / "scripts" / "auth_resolver.py",
        ROOT_DIR / "scripts" / "providers" / "gateway_provider.py",
        ROOT_DIR / "scripts" / "providers" / "sdk_provider.py",
        ROOT_DIR / "references" / "usage.md",
        ROOT_DIR / "references" / "api-diff.md",
        ROOT_DIR / "persist",
    ]
    ok = True
    for path in required:
        exists = path.exists()
        kind = "目录" if path.is_dir() else "文件"
        print(f"[CHECK] {kind}: {path.relative_to(ROOT_DIR)!s} -> {'OK' if exists else 'MISSING'}")
        if not exists:
            ok = False
    return ok


def check_env() -> None:
    print("[INFO] 环境变量检查（未设置不影响静态自检，仅影响实际调用）:")
    missing_key = False
    for name in [
        "ARK_SKILL_API_BASE",
        "ARK_SKILL_API_KEY",
        "VOLCSTACK_ACCESS_KEY_ID",
        "VOLCSTACK_SECRET_ACCESS_KEY",
        "VOLCSTACK_REGION",
        "MARKET_INSIGHT_AUTO_PIP",
    ]:
        value = os.getenv(name)
        if value:
            print(f"  - {name}: 已设置（长度={len(value)}）")
        else:
            print(f"  - {name}: 未设置")
            if name in {
                "ARK_SKILL_API_BASE",
                "ARK_SKILL_API_KEY",
                "VOLCSTACK_ACCESS_KEY_ID",
                "VOLCSTACK_SECRET_ACCESS_KEY",
            }:
                missing_key = True
    if missing_key:
        print(
            "[HINT] 检测到部分关键环境变量未设置，可在 ~/.bashrc / ~/.bash_profile / ~/.zshrc / ~/.profile 中查找 export 语句（仅提示，不自动生效）。"
        )


def check_sdk_optional() -> None:
    print("[INFO] 官方 SDK 可选依赖检查：")
    spec_core = importlib.util.find_spec("volcenginesdkcore")
    spec_insight = importlib.util.find_spec("volcenginesdkinsight")
    if spec_core and spec_insight:
        print("  - volcengine SDK: 已检测到 (volcenginesdkcore / volcenginesdkinsight)")
    else:
        print(
            "  - volcengine SDK: 未检测到（运行时需要时将由 auth_resolver 自动尝试安装）"
        )


def main() -> int:
    print("=== byted-market-insight-agent 静态自检 ===")
    ok_py = check_python_version()
    ok_files = check_files()
    check_env()
    check_sdk_optional()

    all_ok = ok_py and ok_files
    print(f"[RESULT] 静态自检 {'通过' if all_ok else '存在问题，请根据上方提示修复后再打包'}")
    return 0 if all_ok else 1


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main())