Linkfox Amazon Store Listings

#!/usr/bin/env python3
"""
Dependency Check - linkfox-amazon-store-listings
==================================================

本脚本用于判断当前运行环境里是否已经安装 / 加载了依赖 skill
`linkfox-amazon-store-auth`（与 `linkfox-amazon-store-report` 共用同一检查逻辑）。

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

退出码约定（供 agent 程序化解析）:
    0   → 依赖已满足（找到 linkfox-amazon-store-auth 的 SKILL.md）
    42  → DEPENDENCY_MISSING: 未找到依赖 skill，agent 需要触发安装流程

stderr 结构化信号:
    - 若依赖缺失，stderr 第一行会以 `DEPENDENCY_MISSING:` 开头，
      后跟 JSON payload，包含所需 skill 名与建议的安装动作。
    - 成功时 stderr 以 `DEPENDENCY_OK:` 开头。

注意:
    这是一个**不联网**的本地探测脚本。它只检查文件系统上常见的
    skill 安装路径（含 **OpenClaw**、**Hermes Agent** 的常见布局）；
    真正的"能不能调授权接口"取决于依赖 skill 的脚本是否可执行——
    这一点由 get_report.py 在运行时再做一次二次校验（通过尝试调用
    /spApi/storeTokens）。

    OpenClaw 参考: workspace 下 `<workspace>/skills`、`<workspace>/.agents/skills`，
    以及 `~/.openclaw/skills`、`~/.agents/skills`（与官方文档优先级一致）。

    Hermes Agent 参考: `~/.hermes/skills/<category>/<skill-name>/SKILL.md`，
    以及 `~/.hermes/plugins/<plugin>/skills/<skill-name>/SKILL.md`；
    额外目录可在 `~/.hermes/config.yaml` 的 `skills.external_dirs` 中配置，
    本脚本无法解析 YAML，请通过环境变量 `HERMES_SKILLS_EXTERNAL_DIRS`（冒号
    或分号分隔的多个路径）或通用的 `LINKFOX_SKILLS_DIR` / `SKILLS_DIR` 注入。
"""

from __future__ import annotations

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

REQUIRED_SKILL = "linkfox-amazon-store-auth"
# 仍兼容本机未重命名时的旧目录名（历史安装 linkfox-amazon-spapi-auth）
_AUTH_SKILL_DIR_ALIASES = ("linkfox-amazon-store-auth", "linkfox-amazon-spapi-auth")
DEPENDENCY_EXIT_CODE = 42


def _split_path_list(raw: str | None) -> list[Path]:
    """按 OS 路径分隔符拆分（Windows 为 `;`，Unix 为 `:`），避免误拆盘符。"""
    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]:
    """按常见的 skill 存放位置，由近到远返回候选根目录（扁平：root/<skill>/SKILL.md）。"""
    roots: list[Path] = []

    # 1) 通过环境变量显式指定（最高优先级）
    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())

    # 1b) Hermes config.yaml skills.external_dirs 等价注入（本脚本不读 YAML）
    #     例: export HERMES_SKILLS_EXTERNAL_DIRS="$HOME/.agents/skills"
    roots.extend(_split_path_list(os.environ.get("HERMES_SKILLS_EXTERNAL_DIRS")))

    # 2) OpenClaw：工作区下的 skills（与 linkfoxskill / OpenClaw 文档一致）
    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")

    # 2b) OpenClaw 显式 skills 目录
    oc_skills = os.environ.get("OPENCLAW_SKILLS_DIR")
    if oc_skills:
        roots.append(Path(oc_skills).expanduser())

    # 2c) 当前工作目录下的 workspace skills（CLI 常在项目根执行）
    try:
        cwd = Path.cwd()
        roots.append(cwd / "skills")
        roots.append(cwd / ".agents" / "skills")
    except OSError:
        pass

    # 3) 与本脚本相邻的 skills/ 目录（仓库开发场景）
    here = Path(__file__).resolve()
    # .../skills/linkfox-amazon-store-report/scripts/check_auth_dependency.py
    # parents[2] -> .../skills
    if len(here.parents) >= 3:
        roots.append(here.parents[2])

    # 4) 用户级常见安装位置（Claude / Cursor / LinkFox）
    home = Path.home()
    roots.extend([
        home / ".claude" / "skills",
        home / ".cursor" / "skills",
        home / ".cursor" / "skills-cursor",
        home / ".linkfox" / "skills",
    ])

    # 5) OpenClaw 全局与跨工具共享目录
    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, skill_dir_name: str) -> Path | None:
    """
    Hermes Agent 布局: ~/.hermes/skills/<category>/<skill-name>/SKILL.md
    跳过 .hub、点目录等非 category 项。
    """
    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 / skill_dir_name / "SKILL.md"
        if candidate.is_file():
            return candidate
    return None


