Linkfox Amazon Store Orders

"""Shared helpers for linkfox-amazon-store-orders (storeTokens + developerProxy)."""

from __future__ import annotations

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

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(caller: str = "orders script") -> None:
    here = Path(__file__).resolve().parent
    checker = here / "check_auth_dependency.py"
    if not checker.exists():
        payload = {
            "missingSkill": REQUIRED_SKILL,
            "reason": f"check_auth_dependency.py not found next to {caller}",
        }
        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_call(
    region: str,
    path: str,
    method: str,
    access_token: str,
    query_string: Optional[str] = None,
    body: Optional[str] = None,
    content_type: str = "application/json",
) -> dict:
    params: dict[str, Any] = {
        "region": region,
        "path": path,
        "method": method,
        "amzAccessToken": access_token,
    }
    if query_string:
        params["queryString"] = query_string
    if body is not None:
        params["body"] = body
        params["contentType"] = content_type
    return call_api(DEVELOPER_PROXY_ENDPOINT, params)


def merge_json_body(out: dict, proxy: dict, key: str) -> None:
    """If proxy indicates HTTP 200 with JSON body, parse into out[key]."""
    if proxy.get("errcode") != 200 or int(proxy.get("httpStatus") or 0) != 200:
        return
    body_raw = proxy.get("body")
    if not body_raw or not str(body_raw).strip():
        out[key] = None
        return
    try:
        out[key] = json.loads(str(body_raw))
    except json.JSONDecodeError:
        out[key] = None
        out[f"{key}Raw"] = body_raw

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

