Linkfox Amazon Ads Entity

"""
Shared helpers for linkfox-amazon-ads-entity scripts.

All list_*.py scripts import from this module for:
  - Dependency check (linkfox-amazon-ads-auth must be installed)
  - LINKFOXAGENT_API_KEY retrieval
  - /amazonAds/storeTokens call to get access token
  - /amazonAds/developerProxy call with the right method / Content-Type per ad product
  - Auto-pagination across Sponsored Products / Sponsored Brands (nextToken) and
    Sponsored Display (startIndex + count offset)

This module is NOT intended to be run directly; it is imported by list_*.py siblings.

Import convention (at top of each list_*.py):

    import sys
    from pathlib import Path
    sys.path.insert(0, str(Path(__file__).resolve().parent))
    from _common import ensure_auth_skill_available, get_access_token, list_sp_entities
    # or list_sd_entities for Sponsored Display
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen


# 生产默认走 tool-gateway.linkfox.com；开发/测试期可 export AMAZON_ADS_BASE_URL=<url> 覆盖
API_BASE_URL = os.environ.get("AMAZON_ADS_BASE_URL") or "https://tool-gateway.linkfox.com"
STORE_TOKENS_ENDPOINT = f"{API_BASE_URL}/amazonAds/storeTokens"
DEVELOPER_PROXY_ENDPOINT = f"{API_BASE_URL}/amazonAds/developerProxy"

REQUIRED_SKILL = "linkfox-amazon-ads-auth"
DEPENDENCY_EXIT_CODE = 42
DEFAULT_MAX_PAGES = 50
DEFAULT_PAGE_SIZE = 100


# ---------- Dependency check ----------

def ensure_auth_skill_available() -> None:
    """Invoke check_auth_dependency.py sibling; exit 42 if auth skill missing."""
    here = Path(__file__).resolve().parent
    checker = here / "check_auth_dependency.py"
    if not checker.exists():
        payload = {
            "missingSkill": REQUIRED_SKILL,
            "reason": "check_auth_dependency.py not found next to this script",
            "suggestedActions": [
                f"Install skill '{REQUIRED_SKILL}' before running linkfox-amazon-ads-entity.",
            ],
        }
        print(f"DEPENDENCY_MISSING: {json.dumps(payload, ensure_ascii=False)}", file=sys.stderr)
        sys.exit(DEPENDENCY_EXIT_CODE)

    try:
        result = subprocess.run(
            [sys.executable, str(checker)],
            capture_output=True,
            text=True,
            timeout=10,
        )
    except Exception as exc:
        payload = {
            "missingSkill": REQUIRED_SKILL,
            "reason": f"Failed to run dependency check: {exc}",
        }
        print(f"DEPENDENCY_MISSING: {json.dumps(payload, ensure_ascii=False)}", file=sys.stderr)
        sys.exit(DEPENDENCY_EXIT_CODE)

    if result.stderr:
        sys.stderr.write(result.stderr)
        if not result.stderr.endswith("\n"):
            sys.stderr.write("\n")
    if result.returncode != 0:
        sys.exit(DEPENDENCY_EXIT_CODE)


# ---------- LinkFox gateway plumbing ----------

def get_api_key() -> str:
    key = os.environ.get("LINKFOXAGENT_API_KEY")
    if not key:
        print(
            "❌ LINKFOXAGENT_API_KEY not configured. Please set:\n"
            "  export LINKFOXAGENT_API_KEY=your-key-here",
            file=sys.stderr,
        )
        sys.exit(1)
    return key


def call_gateway(endpoint: str, payload: dict) -> dict:
    api_key = get_api_key()
    data = json.dumps(payload).encode("utf-8")
    req = Request(
        endpoint,
        data=data,
        headers={
            "Authorization": api_key,
            "Content-Type": "application/json",
            "User-Agent": "LinkFox-Skill/1.0",
        },
        method="POST",
    )
    try:
        with urlopen(req, timeout=60) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except HTTPError as e:
        body = e.read().decode("utf-8") if e.fp else ""
        return {"error": f"HTTP {e.code}: {e.reason}", "details": body}
    except URLError as e:
        return {"error": f"Connection failed: {e.reason}"}


def get_access_token(profile_id: int) -> str:
    """Fetch access token for the given profileId via /amazonAds/storeTokens."""
    print(f"🔑 Fetching access token for profileId={profile_id}…", file=sys.stderr)
    result = call_gateway(STORE_TOKENS_ENDPOINT, {"profileId": int(profile_id)})
    if "error" in result or "accessToken" not in result:
        print(f"❌ Failed to get access token: {result}", file=sys.stderr)
        sys.exit(1)
    return result["accessToken"]


def _developer_proxy_call(region: str, path: str, method: str, access_token: str,
                          profile_id: int, body: str | None, content_type: str | None,
                          query_string: str | None) -> dict:
    payload: dict[str, Any] = {
        "region": region,
        "path": path,
        "method": method,
        "amzAccessToken": access_token,
        "profileId": int(profile_id),
    }
    if body is not None:
        payload["body"] = body
    if content_type:
        payload["contentType"] = content_type
    if query_string:
        payload["queryString"] = query_string
    return call_gateway(DEVELOPER_PROXY_ENDPOINT, payload)


# ---------- SP list (POST, v3, nextToken paginated) ----------

def list_sp_entities(region: str, profile_id: int, access_token: str,
                     entity_path: str, entity_content_type: str, response_key: str,
                     request_body: dict, fetch_all: bool = True,
                     max_pages: int = DEFAULT_MAX_PAGES) -> dict:
    """
    POST a SP v3 list endpoint and optionally auto-paginate via nextToken.

    Returns either:
        {"items": [...], "pagesFetched": N, "truncated": bool}
    or:
        {"error": "...", "httpStatus": N, "body": "<raw>", "details": "..."}

    The caller re-keys "items" to the entity-specific key (campaigns / adGroups / …)
    on the way out, so this function stays entity-agnostic.
    """
    base_body = dict(request_body or {})
    base_body.setdefault("maxResults", DEFAULT_PAGE_SIZE)

    collected: list = []
    token: str | None = None
    pages = 0
    truncated = False

    while True:
        page_body = dict(base_body)
        if token:
            page_body["nextToken"] = token

        resp = _developer_proxy_call(
            region=region,
            path=entity_path,
            method="POST",
            access_token=access_token,
            profile_id=profile_id,
            body=json.dumps(page_body),
            content_type=entity_content_type,
            query_string=None,
        )

        if "error" in resp:
            return {
                "error": resp["error"],
                "details": resp.get("details"),
                "pagesFetched": pages,
            }

        http_status = resp.get("httpStatus")
        if http_status is None or http_status // 100 != 2:
            return {
                "error": f"Upstream HTTP {http_status}",
                "httpStatus": http_status,
                "contentType": resp.get("contentType"),
                "body": resp.get("body"),
                "pagesFetched": pages,
            }

        try:
            parsed = json.loads(resp.get("body") or "{}")
        except Exception as e:
            return {
                "error": f"Failed to parse upstream body as JSON: {e}",
                "body": resp.get("body"),
                "pagesFetched": pages,
            }

        page_items = parsed.get(response_key) or []
        if isinstance(page_items, list):
            collected.extend(page_items)

        pages += 1
        token = parsed.get("nextToken")

        if not token or not fetch_all:
            break
        if pages >= max_pages:
            truncated = True
            break

    return {
        "items": collected,
        "pagesFetched": pages,
        "truncated": truncated,
    }


# ---------- Argv / param helpers ----------

def parse_argv_params(usage_text: str) -> dict:
    """Read sys.argv[1] as JSON, print usage and exit(1) if missing / invalid."""
    if len(sys.argv) < 2:
        print(usage_text, file=sys.stderr)
        sys.exit(1)
    try:
        return json.loads(sys.argv[1])
    except json.JSONDecodeError as e:
        print(f"❌ Invalid JSON: {e}", file=sys.stderr)
        sys.exit(1)


def require_fields(params: dict, required: list[str]) -> None:
    missing = [f for f in required if f not in params]
    if missing:
        print(f"❌ Missing required parameters: {', '.join(missing)}", file=sys.stderr)
        sys.exit(1)


# 字段结构规范表 —— 每个 filter 字段对应的请求体结构（封装层按此规范化入参后再发给上游）
#
# - "object"      : {"include":[...]} / {"exclude":[...]}（多数 id/state 类过滤器）
# - "array"       : 裸数组 ["EXACT","BROAD"]（matchType/expressionType 等枚举过滤器）
# - "scalar"      : 裸字符串 "AUTO"（campaignTargetingType 等单值字段）
# - "text_filter" : {"queryTermMatchType":"BROAD_MATCH","include":["soap"]}（文本搜索类）
# - "client_side" : 上游不支持，由本 skill 在拉回结果后本地过滤（asinFilter / skuFilter）
FILTER_STRUCTURE: dict[str, str] = {
    # 对象型（include/exclude 列表）
    "stateFilter":              "object",
    "campaignIdFilter":         "object",
    "adGroupIdFilter":          "object",
    "keywordIdFilter":          "object",
    "targetIdFilter":           "object",
    "negativeTargetIdFilter":   "object",  # SD negativeTargets
    "negativeKeywordIdFilter":  "object",  # SP negativeKeywords
    "creativeIdFilter":         "object",  # SD creatives
    "adIdFilter":               "object",
    "portfolioIdFilter":        "object",
    # 裸数组型
    "matchTypeFilter":     "array",
    # 注意：expressionTypeFilter 实证是 object-include 结构（与 matchTypeFilter 不同）
    "expressionTypeFilter":"object",
    # 裸字符串型
    "campaignTargetingTypeFilter": "scalar",
    # 文本搜索型（queryTermMatchType + include）
    "nameFilter":          "text_filter",
    "keywordTextFilter":   "text_filter",
    # 本 skill 客户端过滤（上游 Amazon API 未原生支持）
    "asinFilter":          "client_side",
    "skuFilter":           "client_side",
}

# 参与客户端过滤的字段，匹配到返回条目中的哪个字段
CLIENT_SIDE_FILTER_TARGETS: dict[str, str] = {
    "asinFilter": "asin",
    "skuFilter":  "sku",
}


def _normalize_filter_value(stype: str, val):
    """把用户传入的灵活结构规范化为上游需要的形状。

    封装策略：对调用方常见的"写法变体"做宽松兼容，封装掉上游字段结构的差异。
    - "array"  目标形态 ["A","B"]
        接受：["A","B"] / {"include":["A","B"]} / "A"
    - "object" 目标形态 {"include":[...]}
        接受：{"include":[...]} / ["A","B"]（自动包 include） / "A"（包 include 单值）
    - "scalar" 目标形态 "AUTO"
        接受："AUTO" / {"include":["AUTO"]} / ["AUTO"]
    - "text_filter" 目标形态 {"queryTermMatchType":"...","include":[...]}
        原样透传（该字段必须用户按规范写）
    """
    if val is None:
        return None
    if stype == "array":
        if isinstance(val, list):
            return val
        if isinstance(val, dict) and isinstance(val.get("include"), list):
            return val["include"]
        if isinstance(val, str):
            return [val]
        return val
    if stype == "scalar":
        if isinstance(val, str):
            return val
        if isinstance(val, dict) and isinstance(val.get("include"), list) and val["include"]:
            return val["include"][0]
        if isinstance(val, list) and val:
            return val[0]
        return val
    if stype == "object":
        if isinstance(val, dict):
            return val  # 已是 {"include":[...]} / {"exclude":[...]}
        if isinstance(val, list):
            return {"include": val}  # 裸数组兜底包装
        if isinstance(val, str):
            return {"include": [val]}
        return val
    # text_filter / 未知 → 原样透传
    return val


def split_server_client_filters(params: dict, filter_keys: list[str]):
    """将入参拆成「上游请求体」+「本地需过滤的 client-side 过滤器」。

    返回 (server_body, client_filters)：
    - server_body：已按字段结构规范化，可直接作为 /list endpoint 的 JSON body
    - client_filters：{"asinFilter": ["B0XXX"], ...}，待拉回数据后本地筛
    """
    server_body: dict[str, Any] = {}
    client_filters: dict[str, list] = {}

    for k in filter_keys:
        if k not in params or params[k] is None:
            continue
        stype = FILTER_STRUCTURE.get(k, "object")
        val = params[k]
        if stype == "client_side":
            # 归一化成数组，便于后续本地匹配
            if isinstance(val, list):
                values = val
            elif isinstance(val, dict) and isinstance(val.get("include"), list):
                values = val["include"]
            elif isinstance(val, str):
                values = [val]
            else:
                values = []
            if values:
                client_filters[k] = values
        else:
            normalized = _normalize_filter_value(stype, val)
            if normalized is not None:
                server_body[k] = normalized

    # maxResults 透传
    if "maxResults" in params and params["maxResults"] is not None:
        server_body["maxResults"] = int(params["maxResults"])

    # 其他可选顶层字段（扩展数据/本地化）
    for extra in ("includeExtendedDataFields", "locale"):
        if extra in params and params[extra] is not None:
            server_body[extra] = params[extra]

    return server_body, client_filters


def build_filter_body(params: dict, filter_keys: list[str]) -> dict:
    """兼容旧调用方：仅返回上游请求体部分（忽略 client-side 过滤器）。

    新调用方建议直接用 `split_server_client_filters()`，以便拿到 client-side 过滤器做本地筛选。
    """
    server_body, _ = split_server_client_filters(params, filter_keys)
    return server_body


def apply_client_side_filters(items: list, client_filters: dict) -> list:
    """对已拉回的 items 按 client-side 过滤器筛选（用于上游不支持的过滤字段）。"""
    if not client_filters:
        return items
    filtered = items
    for fkey, values in client_filters.items():
        target_field = CLIENT_SIDE_FILTER_TARGETS.get(fkey)
        if not target_field:
            continue
        wanted = set(values)
        filtered = [it for it in filtered if it.get(target_field) in wanted]
    return filtered


# ---------- Sponsored Display (v3) 支持 ----------
#
# Sponsored Display v3 list endpoint 的形态：
#   - 方法：GET，参数位于 querystring
#   - 分页：startIndex + count 偏移分页
#   - 过滤器：扁平字符串（id 类逗号分隔；state 类逗号分隔小写）
#   - 扩展字段：通过路径区分（/sd/<entity> 与 /sd/<entity>/extended）
#
# 过滤字段统一接受 {"include":[...]} / 裸数组 / 单值字符串三种形态，转换为
# Sponsored Display 端所需的 querystring：
#   - build_sd_query()    将 *Filter 入参规范化为 querystring dict + 是否使用 /extended
#   - list_sd_entities()  按 startIndex + count 循环 GET，直到本页 < count 或达 max_pages

# *Filter 入参到 Sponsored Display querystring 字段的映射。
SD_QUERY_ID_FIELDS: dict[str, str] = {
    "campaignIdFilter":       "campaignIdFilter",
    "adGroupIdFilter":        "adGroupIdFilter",
    "adIdFilter":             "adIdFilter",
    "targetIdFilter":         "targetIdFilter",
    # /sd/negativeTargets querystring 不支持 negativeTargetIdFilter；保留映射避免
    # 静默丢字段，传入时上游会返回 400 / 422 由用户感知。
    "negativeTargetIdFilter": "negativeTargetIdFilter",
    "portfolioIdFilter":      "portfolioIdFilter",
    "creativeIdFilter":       "creativeIdFilter",
}

SD_STATE_FIELDS = ("stateFilter",)
SD_NAME_FIELDS = ("nameFilter",)  # SD 上游字段名是 `name`（精确匹配）


def _to_list(val) -> list:
    """把 {"include":[...]} / 裸数组 / 标量 都归一化成一个 list（用于 SD querystring 拼接）。"""
    if val is None:
        return []
    if isinstance(val, list):
        return [str(x) for x in val if x is not None]
    if isinstance(val, dict):
        if isinstance(val.get("include"), list):
            return [str(x) for x in val["include"] if x is not None]
        return []
    return [str(val)]


def build_sd_query(params: dict, filter_keys: list[str]):
    """将 *Filter 入参规范化为 Sponsored Display GET endpoint 所需的扁平 querystring。

    返回 (query_dict, use_extended_path, client_filters)：
      - query_dict     : {"stateFilter": "enabled,paused", "campaignIdFilter": "1,2", ...}
      - use_extended_path : 传 includeExtendedDataFields=true 时为 True，否则 False
      - client_filters : {"asinFilter": ["B0XX"], ...}（上游不支持的过滤字段，由调用方拉回后本地匹配）

    规范化规则：
      - state 类做 .lower()（Sponsored Display 接口端要求 `enabled` 而非 `ENABLED`）
      - id 类合并为逗号分隔字符串
      - nameFilter：Sponsored Display 仅支持精确匹配；若传 queryTermMatchType=BROAD_MATCH，
        在 stderr 输出一次提示，并仅取 include[0] 作 `name` 参数
      - asinFilter / skuFilter：Sponsored Display 端不支持，转为 client-side 过滤
    """
    query: dict[str, str] = {}
    client_filters: dict[str, list] = {}

    # 1) id 类（含 portfolio / creative / target / negativeTarget）
    for key, sd_field in SD_QUERY_ID_FIELDS.items():
        if key not in filter_keys or key not in params or params[key] is None:
            continue
        values = _to_list(params[key])
        if values:
            query[sd_field] = ",".join(values)

    # 2) state 类（小写化）
    for key in SD_STATE_FIELDS:
        if key not in filter_keys or key not in params or params[key] is None:
            continue
        values = [str(v).lower() for v in _to_list(params[key])]
        if values:
            query[key] = ",".join(values)

    # 3) name（精确匹配；SD 没有 BROAD_MATCH）
    for key in SD_NAME_FIELDS:
        if key not in filter_keys or key not in params or params[key] is None:
            continue
        val = params[key]
        include: list = []
        match_type = None
        if isinstance(val, dict):
            include = _to_list(val.get("include"))
            match_type = val.get("queryTermMatchType")
        else:
            include = _to_list(val)
        if match_type and match_type != "EXACT_MATCH":
            print(
                f"⚠️  Sponsored Display nameFilter 仅支持精确匹配；忽略 queryTermMatchType={match_type}，"
                f"按 include[0]={include[0] if include else '(empty)'} 作 name 精确匹配。",
                file=sys.stderr,
            )
        if include:
            query["name"] = include[0]

    # 4) client-side 过滤（asin / sku）
    for fkey in ("asinFilter", "skuFilter"):
        if fkey not in filter_keys or fkey not in params or params[fkey] is None:
            continue
        values = _to_list(params[fkey])
        if values:
            client_filters[fkey] = values

    # 5) extended 路径开关
    use_extended_path = bool(params.get("includeExtendedDataFields"))

    return query, use_extended_path, client_filters


def list_sd_entities(region: str, profile_id: int, access_token: str,
                     entity_path: str,
                     response_key: str,
                     server_query: dict,
                     fetch_all: bool = True,
                     max_pages: int = DEFAULT_MAX_PAGES,
                     page_size: int = DEFAULT_PAGE_SIZE) -> dict:
    """GET 一个 Sponsored Display v3 list endpoint，并按 startIndex + count 自动翻页。

    Sponsored Display 响应顶层是数组（非 `{<entityKey>:[...]}` 结构）；本函数对两种形态都兼容。
    终止条件：本页返回长度 < count（已到最后一页），或累计 pages >= max_pages（兜底）。

    返回结构：
        {"items":[...], "pagesFetched":N, "truncated":bool}
        或 {"error":..., "httpStatus":..., "body":..., "pagesFetched":...}
    """
    if page_size < 1:
        page_size = DEFAULT_PAGE_SIZE
    if page_size > 100:
        page_size = 100

    collected: list = []
    start_index = 0
    pages = 0
    truncated = False

    while True:
        page_query = dict(server_query or {})
        page_query["startIndex"] = str(start_index)
        page_query["count"] = str(page_size)
        qs = urlencode(page_query, doseq=False)

        resp = _developer_proxy_call(
            region=region,
            path=entity_path,
            method="GET",
            access_token=access_token,
            profile_id=profile_id,
            body=None,
            content_type=None,
            query_string=qs,
        )

        if "error" in resp:
            return {
                "error": resp["error"],
                "details": resp.get("details"),
                "pagesFetched": pages,
            }

        http_status = resp.get("httpStatus")
        if http_status is None or http_status // 100 != 2:
            return {
                "error": f"Upstream HTTP {http_status}",
                "httpStatus": http_status,
                "contentType": resp.get("contentType"),
                "body": resp.get("body"),
                "pagesFetched": pages,
            }

        try:
            parsed = json.loads(resp.get("body") or "[]")
        except Exception as e:
            return {
                "error": f"Failed to parse upstream body as JSON: {e}",
                "body": resp.get("body"),
                "pagesFetched": pages,
            }

        if isinstance(parsed, list):
            page_items = parsed
        elif isinstance(parsed, dict):
            page_items = parsed.get(response_key) or []
        else:
            page_items = []

        if isinstance(page_items, list):
            collected.extend(page_items)

        pages += 1
        page_len = len(page_items) if isinstance(page_items, list) else 0

        if not fetch_all or page_len < page_size:
            break
        if pages >= max_pages:
            truncated = True
            break
        start_index += page_size

    return {
        "items": collected,
        "pagesFetched": pages,
        "truncated": truncated,
    }


# ---------- Mutation (POST create / PUT update) ----------

def mutate_entity(region: str, profile_id: int, access_token: str,
                  path: str, method: str, content_type: str,
                  payload) -> dict:
    """POST (create) or PUT (update) an entity batch via developerProxy.

    Returns either:
        {"success": True, "httpStatus": N, "data": <parsed response>}
    or:
        {"error": "...", ...}
    """
    body_str = json.dumps(payload) if payload is not None else None

    resp = _developer_proxy_call(
        region=region,
        path=path,
        method=method,
        access_token=access_token,
        profile_id=profile_id,
        body=body_str,
        content_type=content_type,
        query_string=None,
    )

    if "error" in resp:
        return resp

    http_status = resp.get("httpStatus")
    body_raw = resp.get("body") or ""

    try:
        parsed = json.loads(body_raw) if body_raw else {}
    except (json.JSONDecodeError, TypeError):
        parsed = body_raw

    return {"success": True, "httpStatus": http_status, "data": parsed}

#!/usr/bin/env python3
"""
Dependency Check - linkfox-amazon-ads-entity
============================================