def _hermes_plugin_skill_md(home: Path, skill_dir_name: str) -> Path | None:
    """~/.hermes/plugins/<plugin>/skills/<skill-name>/SKILL.md"""
    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" / skill_dir_name / "SKILL.md"
        if candidate.is_file():
            return candidate
    return None


def locate_dependency() -> Path | None:
    """返回依赖 skill 的 SKILL.md 路径；未找到返回 None。"""
    home = Path.home()

    for skill_dir_name in _AUTH_SKILL_DIR_ALIASES:
        # A) 扁平布局：root/<skill_dir_name>/SKILL.md
        for root in candidate_skill_roots():
            target = root / skill_dir_name / "SKILL.md"
            if target.is_file():
                return target

        # B) Hermes：~/.hermes/skills/<category>/<skill_dir_name>/SKILL.md
        hermes_default = home / ".hermes" / "skills"
        found = _hermes_category_skill_md(hermes_default, skill_dir_name)
        if found is not None:
            return found

        # C) Hermes：显式 HERMES_SKILLS_HOME（若用户把 category 根指到别处）
        hsh = os.environ.get("HERMES_SKILLS_HOME")
        if hsh:
            found = _hermes_category_skill_md(Path(hsh).expanduser(), skill_dir_name)
            if found is not None:
                return found

        # D) Hermes 插件内 skills
        found = _hermes_plugin_skill_md(home, skill_dir_name)
        if found is not None:
            return found

    return None