与 `linkfox-amazon-store-listings` / `linkfox-amazon-store-pricing` 相同逻辑：
探测是否已安装 `linkfox-amazon-store-auth`。
"""

from __future__ import annotations

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

REQUIRED_SKILL = "linkfox-amazon-store-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]:
    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",
        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:
    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
        if category_dir.name.startswith(".") or category_dir.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:
    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:
    home = Path.home()
    for skill_dir_name in _AUTH_SKILL_DIR_ALIASES:
        for root in candidate_skill_roots():
            target = root / skill_dir_name / "SKILL.md"
            if target.is_file():
                return target
        found = _hermes_category_skill_md(home / ".hermes" / "skills", skill_dir_name)
        if found is not None:
            return found
        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
        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]:
    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-store-orders 依赖 `{REQUIRED_SKILL}`，"
            "但在常见 skill 安装路径下未找到其 SKILL.md。"
        ),
        "searchedRoots": searched_locations_for_report(),
        "suggestedActions": [
            f"If a skill installer tool is available, install '{REQUIRED_SKILL}' immediately.",
            "Otherwise ask the user to install from https://skill.linkfox.com/ and retry.",
        ],
        "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 — confirmShipment (SP-API Orders v0)
=================================================

POST `orders/v0/orders/{orderId}/shipmentConfirmation`，提交发货确认信息。

请将 Amazon 要求的完整 JSON 放在 **requestBody** 对象中（通常含 packageDetail 等字段），
脚本原样转发。

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

from __future__ import annotations

import json
import sys
from typing import Dict
from urllib.parse import quote

from _spapi_orders_common import (
    developer_proxy_call,
    ensure_auth_skill_available,
    get_store_tokens,
)


def _path(order_id: str) -> str:
    oid = quote(str(order_id).strip(), safe="")
    return f"orders/v0/orders/{oid}/shipmentConfirmation"


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: confirm_shipment.py '<JSON>'\n"
            "Required: sellerId, region, orderId, requestBody (object per Amazon confirmShipment schema)",
            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("confirm_shipment.py")

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

    rb = params["requestBody"]
    if not isinstance(rb, dict):
        print("requestBody must be a JSON object.", file=sys.stderr)
        sys.exit(1)

    path = _path(params["orderId"])
    body_str = json.dumps(rb, ensure_ascii=False)

    tokens = get_store_tokens(str(params["sellerId"]), str(params["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_call(
        str(params["region"]),
        path,
        "POST",
        tokens["accessToken"],
        body=body_str,
    )
    out: dict = {"developerProxy": proxy, "resolvedPath": path, "requestBody": rb}
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Amazon Store — getOrderAddress (SP-API Orders v0, deprecated)
=============================================================

GET `orders/v0/orders/{orderId}/address`（收货地址）。deprecated；可能需 RDT / 受限角色。

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

from __future__ import annotations

import json
import sys
from urllib.parse import quote

from _spapi_orders_common import (
    developer_proxy_call,
    ensure_auth_skill_available,
    get_store_tokens,
    merge_json_body,
)


def _path(order_id: str) -> str:
    oid = quote(str(order_id).strip(), safe="")
    return f"orders/v0/orders/{oid}/address"


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: get_order_address.py '<JSON>'\nRequired: sellerId, region, orderId",
            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("get_order_address.py")

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

    path = _path(params["orderId"])
    tokens = get_store_tokens(str(params["sellerId"]), str(params["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_call(str(params["region"]), path, "GET", tokens["accessToken"])
    out: dict = {"developerProxy": proxy, "resolvedPath": path}
    merge_json_body(out, proxy, "shippingAddress")
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Amazon Store — getOrderBuyerInfo (SP-API Orders v0, deprecated)
================================================================

GET `orders/v0/orders/{orderId}/buyerInfo`。官方标注 deprecated；买家 PII 可能需 RDT，以 Amazon 为准。

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

from __future__ import annotations

import json
import sys
from urllib.parse import quote

from _spapi_orders_common import (
    developer_proxy_call,
    ensure_auth_skill_available,
    get_store_tokens,
    merge_json_body,
)


def _path(order_id: str) -> str:
    oid = quote(str(order_id).strip(), safe="")
    return f"orders/v0/orders/{oid}/buyerInfo"


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: get_order_buyer_info.py '<JSON>'\nRequired: sellerId, region, orderId",
            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("get_order_buyer_info.py")

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

    path = _path(params["orderId"])
    tokens = get_store_tokens(str(params["sellerId"]), str(params["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_call(str(params["region"]), path, "GET", tokens["accessToken"])
    out: dict = {"developerProxy": proxy, "resolvedPath": path}
    merge_json_body(out, proxy, "buyerInfo")
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Amazon Store — getOrderItemsBuyerInfo (SP-API Orders v0, deprecated)
===================================================================

GET `orders/v0/orders/{orderId}/orderItems/buyerInfo`，可选 NextToken。

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

from __future__ import annotations

import json
import sys
from urllib.parse import quote

from _spapi_orders_common import (
    developer_proxy_call,
    ensure_auth_skill_available,
    get_store_tokens,
    merge_json_body,
)


def _path(order_id: str) -> str:
    oid = quote(str(order_id).strip(), safe="")
    return f"orders/v0/orders/{oid}/orderItems/buyerInfo"


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: get_order_items_buyer_info.py '<JSON>'\n"
            "Required: sellerId, region, orderId\nOptional: nextToken",
            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("get_order_items_buyer_info.py")

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

    path = _path(params["orderId"])
    qs = None
    nt = params.get("nextToken")
    if nt:
        qs = f"NextToken={quote(str(nt).strip(), safe='')}"

    tokens = get_store_tokens(str(params["sellerId"]), str(params["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_call(str(params["region"]), path, "GET", tokens["accessToken"], query_string=qs)
    out: dict = {"developerProxy": proxy, "resolvedPath": path, "queryString": qs}
    merge_json_body(out, proxy, "orderItemsBuyerInfo")
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Amazon Store — getOrderItems (SP-API Orders v0, deprecated)
===========================================================

GET `orders/v0/orders/{orderId}/orderItems`，可选 NextToken 分页。

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

from __future__ import annotations

import json
import sys
from urllib.parse import quote

from _spapi_orders_common import (
    developer_proxy_call,
    ensure_auth_skill_available,
    get_store_tokens,
    merge_json_body,
)


def _path(order_id: str) -> str:
    oid = quote(str(order_id).strip(), safe="")
    return f"orders/v0/orders/{oid}/orderItems"


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: get_order_items.py '<JSON>'\n"
            "Required: sellerId, region, orderId\nOptional: nextToken",
            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("get_order_items.py")

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

    path = _path(params["orderId"])
    qs = None
    nt = params.get("nextToken")
    if nt:
        qs = f"NextToken={quote(str(nt).strip(), safe='')}"

    tokens = get_store_tokens(str(params["sellerId"]), str(params["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_call(str(params["region"]), path, "GET", tokens["accessToken"], query_string=qs)
    out: dict = {"developerProxy": proxy, "resolvedPath": path, "queryString": qs}
    merge_json_body(out, proxy, "orderItems")
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Amazon Store — getOrderRegulatedInfo (SP-API Orders v0)
======================================================

GET `orders/v0/orders/{orderId}/regulatedInfo`（管制商品订单信息）。

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

from __future__ import annotations

import json
import sys
from urllib.parse import quote

from _spapi_orders_common import (
    developer_proxy_call,
    ensure_auth_skill_available,
    get_store_tokens,
    merge_json_body,
)


def _path(order_id: str) -> str:
    oid = quote(str(order_id).strip(), safe="")
    return f"orders/v0/orders/{oid}/regulatedInfo"


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: get_order_regulated_info.py '<JSON>'\nRequired: sellerId, region, orderId",
            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("get_order_regulated_info.py")

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

    path = _path(params["orderId"])
    tokens = get_store_tokens(str(params["sellerId"]), str(params["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_call(str(params["region"]), path, "GET", tokens["accessToken"])
    out: dict = {"developerProxy": proxy, "resolvedPath": path}
    merge_json_body(out, proxy, "regulatedOrder")
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Amazon Store — getOrder (SP-API Orders v2026-01-01)
==================================================

GET `orders/2026-01-01/orders/{orderId}`，可选 includedData 数据集。

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

from __future__ import annotations

import json
import sys
from typing import List
from urllib.parse import quote

from _spapi_orders_common import (
    developer_proxy_call,
    ensure_auth_skill_available,
    get_store_tokens,
    merge_json_body,
)


def _str_list(val: object, name: str) -> List[str]:
    if val is None:
        return []
    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()]
    print(f"{name} must be string or string[]", file=sys.stderr)
    sys.exit(1)


def _path(order_id: str) -> str:
    return f"orders/2026-01-01/orders/{quote(str(order_id).strip(), safe='')}"


def _query(included_data: List[str]) -> str | None:
    if not included_data:
        return None
    parts = [f"includedData={quote(x, safe='')}" for x in included_data]
    return "&".join(parts)


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: get_order.py '<JSON>'\n"
            'Required: sellerId, region, orderId\n'
            "Optional: includedData (e.g. BUYER,FULFILLMENT,PACKAGES), skipDepCheck",
            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("get_order.py")

    for f in ("sellerId", "region", "orderId"):
        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"])
    order_id = str(params["orderId"]).strip()
    path = _path(order_id)
    qs = _query(_str_list(params.get("includedData"), "includedData"))

    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_call(region, path, "GET", tokens["accessToken"], query_string=qs)
    out: dict = {"developerProxy": proxy, "resolvedPath": path, "queryString": qs}
    merge_json_body(out, proxy, "order")
    print(json.dumps(out, indent=2, ensure_ascii=False))


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 sys
if sys.version_info < (3, 10):
    sys.exit(
        "Error: Python 3.10+ required (current: "
        f"{sys.version_info.major}.{sys.version_info.minor}). "
        "Please upgrade Python."
    )

import argparse
import csv
import io
import json
import os
import re
import secrets
import subprocess
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
"""
Amazon Store — searchOrders (SP-API Orders v2026-01-01)
=======================================================