本脚本判断当前环境是否已安装依赖 skill `linkfox-amazon-ads-auth`。

用法:
    python check_auth_dependency.py            # 默认检查
    python check_auth_dependency.py --json     # 以 JSON 输出结果

退出码约定（供 agent 程序化解析）:
    0   → 依赖已满足（找到 linkfox-amazon-ads-auth 的 SKILL.md）
    42  → DEPENDENCY_MISSING: 未找到依赖 skill

stderr 结构化信号:
    - 缺失时：stderr 以 `DEPENDENCY_MISSING:` 开头 + JSON payload
    - 成功时：stderr 以 `DEPENDENCY_OK:` 开头

注意：这是不联网的本地探测，只检查文件系统常见 skill 安装路径。
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path

REQUIRED_SKILL = "linkfox-amazon-ads-auth"
DEPENDENCY_EXIT_CODE = 42


def _split_path_list(raw: str | None) -> list[Path]:
    if not raw or not raw.strip():
        return []
    parts = [p.strip() for p in raw.split(os.pathsep) if p.strip()]
    return [Path(p).expanduser() for p in parts]


def candidate_skill_roots() -> list[Path]:
    roots: list[Path] = []

    for env_var in ("LINKFOX_SKILLS_DIR", "SKILLS_DIR", "CURSOR_SKILLS_DIR"):
        p = os.environ.get(env_var)
        if p:
            roots.append(Path(p).expanduser())

    roots.extend(_split_path_list(os.environ.get("HERMES_SKILLS_EXTERNAL_DIRS")))

    for env_var in ("OPENCLAW_WORKSPACE", "OPENCLAW_ROOT", "OPENCLAW_WORKDIR"):
        ws = os.environ.get(env_var)
        if ws:
            w = Path(ws).expanduser()
            roots.append(w / "skills")
            roots.append(w / ".agents" / "skills")

    oc_skills = os.environ.get("OPENCLAW_SKILLS_DIR")
    if oc_skills:
        roots.append(Path(oc_skills).expanduser())

    try:
        cwd = Path.cwd()
        roots.append(cwd / "skills")
        roots.append(cwd / ".agents" / "skills")
    except OSError:
        pass

    here = Path(__file__).resolve()
    if len(here.parents) >= 3:
        roots.append(here.parents[2])

    home = Path.home()
    roots.extend([
        home / ".claude" / "skills",
        home / ".cursor" / "skills",
        home / ".cursor" / "skills-cursor",
        home / ".linkfox" / "skills",
    ])
    roots.extend([
        home / ".openclaw" / "skills",
        home / ".hermes" / "skills",
    ])

    seen: set[Path] = set()
    unique: list[Path] = []
    for r in roots:
        try:
            rr = r.resolve()
        except OSError:
            rr = r
        if rr not in seen:
            seen.add(rr)
            unique.append(r)
    return unique