def searched_locations_for_report() -> list[str]:
    """供 DEPENDENCY_MISSING 调试：列出已扫描的扁平根目录 + Hermes 特化路径。"""
    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:
        # 把 prefix 也放进 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)

    # 未找到：给 agent 一份结构化的行动指引
    payload = {
        "missingSkill": REQUIRED_SKILL,
        "reason": (
            f"linkfox-amazon-store-listings 依赖 `{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.",
            "On OpenClaw: ensure the dependency is under <workspace>/skills, ~/.openclaw/skills, or ~/.agents/skills; set OPENCLAW_WORKSPACE or OPENCLAW_SKILLS_DIR if installs are non-default.",
            "On Hermes Agent: ensure the dependency is under ~/.hermes/skills/<category>/ or a plugin skills/ folder; for external_dirs from config.yaml, export HERMES_SKILLS_EXTERNAL_DIRS with OS path separators.",
            "Do NOT bypass the dependency by calling /spApi/authorizeUrl or /spApi/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
"""
Amazon Store — Delete Listings Item (SP-API Listings Items v2021-08-01)
=======================================================================

通过 LinkFox **POST /spApi/developerProxy** 转发 **DELETE deleteListingsItem**，
删除指定 marketplace 下的刊登；与 get 相同：先 **storeTokens**，再代理调用（**无**请求体）。

官方参考: https://developer-docs.amazon.com/sp-api/reference/deletelistingsitem

Usage:
  python delete_listings_item.py '{
    "sellerId": "A1BCDEFGHIJK2",
    "region": "NA",
    "sku": "My-Seller-SKU-001",
    "marketplaceIds": ["ATVPDKIKX0DER"]
  }'

Optional JSON:
  issueLocale, skipDepCheck

说明：官方 **marketplaceIds** 约束为 **length ≤ 1**；脚本在多于 1 个 id 时报错。
"""

from typing import List

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

LISTINGS_API_VERSION = "2021-08-01"

API_BASE_URL = os.environ.get("STORE_API_BASE_URL") or os.environ.get(
    "SPAPI_BASE_URL", "https://tool-gateway.linkfox.com"
)
STORE_TOKENS_ENDPOINT = f"{API_BASE_URL.rstrip('/')}/spApi/storeTokens"
DEVELOPER_PROXY_ENDPOINT = f"{API_BASE_URL.rstrip('/')}/spApi/developerProxy"

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


def ensure_auth_skill_available() -> None:
    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 delete_listings_item.py",
        }
        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:  # pragma: no cover
        payload = {"missingSkill": REQUIRED_SKILL, "reason": str(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)


def get_api_key() -> str:
    key = os.environ.get("LINKFOXAGENT_API_KEY")
    if not key:
        print(
            "API Key not configured. Set:\n  export LINKFOXAGENT_API_KEY=<your-key>",
            file=sys.stderr,
        )
        sys.exit(1)
    return key


def call_api(endpoint: str, params: dict) -> dict:
    api_key = get_api_key()
    data = json.dumps(params).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 response:
            return json.loads(response.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_store_tokens(seller_id: str, region: str) -> dict:
    return call_api(STORE_TOKENS_ENDPOINT, {"sellerId": seller_id, "region": region})


def developer_proxy_delete(
    region: str,
    path: str,
    access_token: str,
    query_string: str,
) -> dict:
    params: dict = {
        "region": region,
        "path": path,
        "method": "DELETE",
        "amzAccessToken": access_token,
        "queryString": query_string,
    }
    return call_api(DEVELOPER_PROXY_ENDPOINT, params)


def _path_delete(seller_id: str, sku: str) -> str:
    return (
        f"listings/{LISTINGS_API_VERSION}/items/"
        f"{quote(seller_id, safe='')}/{quote(sku, safe='')}"
    )


def _norm_marketplace_ids(val: object) -> List[str]:
    if isinstance(val, str):
        return [x.strip() for x in val.split(",") if x.strip()]
    if isinstance(val, list):
        return [str(x).strip() for x in val if str(x).strip()]
    raise ValueError("marketplaceIds must be a string or non-empty array")


def _build_delete_query(params: dict) -> str:
    mids = _norm_marketplace_ids(params["marketplaceIds"])
    if not mids:
        raise ValueError("marketplaceIds is required")
    if len(mids) > 1:
        raise ValueError(
            "deleteListingsItem requires at most one marketplace id (Amazon: marketplaceIds length ≤ 1)"
        )
    parts: List[str] = [f"marketplaceIds={quote(mids[0], safe='')}"]

    il = params.get("issueLocale")
    if il is not None and str(il).strip():
        parts.append(f"issueLocale={quote(str(il).strip(), safe='')}")

    return "&".join(parts)


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: delete_listings_item.py '<JSON>'\n"
            "Required: sellerId, region, sku, marketplaceIds (exactly one id)\n"
            "Optional: issueLocale\n"
            "See https://developer-docs.amazon.com/sp-api/reference/deletelistingsitem",
            file=sys.stderr,
        )
        sys.exit(1)
    try:
        params = json.loads(sys.argv[1])
    except json.JSONDecodeError as e:
        print(f"Invalid JSON: {e}", file=sys.stderr)
        sys.exit(1)

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

    for f in ("sellerId", "region", "sku", "marketplaceIds"):
        if f not in params:
            print(f"Missing required field: {f}", file=sys.stderr)
            sys.exit(1)

    seller_id = str(params["sellerId"])
    region = str(params["region"])
    sku = str(params["sku"])

    try:
        query_string = _build_delete_query(params)
    except ValueError as e:
        print(str(e), file=sys.stderr)
        sys.exit(1)

    path = _path_delete(seller_id, sku)
    tokens = get_store_tokens(seller_id, region)
    if "error" in tokens or "accessToken" not in tokens:
        print(json.dumps(tokens, indent=2, ensure_ascii=False))
        sys.exit(1)

    access_token = tokens["accessToken"]
    proxy = developer_proxy_delete(region, path, access_token, query_string)

    out = {
        "developerProxy": proxy,
        "resolvedPath": path,
        "queryString": query_string,
    }
    if proxy.get("errcode") == 200 and proxy.get("httpStatus") == 200:
        body_raw = proxy.get("body") or "{}"
        try:
            out["deleteResult"] = json.loads(body_raw)
        except json.JSONDecodeError:
            out["deleteResult"] = None
            out["deleteResultRaw"] = body_raw
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Amazon Store — Get Definitions Product Type (SP-API Product Type Definitions v2020-09-01)
==========================================================================================

通过 LinkFox **POST /spApi/developerProxy** 转发 **GET getDefinitionsProductType**，
获取指定 Amazon product type 的 JSON Schema 定义。

官方参考: https://developer-docs.amazon.com/sp-api/reference/getdefinitionsproducttype

Usage:
  python get_definitions_product_type.py '{
    "sellerId": "A1BCDEFGHIJK2",
    "region": "NA",
    "productType": "LUGGAGE",
    "marketplaceIds": ["ATVPDKIKX0DER"]
  }'

Optional JSON:
  querySellerId → 写入 Query 的 sellerId（卖家专属 schema；可选，常与取令牌用的 sellerId 相同）
  productTypeVersion, requirements, requirementsEnforced, locale, skipDepCheck
"""

from typing import List

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

DEFINITIONS_API_VERSION = "2020-09-01"

API_BASE_URL = os.environ.get("STORE_API_BASE_URL") or os.environ.get(
    "SPAPI_BASE_URL", "https://tool-gateway.linkfox.com"
)
STORE_TOKENS_ENDPOINT = f"{API_BASE_URL.rstrip('/')}/spApi/storeTokens"
DEVELOPER_PROXY_ENDPOINT = f"{API_BASE_URL.rstrip('/')}/spApi/developerProxy"

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


def ensure_auth_skill_available() -> None:
    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 get_definitions_product_type.py",
        }
        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:  # pragma: no cover
        payload = {"missingSkill": REQUIRED_SKILL, "reason": str(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)


def get_api_key() -> str:
    key = os.environ.get("LINKFOXAGENT_API_KEY")
    if not key:
        print(
            "API Key not configured. Set:\n  export LINKFOXAGENT_API_KEY=<your-key>",
            file=sys.stderr,
        )
        sys.exit(1)
    return key


def call_api(endpoint: str, params: dict) -> dict:
    api_key = get_api_key()
    data = json.dumps(params).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=120) as response:
            return json.loads(response.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_store_tokens(seller_id: str, region: str) -> dict:
    return call_api(STORE_TOKENS_ENDPOINT, {"sellerId": seller_id, "region": region})


def developer_proxy_get(
    region: str,
    path: str,
    access_token: str,
    query_string: str,
) -> dict:
    params: dict = {
        "region": region,
        "path": path,
        "method": "GET",
        "amzAccessToken": access_token,
        "queryString": query_string,
    }
    return call_api(DEVELOPER_PROXY_ENDPOINT, params)


def _norm_marketplace_ids(val: object) -> List[str]:
    if isinstance(val, str):
        return [x.strip() for x in val.split(",") if x.strip()]
    if isinstance(val, list):
        return [str(x).strip() for x in val if str(x).strip()]
    raise ValueError("marketplaceIds must be a string or non-empty array")


def _path_for_product_type(product_type: str) -> str:
    enc = quote(product_type.strip(), safe="")
    return f"definitions/{DEFINITIONS_API_VERSION}/productTypes/{enc}"


def _build_query(params: dict) -> str:
    mids = _norm_marketplace_ids(params["marketplaceIds"])
    if not mids:
        raise ValueError("marketplaceIds is required")
    if len(mids) > 1:
        raise ValueError(
            "getDefinitionsProductType: marketplaceIds is limited to one id at this time (Amazon)"
        )
    parts: List[str] = [f"marketplaceIds={quote(mids[0], safe='')}"]

    qs = params.get("querySellerId")
    if qs is not None and str(qs).strip():
        parts.append(f"sellerId={quote(str(qs).strip(), safe='')}")

    ptv = params.get("productTypeVersion")
    if ptv is not None and str(ptv).strip():
        parts.append(f"productTypeVersion={quote(str(ptv).strip(), safe='')}")

    req = params.get("requirements")
    if req is not None and str(req).strip():
        parts.append(f"requirements={quote(str(req).strip(), safe='')}")

    re_ = params.get("requirementsEnforced")
    if re_ is not None and str(re_).strip():
        parts.append(f"requirementsEnforced={quote(str(re_).strip(), safe='')}")

    loc = params.get("locale")
    if loc is not None and str(loc).strip():
        parts.append(f"locale={quote(str(loc).strip(), safe='')}")

    return "&".join(parts)


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: get_definitions_product_type.py '<JSON>'\n"
            "Required: sellerId (用于 storeTokens), region, productType, marketplaceIds (exactly one id)\n"
            "Optional: querySellerId → Query sellerId; productTypeVersion; requirements; "
            "requirementsEnforced; locale\n"
            "See https://developer-docs.amazon.com/sp-api/reference/getdefinitionsproducttype",
            file=sys.stderr,
        )
        sys.exit(1)
    try:
        params = json.loads(sys.argv[1])
    except json.JSONDecodeError as e:
        print(f"Invalid JSON: {e}", file=sys.stderr)
        sys.exit(1)

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

    for f in ("sellerId", "region", "productType", "marketplaceIds"):
        if f not in params:
            print(f"Missing required field: {f}", file=sys.stderr)
            sys.exit(1)

    seller_id = str(params["sellerId"])
    region = str(params["region"])
    product_type = str(params["productType"]).strip()
    if not product_type:
        print("productType must be non-empty", file=sys.stderr)
        sys.exit(1)

    try:
        query_string = _build_query(params)
    except ValueError as e:
        print(str(e), file=sys.stderr)
        sys.exit(1)

    path = _path_for_product_type(product_type)
    tokens = get_store_tokens(seller_id, region)
    if "error" in tokens or "accessToken" not in tokens:
        print(json.dumps(tokens, indent=2, ensure_ascii=False))
        sys.exit(1)

    proxy = developer_proxy_get(region, path, tokens["accessToken"], query_string)
    out = {
        "developerProxy": proxy,
        "resolvedPath": path,
        "queryString": query_string,
    }
    if proxy.get("errcode") == 200 and proxy.get("httpStatus") == 200:
        body_raw = proxy.get("body") or "{}"
        try:
            out["productTypeDefinitionResult"] = json.loads(body_raw)
        except json.JSONDecodeError:
            out["productTypeDefinitionResult"] = None
            out["productTypeDefinitionResultRaw"] = body_raw
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Amazon Store — Get Listings Item (SP-API Listings Items v2021-08-01)
====================================================================

通过 LinkFox 店铺网关 **POST /spApi/developerProxy** 转发 **GET getListingsItem**，
与 `linkfox-amazon-store-report` 使用同一套通用代理接口。

官方参考: https://developer-docs.amazon.com/sp-api/reference/getlistingsitem

Usage:
  python get_listings_item.py '{
    "sellerId": "A1BCDEFGHIJK2",
    "region": "NA",
    "sku": "My-Seller-SKU-001",
    "marketplaceIds": ["ATVPDKIKX0DER"]
  }'

Optional JSON fields:
  - includedData: string[] 或逗号分隔字符串；默认不传则上游通常为 summaries
  - issueLocale: 例如 en_US（可选）
  - skipDepCheck: boolean，跳过依赖检查（默认 false）

Dependency:
  `linkfox-amazon-store-auth`（与 store-report 相同；启动时运行 check_auth_dependency.py）
"""

from typing import List, Optional

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

LISTINGS_API_VERSION = "2021-08-01"

API_BASE_URL = os.environ.get("STORE_API_BASE_URL") or os.environ.get(
    "SPAPI_BASE_URL", "https://tool-gateway.linkfox.com"
)
STORE_TOKENS_ENDPOINT = f"{API_BASE_URL.rstrip('/')}/spApi/storeTokens"
DEVELOPER_PROXY_ENDPOINT = f"{API_BASE_URL.rstrip('/')}/spApi/developerProxy"

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


def ensure_auth_skill_available() -> None:
    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 get_listings_item.py",
        }
        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:  # pragma: no cover
        payload = {"missingSkill": REQUIRED_SKILL, "reason": str(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)


def get_api_key() -> str:
    key = os.environ.get("LINKFOXAGENT_API_KEY")
    if not key:
        print(
            "API Key not configured. Set:\n  export LINKFOXAGENT_API_KEY=<your-key>",
            file=sys.stderr,
        )
        sys.exit(1)
    return key


def call_api(endpoint: str, params: dict) -> dict:
    api_key = get_api_key()
    data = json.dumps(params).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 response:
            return json.loads(response.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_store_tokens(seller_id: str, region: str) -> dict:
    return call_api(STORE_TOKENS_ENDPOINT, {"sellerId": seller_id, "region": region})


def developer_proxy_get(
    region: str,
    path: str,
    access_token: str,
    query_string: Optional[str] = None,
) -> dict:
    params: dict = {
        "region": region,
        "path": path,
        "method": "GET",
        "amzAccessToken": access_token,
    }
    if query_string:
        params["queryString"] = query_string
    return call_api(DEVELOPER_PROXY_ENDPOINT, params)


def _path_for_get_listings_item(seller_id: str, sku: str) -> str:
    # Path segments must be percent-encoded (SKU 可含空格、斜杠等)
    enc_seller = quote(seller_id, safe="")
    enc_sku = quote(sku, safe="")
    return f"listings/{LISTINGS_API_VERSION}/items/{enc_seller}/{enc_sku}"


def _build_query_string(
    marketplace_ids: List[str],
    included_data: Optional[object],
    issue_locale: Optional[str],
) -> str:
    parts: list[str] = []
    # 官方要求 marketplaceIds；文档常见为单站点（length ≤ 1）。多传时仅取第一个并告警。
    if not marketplace_ids:
        raise ValueError("marketplaceIds must be a non-empty array of strings")
    mid = marketplace_ids[0]
    parts.append(f"marketplaceIds={quote(mid, safe='')}")
    if len(marketplace_ids) > 1:
        print(
            "⚠️  Warning: getListingsItem marketplaceIds length is typically 1; using first only.",
            file=sys.stderr,
        )
    if included_data is not None:
        if isinstance(included_data, list):
            id_val = ",".join(included_data)
        else:
            id_val = str(included_data).strip()
        if id_val:
            parts.append(f"includedData={quote(id_val, safe='')}")
    if issue_locale:
        parts.append(f"issueLocale={quote(issue_locale, safe='')}")
    return "&".join(parts)


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: get_listings_item.py '<JSON>'\n"
            "Required: sellerId, region, sku, marketplaceIds (array with ≥1 id)\n"
            "Example: get_listings_item.py "
            '\'{"sellerId":"A1...","region":"NA","sku":"ABC-1","marketplaceIds":["ATVPDKIKX0DER"]}\'',
            file=sys.stderr,
        )
        sys.exit(1)
    try:
        params = json.loads(sys.argv[1])
    except json.JSONDecodeError as e:
        print(f"Invalid JSON: {e}", file=sys.stderr)
        sys.exit(1)

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

    for f in ("sellerId", "region", "sku", "marketplaceIds"):
        if f not in params:
            print(f"Missing required field: {f}", file=sys.stderr)
            sys.exit(1)

    seller_id = str(params["sellerId"])
    region = str(params["region"])
    sku = str(params["sku"])
    mids = params["marketplaceIds"]
    if isinstance(mids, str):
        marketplace_ids = [mids]
    elif isinstance(mids, list):
        marketplace_ids = [str(x) for x in mids]
    else:
        print("marketplaceIds must be a string or array of strings", file=sys.stderr)
        sys.exit(1)

    included = params.get("includedData")
    issue_locale = params.get("issueLocale")
    if issue_locale is not None:
        issue_locale = str(issue_locale)

    try:
        query_string = _build_query_string(marketplace_ids, included, issue_locale)
    except ValueError as e:
        print(str(e), file=sys.stderr)
        sys.exit(1)

    path = _path_for_get_listings_item(seller_id, sku)

    tokens = get_store_tokens(seller_id, region)
    if "error" in tokens or "accessToken" not in tokens:
        print(json.dumps(tokens, indent=2, ensure_ascii=False))
        sys.exit(1)

    access_token = tokens["accessToken"]
    proxy = developer_proxy_get(region, path, access_token, query_string)

    out = {
        "developerProxy": proxy,
        "resolvedPath": path,
        "queryString": query_string,
    }
    if proxy.get("errcode") == 200 and proxy.get("httpStatus") == 200:
        body_raw = proxy.get("body") or "{}"
        try:
            out["listing"] = json.loads(body_raw)
        except json.JSONDecodeError:
            out["listing"] = None
            out["listingRaw"] = body_raw
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Amazon Store — Get Listings Restrictions (SP-API Listings Restrictions v2021-08-01)
==================================================================================

通过 LinkFox **POST /spApi/developerProxy** 转发 **GET getListingsRestrictions**，
查询目录 ASIN 在指定卖家与市场下的刊登限制。

官方参考: https://developer-docs.amazon.com/sp-api/reference/getlistingsrestrictions

Usage:
  python get_listings_restrictions.py '{
    "sellerId": "A1BCDEFGHIJK2",
    "region": "NA",
    "asin": "B08N5WRWNW",
    "marketplaceIds": ["ATVPDKIKX0DER"]
  }'

