Linkfox Amazon Store Uploads

"""Shared helpers for linkfox-amazon-store-uploads (Uploads API v2020-11-01)."""

from __future__ import annotations

import base64
import hashlib
import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Iterable, Optional

from urllib.error import HTTPError, URLError
from urllib.parse import quote
from urllib.request import Request, urlopen

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

UPLOADS_API_VERSION = "2020-11-01"
UPLOADS_PATH_PREFIX = f"uploads/{UPLOADS_API_VERSION}"

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"

SUCCESS_HTTP_STATUSES = frozenset({200, 201})


def ensure_auth_skill_available(caller: str = "uploads 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, timeout: int = 120) -> 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=timeout) 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_post(
    region: str,
    path: str,
    access_token: str,
    *,
    query_string: Optional[str] = None,
    timeout: int = 120,
) -> dict:
    params: dict = {
        "region": region,
        "path": path,
        "method": "POST",
        "amzAccessToken": access_token,
        "contentType": "application/json",
    }
    if query_string:
        params["queryString"] = query_string
    return call_api(DEVELOPER_PROXY_ENDPOINT, params, timeout=timeout)


def path_for_upload_destination(resource: str) -> str:
    """resource 为下游 API 资源路径，如 aplus/2020-11-01/contentDocuments（勿带前导 /）。"""
    res = str(resource).strip().lstrip("/")
    enc = quote(res, safe="")
    return f"{UPLOADS_PATH_PREFIX}/uploadDestinations/{enc}"


def content_md5_base64(data: bytes) -> str:
    return base64.b64encode(hashlib.md5(data).digest()).decode("ascii")


def load_file_bytes(params: dict) -> bytes:
    if params.get("filePath"):
        p = Path(str(params["filePath"]).expanduser())
        if not p.is_file():
            print(f"filePath not found: {p}", file=sys.stderr)
            sys.exit(1)
        return p.read_bytes()
    if params.get("contentBase64"):
        try:
            return base64.b64decode(str(params["contentBase64"]))
        except Exception as e:
            print(f"Invalid contentBase64: {e}", file=sys.stderr)
            sys.exit(1)
    if "content" in params:
        raw = params["content"]
        if isinstance(raw, str):
            return raw.encode("utf-8")
        print("content must be a string", file=sys.stderr)
        sys.exit(1)
    return b""


def resolve_content_md5(params: dict) -> str:
    if params.get("contentMD5"):
        return str(params["contentMD5"]).strip()
    data = load_file_bytes(params)
    if not data:
        print(
            "Missing contentMD5, or provide filePath/content/contentBase64 to auto-compute.",
            file=sys.stderr,
        )
        sys.exit(1)
    return content_md5_base64(data)


def resolve_marketplace_id(params: dict) -> str:
    mid = params.get("marketplaceId")
    if mid is None and params.get("marketplaceIds") is not None:
        mids = params["marketplaceIds"]
        if isinstance(mids, list) and mids:
            mid = mids[0]
            if len(mids) > 1:
                print(
                    "Warning: createUploadDestinationForResource uses one marketplaceId; using first.",
                    file=sys.stderr,
                )
        elif isinstance(mids, str) and mids.strip():
            mid = mids.strip()
    if mid is None or not str(mid).strip():
        print("Missing marketplaceId (or marketplaceIds).", file=sys.stderr)
        sys.exit(1)
    return str(mid).strip()


def merge_success_json(
    out: dict,
    proxy: dict,
    result_key: str,
    *,
    success_http: Iterable[int] = SUCCESS_HTTP_STATUSES,
) -> None:
    if proxy.get("errcode") != 200:
        return
    try:
        status = int(proxy.get("httpStatus") or 0)
    except (TypeError, ValueError):
        return
    if status not in success_http:
        return
    body_raw = proxy.get("body")
    if body_raw is None or not str(body_raw).strip():
        out[result_key] = None
        return
    try:
        out[result_key] = json.loads(str(body_raw))
    except json.JSONDecodeError:
        out[result_key] = None
        out[f"{result_key}Raw"] = body_raw


def load_cli_params() -> dict:
    if len(sys.argv) < 2:
        return {}
    try:
        return json.loads(sys.argv[1])
    except json.JSONDecodeError as e:
        print(f"Invalid JSON: {e}", file=sys.stderr)
        sys.exit(1)


def require_seller_region(params: dict) -> tuple[str, str]:
    for f in ("sellerId", "region"):
        if f not in params:
            print(f"Missing required field: {f}", file=sys.stderr)
            sys.exit(1)
    return str(params["sellerId"]), str(params["region"])

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