def _hermes_category_skill_md(hermes_skills_root: Path) -> Path | None:
    if not hermes_skills_root.is_dir():
        return None
    for category_dir in sorted(hermes_skills_root.iterdir()):
        if not category_dir.is_dir():
            continue
        name = category_dir.name
        if name.startswith(".") or name == ".hub":
            continue
        candidate = category_dir / REQUIRED_SKILL / "SKILL.md"
        if candidate.is_file():
            return candidate
    return None


def _hermes_plugin_skill_md(home: Path) -> Path | None:
    plugins_root = home / ".hermes" / "plugins"
    if not plugins_root.is_dir():
        return None
    for plugin_dir in sorted(plugins_root.iterdir()):
        if not plugin_dir.is_dir():
            continue
        candidate = plugin_dir / "skills" / REQUIRED_SKILL / "SKILL.md"
        if candidate.is_file():
            return candidate
    return None


def locate_dependency() -> Path | None:
    home = Path.home()

    for root in candidate_skill_roots():
        target = root / REQUIRED_SKILL / "SKILL.md"
        if target.is_file():
            return target

    hermes_default = home / ".hermes" / "skills"
    found = _hermes_category_skill_md(hermes_default)
    if found is not None:
        return found

    hsh = os.environ.get("HERMES_SKILLS_HOME")
    if hsh:
        found = _hermes_category_skill_md(Path(hsh).expanduser())
        if found is not None:
            return found

    found = _hermes_plugin_skill_md(home)
    if found is not None:
        return found

    return None