Optional JSON:
  conditionType (enum, 如 new / used_like_new 等), reasonLocale, skipDepCheck
"""

from typing import List

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

LISTINGS_API_VERSION = "2021-08-01"
RESTRICTIONS_PATH = f"listings/{LISTINGS_API_VERSION}/restrictions"

API_BASE_URL = os.environ.get("STORE_API_BASE_URL") or os.environ.get(
    "SPAPI_BASE_URL", "https://tool-gateway.linkfox.com"
)
STORE_TOKENS_ENDPOINT = f"{API_BASE_URL.rstrip('/')}/spApi/storeTokens"
DEVELOPER_PROXY_ENDPOINT = f"{API_BASE_URL.rstrip('/')}/spApi/developerProxy"

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


def ensure_auth_skill_available() -> None:
    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 get_listings_restrictions.py",
        }
        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:  # pragma: no cover
        payload = {"missingSkill": REQUIRED_SKILL, "reason": str(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)


def get_api_key() -> str:
    key = os.environ.get("LINKFOXAGENT_API_KEY")
    if not key:
        print(
            "API Key not configured. Set:\n  export LINKFOXAGENT_API_KEY=<your-key>",
            file=sys.stderr,
        )
        sys.exit(1)
    return key


def call_api(endpoint: str, params: dict) -> dict:
    api_key = get_api_key()
    data = json.dumps(params).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 response:
            return json.loads(response.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_store_tokens(seller_id: str, region: str) -> dict:
    return call_api(STORE_TOKENS_ENDPOINT, {"sellerId": seller_id, "region": region})


def developer_proxy_get(
    region: str,
    path: str,
    access_token: str,
    query_string: str,
) -> dict:
    params: dict = {
        "region": region,
        "path": path,
        "method": "GET",
        "amzAccessToken": access_token,
        "queryString": query_string,
    }
    return call_api(DEVELOPER_PROXY_ENDPOINT, params)


def _norm_marketplace_ids(val: object) -> List[str]:
    if isinstance(val, str):
        return [x.strip() for x in val.split(",") if x.strip()]
    if isinstance(val, list):
        return [str(x).strip() for x in val if str(x).strip()]
    raise ValueError("marketplaceIds must be a string or non-empty array")


def _build_query(params: dict, seller_id: str, asin: str) -> str:
    mids = _norm_marketplace_ids(params["marketplaceIds"])
    if not mids:
        raise ValueError("marketplaceIds is required")
    parts: List[str] = [
        f"asin={quote(asin, safe='')}",
        f"sellerId={quote(seller_id, safe='')}",
        f"marketplaceIds={quote(','.join(mids), safe='')}",
    ]
    ct = params.get("conditionType")
    if ct is not None and str(ct).strip():
        parts.append(f"conditionType={quote(str(ct).strip(), safe='')}")
    rl = params.get("reasonLocale")
    if rl is not None and str(rl).strip():
        parts.append(f"reasonLocale={quote(str(rl).strip(), safe='')}")
    return "&".join(parts)


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: get_listings_restrictions.py '<JSON>'\n"
            "Required: sellerId, region, asin, marketplaceIds\n"
            "Optional: conditionType, reasonLocale\n"
            "See https://developer-docs.amazon.com/sp-api/reference/getlistingsrestrictions",
            file=sys.stderr,
        )
        sys.exit(1)
    try:
        params = json.loads(sys.argv[1])
    except json.JSONDecodeError as e:
        print(f"Invalid JSON: {e}", file=sys.stderr)
        sys.exit(1)

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

    for f in ("sellerId", "region", "asin", "marketplaceIds"):
        if f not in params:
            print(f"Missing required field: {f}", file=sys.stderr)
            sys.exit(1)

    seller_id = str(params["sellerId"])
    region = str(params["region"])
    asin = str(params["asin"]).strip()
    if not asin:
        print("asin must be non-empty", file=sys.stderr)
        sys.exit(1)

    try:
        query_string = _build_query(params, seller_id, asin)
    except ValueError as e:
        print(str(e), file=sys.stderr)
        sys.exit(1)

    tokens = get_store_tokens(seller_id, region)
    if "error" in tokens or "accessToken" not in tokens:
        print(json.dumps(tokens, indent=2, ensure_ascii=False))
        sys.exit(1)

    proxy = developer_proxy_get(region, RESTRICTIONS_PATH, tokens["accessToken"], query_string)
    out = {
        "developerProxy": proxy,
        "resolvedPath": RESTRICTIONS_PATH,
        "queryString": query_string,
    }
    if proxy.get("errcode") == 200 and proxy.get("httpStatus") == 200:
        body_raw = proxy.get("body") or "{}"
        try:
            out["restrictionsResult"] = json.loads(body_raw)
        except json.JSONDecodeError:
            out["restrictionsResult"] = None
            out["restrictionsResultRaw"] = body_raw
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Amazon Store — Patch Listings Item (SP-API Listings Items v2021-08-01)
======================================================================

通过 LinkFox **POST /spApi/developerProxy** 转发 **PATCH patchListingsItem**，
与 get/search 相同：先 **storeTokens**，再带 **body**（JSON 字符串）调用代理。

官方参考: https://developer-docs.amazon.com/sp-api/reference/patchlistingsitem

Usage:
  python patch_listings_item.py '{
    "sellerId": "A1BCDEFGHIJK2",
    "region": "NA",
    "sku": "My-Seller-SKU-001",
    "marketplaceIds": ["ATVPDKIKX0DER"],
    "productType": "PRODUCT",
    "patches": [
      {"op": "replace", "path": "/attributes/item_name", "value": [{"value": "New Title", "marketplace_id": "ATVPDKIKX0DER"}]}
    ]
  }'

说明：path / value 须符合 Amazon 对该 productType 的 JSON Patch 与属性 schema；上例仅演示结构，请按实网/官方 schema 填写。

Optional JSON:
  includedData, mode (VALIDATION_PREVIEW), issueLocale, skipDepCheck
"""