与 `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-uploads 依赖 `{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 — createUploadDestinationForResource (Uploads API v2020-11-01)
==========================================================================

POST `uploads/2020-11-01/uploadDestinations/{resource}`，获取上传 URL 与 headers，
供后续 `upload_to_destination.py` 将文件 PUT 到 Amazon。

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

Usage:
  python create_upload_destination_for_resource.py '{
    "sellerId": "A1...",
    "region": "NA",
    "resource": "aplus/2020-11-01/contentDocuments",
    "marketplaceId": "ATVPDKIKX0DER",
    "filePath": "/path/to/image.jpg",
    "contentType": "image/jpeg"
  }'

也可直接传 contentMD5（Base64 MD5 摘要），与待上传字节一致。
"""

from __future__ import annotations

import json
import sys
from urllib.parse import quote

from _spapi_uploads_common import (
    developer_proxy_post,
    ensure_auth_skill_available,
    get_store_tokens,
    load_cli_params,
    merge_success_json,
    path_for_upload_destination,
    require_seller_region,
    resolve_content_md5,
    resolve_marketplace_id,
)


def _build_query(params: dict) -> str:
    mid = resolve_marketplace_id(params)
    md5 = resolve_content_md5(params)
    parts = [
        f"marketplaceIds={quote(mid, safe='')}",
        f"contentMD5={quote(md5, safe='')}",
    ]
    ct = params.get("contentType")
    if ct:
        parts.append(f"contentType={quote(str(ct).strip(), safe='')}")
    return "&".join(parts)


def main() -> None:
    params = load_cli_params()
    if not params:
        print(
            "Usage: create_upload_destination_for_resource.py '<JSON>'\n"
            "Required: sellerId, region, resource, marketplaceId, "
            "and contentMD5 OR filePath/content/contentBase64\n"
            "Optional: contentType",
            file=sys.stderr,
        )
        sys.exit(1)

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

    if "resource" not in params:
        print("Missing required field: resource", file=sys.stderr)
        sys.exit(1)

    seller_id, region = require_seller_region(params)
    path = path_for_upload_destination(params["resource"])
    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_post(region, path, tokens["accessToken"], query_string=qs)
    out: dict = {
        "developerProxy": proxy,
        "resolvedPath": path,
        "queryString": qs,
        "resource": str(params["resource"]).strip(),
    }
    merge_success_json(out, proxy, "uploadDestination")
    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 — PUT file to createUploadDestinationForResource URL
=================================================================

在调用 create_upload_destination_for_resource 之后，将文件上传到返回的 **url**，
并附带响应中的 **headers**（不经 developerProxy）。

Usage:
  python upload_to_destination.py '{
    "uploadUrl": "https://...",
    "headers": { "Content-Type": "image/jpeg", "Content-MD5": "..." },
    "filePath": "/path/to/image.jpg"
  }'

也可传 uploadDestination 对象（含 url、headers）：
  "uploadDestination": { "url": "...", "headers": { ... } }
"""

from __future__ import annotations

import json
import sys
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

from _spapi_uploads_common import load_cli_params, load_file_bytes


def _resolve_upload_target(params: dict) -> tuple[str, dict]:
    dest = params.get("uploadDestination")
    if isinstance(dest, dict):
        url = dest.get("url") or params.get("uploadUrl")
        headers = dest.get("headers") or params.get("headers") or {}
    else:
        url = params.get("uploadUrl")
        headers = params.get("headers") or {}
    if not url:
        print("Missing uploadUrl or uploadDestination.url", file=sys.stderr)
        sys.exit(1)
    if not isinstance(headers, dict):
        print("headers must be a JSON object", file=sys.stderr)
        sys.exit(1)
    return str(url).strip(), {str(k): str(v) for k, v in headers.items()}


def main() -> None:
    params = load_cli_params()
    if not params:
        print(
            "Usage: upload_to_destination.py '<JSON>'\n"
            "Required: uploadUrl (or uploadDestination), headers, "
            "and filePath | content | contentBase64",
            file=sys.stderr,
        )
        sys.exit(1)

    upload_url, hdrs = _resolve_upload_target(params)
    payload = load_file_bytes(params)
    if not payload:
        print("Provide filePath, content, or contentBase64", file=sys.stderr)
        sys.exit(1)

    req = Request(upload_url, data=payload, headers=hdrs, method="PUT")
    out: dict = {
        "uploadUrl": upload_url,
        "bytesUploaded": len(payload),
        "requestHeaders": hdrs,
    }
    try:
        with urlopen(req, timeout=300) as resp:
            out["httpStatus"] = resp.status
            out["success"] = 200 <= resp.status < 300
    except HTTPError as e:
        body = e.read().decode("utf-8", errors="replace") if e.fp else ""
        out["httpStatus"] = e.code
        out["success"] = False
        out["error"] = f"HTTP {e.code}: {e.reason}"
        out["details"] = body[:2000]
    except URLError as e:
        out["success"] = False
        out["error"] = f"Connection failed: {e.reason}"

    print(json.dumps(out, indent=2, ensure_ascii=False))
    if not out.get("success"):
        sys.exit(1)


if __name__ == "__main__":
    main()