def searched_locations_for_report() -> list[str]:
    home = Path.home()
    out: list[str] = [str(p) for p in candidate_skill_roots()]
    out.append(str(home / ".hermes" / "skills"))
    out.append(str(home / ".hermes" / "plugins"))
    hsh = os.environ.get("HERMES_SKILLS_HOME")
    if hsh:
        out.append(str(Path(hsh).expanduser()))
    seen: set[str] = set()
    unique: list[str] = []
    for s in out:
        if s not in seen:
            seen.add(s)
            unique.append(s)
    return unique


def emit(as_json: bool, ok: bool, payload: dict) -> None:
    prefix = "DEPENDENCY_OK:" if ok else "DEPENDENCY_MISSING:"
    body = json.dumps(payload, ensure_ascii=False)
    if as_json:
        out = dict(payload)
        out["status"] = "ok" if ok else "missing"
        print(json.dumps(out, ensure_ascii=False, indent=2))
    print(f"{prefix} {body}", file=sys.stderr)


def main() -> None:
    parser = argparse.ArgumentParser(description="Check required dependency skill availability.")
    parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON on stdout.")
    args = parser.parse_args()

    found = locate_dependency()

    if found is not None:
        emit(as_json=args.json, ok=True, payload={
            "skill": REQUIRED_SKILL,
            "skillMdPath": str(found),
        })
        sys.exit(0)

    payload = {
        "missingSkill": REQUIRED_SKILL,
        "reason": (
            f"linkfox-amazon-ads-entity 依赖 `{REQUIRED_SKILL}`，"
            "但在常见 skill 安装路径下未找到其 SKILL.md。"
        ),
        "searchedRoots": searched_locations_for_report(),
        "suggestedActions": [
            f"If a skill installer tool is available (e.g. install_skill / skill marketplace MCP), invoke it to install '{REQUIRED_SKILL}' immediately.",
            "Otherwise, ask the user to install the skill from https://skill.linkfox.com/ and retry.",
            "Do NOT bypass the dependency by calling /amazonAds/authorizeUrl or /amazonAds/storeTokens directly from this skill.",
        ],
        "marketplaceUrl": "https://skill.linkfox.com/",
    }
    emit(as_json=args.json, ok=False, payload=payload)
    sys.exit(DEPENDENCY_EXIT_CODE)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Skill response I/O helper — wraps any main script to persist large API