GET `orders/2026-01-01/orders`，按创建或更新时间窗检索订单。

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

Usage:
  python search_orders.py '{
    "sellerId":"A1...",
    "region":"NA",
    "marketplaceIds":["ATVPDKIKX0DER"],
    "lastUpdatedAfter":"2026-05-01T00:00:00Z"
  }'

必须二选一：createdAfter **或** lastUpdatedAfter（不可同时传另一组时间字段，规则见官方文档）。
可选：createdBefore、lastUpdatedBefore、fulfillmentStatuses、fulfilledBy、maxResultsPerPage、
paginationToken（上一页响应中的 nextToken）、includedData、skipDepCheck
"""

from __future__ import annotations

import json
import sys
from typing import List
from urllib.parse import quote

from _spapi_orders_common import (
    developer_proxy_call,
    ensure_auth_skill_available,
    get_store_tokens,
    merge_json_body,
)

PATH = "orders/2026-01-01/orders"


def _str_list(val: object, name: str) -> List[str]:
    if val is None:
        return []
    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()]
    print(f"{name} must be string or string[]", file=sys.stderr)
    sys.exit(1)


def _build_query(p: dict) -> str:
    ca = p.get("createdAfter")
    cb = p.get("createdBefore")
    lua = p.get("lastUpdatedAfter")
    lub = p.get("lastUpdatedBefore")
    if ca and lua:
        print("Provide only one of createdAfter or lastUpdatedAfter.", file=sys.stderr)
        sys.exit(1)
    if not ca and not lua:
        print("Required: createdAfter OR lastUpdatedAfter (ISO 8601).", file=sys.stderr)
        sys.exit(1)
    if ca and (lua or lub):
        print("When using createdAfter, do not use lastUpdatedAfter/lastUpdatedBefore.", file=sys.stderr)
        sys.exit(1)
    if lua and (ca or cb):
        print("When using lastUpdatedAfter, do not use createdAfter/createdBefore.", file=sys.stderr)
        sys.exit(1)

    parts: list[str] = []

    def add(k: str, v: str) -> None:
        parts.append(f"{k}={quote(v, safe='')}")

    if ca:
        add("createdAfter", str(ca).strip())
    if cb:
        add("createdBefore", str(cb).strip())
    if lua:
        add("lastUpdatedAfter", str(lua).strip())
    if lub:
        add("lastUpdatedBefore", str(lub).strip())

    for mid in _str_list(p.get("marketplaceIds"), "marketplaceIds"):
        add("marketplaceIds", mid)
    for fs in _str_list(p.get("fulfillmentStatuses"), "fulfillmentStatuses"):
        add("fulfillmentStatuses", fs)
    for fb in _str_list(p.get("fulfilledBy"), "fulfilledBy"):
        add("fulfilledBy", fb)
    for inc in _str_list(p.get("includedData"), "includedData"):
        add("includedData", inc)

    mrp = p.get("maxResultsPerPage")
    if mrp is not None:
        add("maxResultsPerPage", str(int(mrp)))

    pt = p.get("paginationToken")
    if pt:
        add("paginationToken", str(pt).strip())

    return "&".join(parts)


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: search_orders.py '<JSON>'\n"
            "Required: sellerId, region, marketplaceIds, "
            "and either createdAfter or lastUpdatedAfter.",
            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("search_orders.py")

    for f in ("sellerId", "region", "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"])
    qs = _build_query(params)

    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_call(region, PATH, "GET", tokens["accessToken"], query_string=qs)
    out: dict = {"developerProxy": proxy, "resolvedPath": PATH, "queryString": qs}
    merge_json_body(out, proxy, "searchOrders")
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Amazon Store — updateShipmentStatus (SP-API Orders v0)
======================================================

POST `orders/v0/orders/{orderId}/shipment`，更新自提等场景的发货状态。

Body 必填字段：marketplaceId、shipmentStatus（ReadyForPickup | PickedUp | RefusedPickup）；
可选 orderItems（部分行更新时的行与数量）。

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

from __future__ import annotations

import json
import sys
from typing import Any, Dict
from urllib.parse import quote

from _spapi_orders_common import (
    developer_proxy_call,
    ensure_auth_skill_available,
    get_store_tokens,
)

SHIPMENT_STATUSES = frozenset({"ReadyForPickup", "PickedUp", "RefusedPickup"})


def _path(order_id: str) -> str:
    oid = quote(str(order_id).strip(), safe="")
    return f"orders/v0/orders/{oid}/shipment"


def _build_body(params: dict) -> Dict[str, Any]:
    mid = params.get("marketplaceId")
    st = params.get("shipmentStatus")
    if not mid or not st:
        print("Missing marketplaceId or shipmentStatus.", file=sys.stderr)
        sys.exit(1)
    st_s = str(st).strip()
    if st_s not in SHIPMENT_STATUSES:
        print(f"shipmentStatus must be one of: {sorted(SHIPMENT_STATUSES)}", file=sys.stderr)
        sys.exit(1)
    body: Dict[str, Any] = {"marketplaceId": str(mid).strip(), "shipmentStatus": st_s}
    oi = params.get("orderItems")
    if oi is not None:
        if not isinstance(oi, list):
            print("orderItems must be an array when provided.", file=sys.stderr)
            sys.exit(1)
        body["orderItems"] = oi
    return body


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: update_shipment_status.py '<JSON>'\n"
            "Required: sellerId, region, orderId, marketplaceId, shipmentStatus\n"
            "Optional: orderItems, skipDepCheck",
            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("update_shipment_status.py")

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

    path = _path(params["orderId"])
    body_obj = _build_body(params)
    body_str = json.dumps(body_obj, ensure_ascii=False)

    tokens = get_store_tokens(str(params["sellerId"]), str(params["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_call(
        str(params["region"]),
        path,
        "POST",
        tokens["accessToken"],
        body=body_str,
    )
    out: dict = {
        "developerProxy": proxy,
        "resolvedPath": path,
        "requestBody": body_obj,
    }
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Amazon Store — updateVerificationStatus (SP-API Orders v0)
===========================================================

PATCH `orders/v0/orders/{orderId}/regulatedInfo`，批准或拒绝管制订单核验状态。

请求体须包含 regulatedOrderVerificationStatus 对象（结构以官方 schema 为准）。

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

from __future__ import annotations

import json
import sys
from typing import Any, Dict
from urllib.parse import quote

from _spapi_orders_common import (
    developer_proxy_call,
    ensure_auth_skill_available,
    get_store_tokens,
)


def _path(order_id: str) -> str:
    oid = quote(str(order_id).strip(), safe="")
    return f"orders/v0/orders/{oid}/regulatedInfo"


def main() -> None:
    if len(sys.argv) < 2:
        print(
            "Usage: update_verification_status.py '<JSON>'\n"
            "Required: sellerId, region, orderId, regulatedOrderVerificationStatus (object)\n"
            "Or pass full Amazon body as requestBody (object).",
            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("update_verification_status.py")

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

    body_obj: Dict[str, Any]
    if "requestBody" in params and isinstance(params["requestBody"], dict):
        body_obj = params["requestBody"]
    elif "regulatedOrderVerificationStatus" in params:
        body_obj = {"regulatedOrderVerificationStatus": params["regulatedOrderVerificationStatus"]}
    else:
        print("Provide regulatedOrderVerificationStatus or requestBody object.", file=sys.stderr)
        sys.exit(1)

    path = _path(params["orderId"])
    body_str = json.dumps(body_obj, ensure_ascii=False)

    tokens = get_store_tokens(str(params["sellerId"]), str(params["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_call(
        str(params["region"]),
        path,
        "PATCH",
        tokens["accessToken"],
        body=body_str,
    )
    out: dict = {
        "developerProxy": proxy,
        "resolvedPath": path,
        "requestBody": body_obj,
    }
    print(json.dumps(out, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()