from typing import Any, Dict, List, Optional

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

LISTINGS_API_VERSION = "2021-08-01"
PATCH_OPS = frozenset({"add", "replace", "merge", "delete"})

API_BASE_URL = os.environ.get("STORE_API_BASE_URL") or os.environ.get(
    "SPAPI_BASE_URL", "https://tool-gateway.linkfox.com"
)
STORE_TOKENS_ENDPOINT = f"{API_BASE_URL.rstrip('/')}/spApi/storeTokens"
DEVELOPER_PROXY_ENDPOINT = f"{API_BASE_URL.rstrip('/')}/spApi/developerProxy"

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


def ensure_auth_skill_available() -> None:
    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 patch_listings_item.py",
        }
        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:  # pragma: no cover
        payload = {"missingSkill": REQUIRED_SKILL, "reason": str(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)


def get_api_key() -> str:
    key = os.environ.get("LINKFOXAGENT_API_KEY")
    if not key:
        print(
            "API Key not configured. Set:\n  export LINKFOXAGENT_API_KEY=<your-key>",
            file=sys.stderr,
        )
        sys.exit(1)
    return key


def call_api(endpoint: str, params: dict) -> dict:
    api_key = get_api_key()
    data = json.dumps(params).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 response:
            return json.loads(response.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_store_tokens(seller_id: str, region: str) -> dict:
    return call_api(STORE_TOKENS_ENDPOINT, {"sellerId": seller_id, "region": region})


def developer_proxy_patch(
    region: str,
    path: str,
    access_token: str,
    query_string: str,
    body_obj: Dict[str, Any],
) -> dict:
    params: dict = {
        "region": region,
        "path": path,
        "method": "PATCH",
        "amzAccessToken": access_token,
        "queryString": query_string,
        "body": json.dumps(body_obj, ensure_ascii=False),
        "contentType": "application/json",
    }
    return call_api(DEVELOPER_PROXY_ENDPOINT, params)


def _path_patch(seller_id: str, sku: str) -> str:
    return (
        f"listings/{LISTINGS_API_VERSION}/items/"
        f"{quote(seller_id, safe='')}/{quote(sku, safe='')}"
    )


def _norm_marketplace_ids(val: object) -> List[str]:
    if isinstance(val, str):
        return [x.strip() for x in val.split(",") if x.strip()]
    if isinstance(val, list):
        return [str(x).strip() for x in val if str(x).strip()]
    raise ValueError("marketplaceIds must be a string or non-empty array")


def _build_patch_query(params: dict) -> str:
    mids = _norm_marketplace_ids(params["marketplaceIds"])
    if not mids:
        raise ValueError("marketplaceIds is required")
    parts: List[str] = [f"marketplaceIds={quote(','.join(mids), safe='')}"]

    inc = params.get("includedData")
    if inc is not None:
        if isinstance(inc, list):
            iv = ",".join(str(x) for x in inc)
        else:
            iv = str(inc).strip()
        if iv:
            parts.append(f"includedData={quote(iv, safe='')}")

    mode = params.get("mode")
    if mode is not None and str(mode).strip():
        parts.append(f"mode={quote(str(mode).strip(), safe='')}")

    il = params.get("issueLocale")
    if il is not None and str(il).strip():
        parts.append(f"issueLocale={quote(str(il).strip(), safe='')}")

    return "&".join(parts)


def _validate_patches(patches: object) -> List[Dict[str, Any]]:
    if not isinstance(patches, list) or len(patches) < 1:
        raise ValueError("patches must be a non-empty array (Amazon: length >= 1)")
    out: List[Dict[str, Any]] = []
    for i, p in enumerate(patches):
        if not isinstance(p, dict):
            raise ValueError(f"patches[{i}] must be an object")
        op = p.get("op")
        path_v = p.get("path")
        if not op or not path_v:
            raise ValueError(f"patches[{i}] requires 'op' and 'path'")
        if str(op) not in PATCH_OPS:
            raise ValueError(
                f"patches[{i}].op must be one of {sorted(PATCH_OPS)} (RFC 6902 subset per Amazon)"
            )
        out.append(p)
    return out


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: patch_listings_item.py '<JSON>'\n"
            "Required: sellerId, region, sku, marketplaceIds, productType, patches (array)\n"
            "Optional: includedData, mode (e.g. VALIDATION_PREVIEW), issueLocale\n"
            "See https://developer-docs.amazon.com/sp-api/reference/patchlistingsitem",
            file=sys.stderr,
        )
        sys.exit(1)
    try:
        params = json.loads(sys.argv[1])
    except json.JSONDecodeError as e:
        print(f"Invalid JSON: {e}", file=sys.stderr)
        sys.exit(1)

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

    for f in ("sellerId", "region", "sku", "marketplaceIds", "productType", "patches"):
        if f not in params:
            print(f"Missing required field: {f}", file=sys.stderr)
            sys.exit(1)

    seller_id = str(params["sellerId"])
    region = str(params["region"])
    sku = str(params["sku"])
    product_type = str(params["productType"])

    try:
        patches = _validate_patches(params["patches"])
        query_string = _build_patch_query(params)
    except ValueError as e:
        print(str(e), file=sys.stderr)
        sys.exit(1)

    body_obj: Dict[str, Any] = {"productType": product_type, "patches": patches}

    path = _path_patch(seller_id, sku)
    tokens = get_store_tokens(seller_id, region)
    if "error" in tokens or "accessToken" not in tokens:
        print(json.dumps(tokens, indent=2, ensure_ascii=False))
        sys.exit(1)

    access_token = tokens["accessToken"]
    proxy = developer_proxy_patch(region, path, access_token, query_string, body_obj)

    out: Dict[str, Any] = {
        "developerProxy": proxy,
        "resolvedPath": path,
        "queryString": query_string,
        "requestBody": body_obj,
    }
    if proxy.get("errcode") == 200 and proxy.get("httpStatus") == 200:
        body_raw = proxy.get("body") or "{}"
        try:
            out["patchResult"] = json.loads(body_raw)
        except json.JSONDecodeError:
            out["patchResult"] = None
            out["patchResultRaw"] = body_raw
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()