responses to disk, then offers a `read` subcommand to extract specific fields
from those persisted files. Generic, business-agnostic.

This script is bundled into each skill's scripts/ directory by tools/response_io/sync.py.
The agent must pass --script <path> to identify which main script to execute.

Usage:
  python scripts/response_io.py run --script <PATH> --out-dir <DIR> '<json_params>' [--label NAME] [--timeout SEC]
  python scripts/response_io.py read <file> (--path "<JMESPath>" | --fields "f1,f2,...") [--limit N] [--offset M] [--format json|jsonl|csv|table]
"""

from __future__ import annotations

import argparse
import csv
import io
import json
import os
import re
import secrets
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from typing import Any

# Force UTF-8 stdout/stderr so non-ASCII chars in previews and API responses
# print correctly on Windows (default cp936 / gbk).
for stream in (sys.stdout, sys.stderr):
    try:
        stream.reconfigure(encoding="utf-8")  # type: ignore[attr-defined]
    except (AttributeError, OSError):
        pass

try:
    import jmespath  # type: ignore
    HAS_JMESPATH = True
except ImportError:
    HAS_JMESPATH = False


MAX_STRING_LEN = 120
MAX_DEPTH = 3
SAMPLE_KEY_CAP = 15
RAW_TEXT_PEEK = 500
DEFAULT_TIMEOUT_SEC = 300


# ---------------------------------------------------------------------------
# Shared helpers
# ---------------------------------------------------------------------------


def _err(msg: str, code: int = 1) -> None:
    print(msg, file=sys.stderr)
    sys.exit(code)


def _resolve_script(script_arg: str) -> Path:
    p = Path(script_arg).expanduser()
    if not p.is_absolute():
        # Resolve relative to the current working directory the agent invoked from.
        p = (Path.cwd() / p).resolve()
    else:
        p = p.resolve()
    if not p.is_file():
        _err(f"--script path not found: {p}")
    return p


def _resolve_skill_name(main_script: Path) -> str:
    """Best-effort skill name extraction for filename prefixing.

    main_script lives at <skill_dir>/scripts/<name>.py — return <skill_dir>'s
    folder name. Fall back to the script's stem if structure differs.
    """
    try:
        if main_script.parent.name == "scripts":
            return main_script.parents[1].name
    except IndexError:
        pass
    return main_script.stem


def _sanitize_label(label: str) -> str:
    """Allow only safe filename chars in --label to prevent path traversal."""
    cleaned = re.sub(r"[^\w\-]", "_", label)
    return cleaned[:64]  # cap length


def _truncate_string(s: str) -> str:
    if len(s) <= MAX_STRING_LEN:
        return s
    return s[:MAX_STRING_LEN] + f"...(truncated, total {len(s)} chars)"


def _truncate_value(value: Any, depth: int = 0) -> Any:
    """Recursively truncate strings, deep nesting, and large arrays for preview."""
    if depth >= MAX_DEPTH:
        if isinstance(value, dict):
            return f"<truncated nested object, keys: {list(value.keys())[:10]}>"
        if isinstance(value, list):
            return f"<truncated nested array, length: {len(value)}>"
        if isinstance(value, str):
            return _truncate_string(value)
        return value
    if isinstance(value, str):
        return _truncate_string(value)
    if isinstance(value, dict):
        out = {k: _truncate_value(v, depth + 1) for k, v in value.items()}
        return out
    if isinstance(value, list):
        if not value:
            return []
        truncated = [_truncate_value(value[0], depth + 1)]
        if len(value) > 1:
            # Note total length on the parent — keep the array type-homogeneous
            # so downstream consumers can iterate without special-casing strings.
            truncated.append({"_omitted_items": len(value) - 1})
        return truncated
    return value


def _shape_of(value: Any, top: bool = False) -> Any:
    """Lightweight schema description for the preview block."""
    if isinstance(value, dict):
        keys = list(value.keys())
        out: dict[str, Any] = {"type": "object", "top_keys" if top else "keys": keys}
        if top:
            for k in keys[:8]:
                out[k] = _shape_of(value[k])
        return out
    if isinstance(value, list):
        out = {"type": "array", "length": len(value)}
        if value and isinstance(value[0], dict):
            out["item_keys"] = list(value[0].keys())
        elif value:
            out["item_type"] = type(value[0]).__name__
        return out
    return {"type": type(value).__name__}


def _build_sample(value: Any) -> Any:
    """First-record sample with explicit truncation marker."""
    if isinstance(value, list):
        if not value:
            return {"_truncated_record": True, "_note": "array is empty"}
        first = value[0]
        if isinstance(first, dict):
            sample = {"_truncated_record": True, "_note": f"first of {len(value)} items"}
            sample.update(_truncate_value(first, depth=1))
            return sample
        return {"_truncated_record": True, "_note": f"first of {len(value)} items", "value": _truncate_value(first, depth=1)}
    if isinstance(value, dict):
        sample = {"_truncated_record": True, "_note": "top-level object (truncated)"}
        sample.update(_truncate_value(value, depth=1))
        return sample
    return {"_truncated_record": True, "value": _truncate_value(value, depth=1)}


def _shrink_preview(preview: dict) -> dict:
    """Cap the sample's value fields when it has many keys.

    `shape.*.item_keys` is the single source of truth for the full key list
    (always complete, no truncation). The sample only ever shows up to
    SAMPLE_KEY_CAP fields with their concrete values, since the agent only
    needs a feel for value shapes — for the full menu of available fields,
    they read `shape`.
    """
    sample = preview.get("sample")
    if isinstance(sample, dict):
        meta_keys = {"_truncated_record", "_note"}
        data_keys = [k for k in sample.keys() if k not in meta_keys]
        if len(data_keys) > SAMPLE_KEY_CAP:
            kept = data_keys[:SAMPLE_KEY_CAP]
            new_sample = {k: v for k, v in sample.items() if k in meta_keys or k in kept}
            base_note = sample.get("_note", "")
            extra = (
                f"showing first {SAMPLE_KEY_CAP} of {len(data_keys)} fields "
                f"(see `shape` for the complete key list)"
            )
            new_sample["_note"] = f"{base_note}; {extra}" if base_note else extra
            preview["sample"] = new_sample
    return preview


# ---------------------------------------------------------------------------
# `run` subcommand
# ---------------------------------------------------------------------------


def cmd_run(args: argparse.Namespace) -> int:
    main_script = _resolve_script(args.script)
    skill_name = _resolve_skill_name(main_script)

    out_dir = Path(args.out_dir).expanduser().resolve()
    try:
        out_dir.mkdir(parents=True, exist_ok=True)
    except OSError as e:
        _err(f"Failed to create --out-dir {out_dir}: {e}")
    if not os.access(out_dir, os.W_OK):
        _err(f"--out-dir is not writable: {out_dir}")

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    rand = secrets.token_hex(3)
    safe_label = _sanitize_label(args.label) if args.label else ""
    label_part = f"__{safe_label}" if safe_label else ""
    out_file = out_dir / f"{skill_name}__{timestamp}_{rand}{label_part}.json"

    # Force the child process to emit UTF-8 regardless of the host console
    # encoding (Windows defaults to cp936 / gbk and would otherwise corrupt
    # non-ASCII bytes when we read them back).
    child_env = os.environ.copy()
    child_env["PYTHONIOENCODING"] = "utf-8"

    timed_out = False
    try:
        proc = subprocess.run(
            [sys.executable, str(main_script), args.params],
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            env=child_env,
            timeout=args.timeout,
        )
        stdout_text = proc.stdout or ""
        stderr_text = proc.stderr or ""
        returncode = proc.returncode
    except subprocess.TimeoutExpired as e:
        timed_out = True
        stdout_text = (e.stdout.decode("utf-8", errors="replace") if isinstance(e.stdout, bytes) else (e.stdout or "")) or ""
        stderr_text = (e.stderr.decode("utf-8", errors="replace") if isinstance(e.stderr, bytes) else (e.stderr or "")) or ""
        returncode = 124  # convention for timeout

    # Always write the captured stdout to disk, even if not JSON.
    try:
        out_file.write_text(stdout_text, encoding="utf-8")
    except OSError as e:
        _err(f"Failed to write output file {out_file}: {e}")

    if stderr_text:
        sys.stderr.write(stderr_text)

    # Try to parse the captured stdout as JSON for the preview.
    try:
        parsed = json.loads(stdout_text) if stdout_text.strip() else None
        format_kind = "json"
    except json.JSONDecodeError:
        parsed = None
        format_kind = "raw_text"

    preview: dict[str, Any] = {
        "_preview": {
            "is_preview": True,
            "warning": (
                "PREVIEW ONLY — NOT FULL DATA. The full response is saved to `file`. "
                "Use `python scripts/response_io.py read <file> --fields '...'` to extract "
                "specific fields, or `--path '<JMESPath>'` for complex projections."
            ),
        },
    }

    # Surface failures prominently so agents don't mistake a stub preview for success.
    if returncode != 0 or timed_out:
        stderr_snippet = stderr_text[-500:] if stderr_text else ""
        preview["_error"] = {
            "exit_code": returncode,
            "timed_out": timed_out,
            "stderr_snippet": stderr_snippet,
            "hint": "The wrapped script failed or timed out. The output file may be empty or partial.",
        }

    preview.update({
        "file": str(out_file),
        "size_bytes": out_file.stat().st_size,
        "skill": skill_name,
        "exit_code": returncode,
        "format": format_kind,
        "label": safe_label or None,
        "next_steps_hint": (
            "use: python scripts/response_io.py read <file> --fields '...' | --path '...'"
        ),
    })

    if format_kind == "json":
        preview["shape"] = _shape_of(parsed, top=True)
        preview["sample"] = _build_sample(parsed)
    else:
        peek = stdout_text[:RAW_TEXT_PEEK]
        preview["raw_text_peek"] = peek
        preview["raw_text_total_chars"] = len(stdout_text)
        preview["sample"] = {
            "_truncated_record": True,
            "_note": f"stdout was not valid JSON; first {RAW_TEXT_PEEK} chars shown above in raw_text_peek",
        }

    preview = _shrink_preview(preview)
    print(json.dumps(preview, ensure_ascii=False, indent=2))
    return returncode


# ---------------------------------------------------------------------------
# `read` subcommand
# ---------------------------------------------------------------------------


def _load_json(path: Path) -> Any:
    try:
        text = path.read_text(encoding="utf-8")
    except OSError as e:
        _err(f"Failed to read file {path}: {e}")
    try:
        return json.loads(text)
    except json.JSONDecodeError as e:
        _err(f"File is not valid JSON: {path}\n{e}")


def _basic_dot_path(data: Any, path: str) -> Any:
    """Pure-stdlib dot-path resolver. No [*] support — callers fall back here only when jmespath is unavailable AND the path has no [*]."""
    cur = data
    for part in path.split("."):
        if isinstance(cur, dict):
            cur = cur.get(part)
        else:
            return None
    return cur


def _resolve_field(data: Any, expr: str) -> Any:
    if HAS_JMESPATH:
        return jmespath.search(expr, data)
    if "[" in expr or "*" in expr:
        _err(
            f"jmespath is required for expression '{expr}'. "
            f"Install with: pip install jmespath"
        )
    return _basic_dot_path(data, expr)


def _project_fields(data: Any, fields: list[str]) -> Any:
    """Run each field expr; if any returns a list, zip them into list-of-dicts."""
    resolved: dict[str, Any] = {f: _resolve_field(data, f) for f in fields}

    list_lengths = [len(v) for v in resolved.values() if isinstance(v, list)]
    if not list_lengths:
        return resolved

    # All list values must be same length to zip cleanly.
    if len(set(list_lengths)) > 1:
        # Fallback: return the dict as-is so caller can inspect mismatches.
        return resolved

    n = list_lengths[0]
    rows = []
    for i in range(n):
        row = {}
        for f, v in resolved.items():
            row[f] = v[i] if isinstance(v, list) else v
        rows.append(row)
    return rows


def _apply_slice(value: Any, limit: int | None, offset: int | None) -> Any:
    if not isinstance(value, list):
        return value
    start = offset or 0
    end = (start + limit) if limit is not None else None
    return value[start:end]


def _format_output(value: Any, fmt: str) -> str:
    if fmt == "json":
        return json.dumps(value, ensure_ascii=False, indent=2)
    if fmt == "jsonl":
        if isinstance(value, list):
            return "\n".join(json.dumps(item, ensure_ascii=False) for item in value)
        return json.dumps(value, ensure_ascii=False)
    if fmt in ("csv", "table"):
        if not isinstance(value, list) or not value:
            _err(f"--format {fmt} requires a non-empty list result")
        if not all(isinstance(item, dict) for item in value):
            _err(f"--format {fmt} requires list-of-objects, got list of {type(value[0]).__name__}")
        keys: list[str] = []
        for item in value:
            for k in item.keys():
                if k not in keys:
                    keys.append(k)
        if fmt == "csv":
            buf = io.StringIO()
            writer = csv.DictWriter(buf, fieldnames=keys, extrasaction="ignore")
            writer.writeheader()
            for item in value:
                writer.writerow({k: _stringify(item.get(k)) for k in keys})
            return buf.getvalue().rstrip("\n")
        # table: simple aligned columns
        rows = [[_stringify(item.get(k)) for k in keys] for item in value]
        widths = [len(k) for k in keys]
        for row in rows:
            for i, cell in enumerate(row):
                widths[i] = max(widths[i], len(cell))
        lines = [
            "  ".join(k.ljust(widths[i]) for i, k in enumerate(keys)),
            "  ".join("-" * widths[i] for i in range(len(keys))),
        ]
        for row in rows:
            lines.append("  ".join(row[i].ljust(widths[i]) for i in range(len(keys))))
        return "\n".join(lines)
    _err(f"Unknown --format: {fmt}")
    return ""  # unreachable


def _stringify(v: Any) -> str:
    if v is None:
        return ""
    if isinstance(v, (dict, list)):
        return json.dumps(v, ensure_ascii=False)
    return str(v)


def cmd_read(args: argparse.Namespace) -> int:
    if not args.path and not args.fields:
        _err("read: either --path or --fields is required")
    if args.path and args.fields:
        _err("read: --path and --fields are mutually exclusive")

    file_path = Path(args.file).expanduser().resolve()
    data = _load_json(file_path)

    if args.path:
        result = _resolve_field(data, args.path)
    else:
        fields = [f.strip() for f in args.fields.split(",") if f.strip()]
        if not fields:
            _err("--fields parsed to empty list")
        result = _project_fields(data, fields)

    result = _apply_slice(result, args.limit, args.offset)
    print(_format_output(result, args.format))
    return 0


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def main() -> int:
    parser = argparse.ArgumentParser(
        prog="response_io.py",
        description="Persist large skill API responses to disk and read fields on demand.",
    )
    sub = parser.add_subparsers(dest="cmd", required=True)

    p_run = sub.add_parser(
        "run",
        help="Execute a main script and persist its stdout to a file; "
             "print only a lightweight preview to stdout.",
    )
    p_run.add_argument("params", help="JSON params string passed verbatim to the main script (argv[1]).")
    p_run.add_argument("--script", required=True, help="Path to the main script to execute, e.g. scripts/my_api.py")
    p_run.add_argument("--out-dir", required=True, help="Directory to write the response file into (created if missing).")
    p_run.add_argument("--label", default=None, help="Optional filename suffix; sanitized to safe filename characters.")
    p_run.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT_SEC, help=f"Subprocess timeout in seconds (default: {DEFAULT_TIMEOUT_SEC}).")
    p_run.set_defaults(func=cmd_run)

    p_read = sub.add_parser(
        "read",
        help="Extract specific fields from a previously persisted response file.",
    )
    p_read.add_argument("file", help="Path to the persisted JSON response file.")
    g = p_read.add_mutually_exclusive_group()
    g.add_argument("--path", default=None, help="JMESPath expression, e.g. 'data[*].{asin: asin, title: title}'.")
    g.add_argument("--fields", default=None, help="Comma-separated field paths, e.g. 'data[*].asin,data[*].title'.")
    p_read.add_argument("--limit", type=int, default=None, help="Take at most N items (when result is a list).")
    p_read.add_argument("--offset", type=int, default=None, help="Skip the first M items (when result is a list).")
    p_read.add_argument("--format", choices=["json", "jsonl", "csv", "table"], default="json", help="Output format (default: json).")
    p_read.set_defaults(func=cmd_read)

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


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
SB Ad Groups Create - linkfox-amazon-ads-entity
===============================================

POST sb/v4/adGroups (Amazon Ads).

Usage:
  python create_ad_groups.py '<JSON params>'

Required:
  profileId (number), region (NA/EU/FE), payload (Amazon-native body)

payload structure:
  {"adGroups":[{...}]}
"""

import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from _common import (  # noqa: E402
    ensure_auth_skill_available,
    get_access_token,
    mutate_entity,
    parse_argv_params,
    require_fields,
)

ENTITY_PATH = "sb/v4/adGroups"
CONTENT_TYPE = "application/vnd.sbadgroupresource.v4+json"
METHOD = "POST"


def main() -> None:
    params = parse_argv_params(__doc__)

    if not params.get("skipDepCheck"):
        ensure_auth_skill_available()

    require_fields(params, ["profileId", "region", "payload"])

    profile_id = int(params["profileId"])
    region = params["region"]

    access_token = get_access_token(profile_id)
    result = mutate_entity(
        region=region,
        profile_id=profile_id,
        access_token=access_token,
        path=ENTITY_PATH,
        method=METHOD,
        content_type=CONTENT_TYPE,
        payload=params["payload"],
    )

    if "error" in result:
        print(json.dumps(result, indent=2, ensure_ascii=False))
        sys.exit(1)

    print(json.dumps(result, indent=2, ensure_ascii=False))
    print(
        f"\n\u2713 {METHOD} {ENTITY_PATH} \u2014 HTTP {result.get('httpStatus', '?')}",
        file=sys.stderr,
    )


if __name__ == "__main__":
    main()