Linkfox Temu Ads Global

#!/usr/bin/env python3
"""Factory for thin US Ads API CLI scripts."""

import sys

import _temu_global_common  # noqa: F401 — gateway helpers in scripts/

from _temu_common import load_json_arg
from _temu_global_common import extract_business_params, run_and_print

def run_cli(api_type: str, usage: str) -> None:
    if len(sys.argv) < 2:
        print(f"Usage: {usage}", file=sys.stderr)
        sys.exit(1)
    params = load_json_arg(sys.argv)
    run_and_print(params, api_type, extract_business_params(params))

#!/usr/bin/env python3
"""Authorization guide data derived from TEMU_API_README.md."""

VALID_SHOP_TYPES = frozenset({"semi-managed", "full-managed", "local-native"})
VALID_TOKEN_PURPOSES = frozenset(
    {"product-inventory", "order-shipping", "full-managed", "local-native", "default"}
)

LOGIN_URLS = {
    "cn": "https://seller.kuajingmaihuo.com",
    "partner": "https://agentseller.temu.com",
}

APPS = {
    "kuniao-seller": "酷鸟卖家助手",
    "cyber-erp": "Cyber-ERP酷鸟助手",
    "cyber-erp-native": "Cyber-ERP",
}

def _steps(*items: str) -> list[str]:
    return list(items)

def build_guide(shop_type: str, token_purpose: str, site=None) -> dict:
    shop_type = shop_type.strip().lower()
    token_purpose = (token_purpose or "default").strip().lower()

    if shop_type not in VALID_SHOP_TYPES:
        raise ValueError(
            f"Invalid shopType '{shop_type}'. Use: semi-managed, full-managed, local-native"
        )
    if token_purpose not in VALID_TOKEN_PURPOSES:
        raise ValueError(
            "Invalid tokenPurpose. Use: product-inventory, order-shipping, "
            "full-managed, local-native, default"
        )

    guide = {
        "shopType": shop_type,
        "tokenPurpose": token_purpose,
        "note": (
            "Temu accessToken 需在卖家后台手动复制，无自动 OAuth。"
            "保存后可使用 save_temu_access_token.py / storeKey 调用 API。"
        ),
    }

    if shop_type == "semi-managed" and token_purpose == "product-inventory":
        site = site or "cn"
        login = LOGIN_URLS.get(site, LOGIN_URLS["cn"])
        guide.update(
            {
                "recommendedSite": site,
                "recommendedManagementType": "semi-managed",
                "loginUrl": login,
                "appName": APPS["kuniao-seller"],
                "menuPath": "系统管理 > 服务市场 > 授权管理",
                "steps": _steps(
                    f"登录 Temu 卖家后台：{login}",
                    "进入：系统管理 > 服务市场 > 授权管理",
                    "点击「获取授权」，选择「酷鸟卖家助手」",
                    "全选常规和特殊授权 > 确认",
                    "复制 access_token，用 save_temu_access_token.py 保存",
                ),
                "usageHint": "用于半托管商品、库存类 API；site 通常为 cn 或 partner。",
            }
        )
    elif shop_type == "semi-managed" and token_purpose == "order-shipping":
        site = site or "us"
        guide.update(
            {
                "recommendedSite": site,
                "recommendedManagementType": "semi-managed",
                "loginUrl": LOGIN_URLS["cn"],
                "appName": APPS["cyber-erp"],
                "menuPath": "Seller Central 对应站点 > 服务市场 > 授权管理",
                "steps": _steps(
                    "登录 Temu 卖家后台（seller.kuajingmaihuo.com 或 agentseller.temu.com）",
                    "右上角 Seller Central，切换到目标区域站点（美区/欧区/全球）",
                    "进入：服务市场 > 授权管理",
                    "点击「获取授权」，选择「Cyber-ERP酷鸟助手」",
                    "全选常规和特殊授权 > 确认",
                    "复制 access_token；调用 API 时 site 用 us / global / eu",
                ),
                "usageHint": "用于半托管订单、发货类 API。",
            }
        )
    elif shop_type == "full-managed" or token_purpose == "full-managed":
        site = site or "cn"
        login = LOGIN_URLS.get(site, LOGIN_URLS["cn"])
        guide.update(
            {
                "recommendedSite": site,
                "recommendedManagementType": "full-managed",
                "loginUrl": login,
                "appName": APPS["kuniao-seller"],
                "menuPath": "系统管理 > 服务市场 > 授权管理",
                "steps": _steps(
                    f"登录 Temu 平台：{login}",
                    "系统管理 > 服务市场 > 授权管理",
                    "获取授权 > 选择「酷鸟卖家助手」",
                    "全选常规和特殊授权 > 确认 > 复制 access_token",
                ),
                "usageHint": "全托管店铺通用；site 通常为 cn 或 partner。",
            }
        )
    elif shop_type == "local-native" or token_purpose == "local-native":
        guide.update(
            {
                "recommendedSite": site or "us",
                "recommendedManagementType": "semi-managed",
                "loginUrl": LOGIN_URLS["cn"],
                "appName": APPS["cyber-erp-native"],
                "menuPath": "Apps And Services > Manage Your Apps",
                "steps": _steps(
                    "登录 Temu 卖家后台",
                    "进入：Apps And Services（应用程序和服务）",
                    "Manage Your Apps（管理您的应用）",
                    "Authorize a new app（授权新应用）",
                    "搜索「Cyber-ERP」",
                    "一般权限和敏感权限全选 > 确认 > 复制 access_token",
                ),
                "usageHint": "适用于美区、欧区本土主体店铺。",
            }
        )
    else:
        guide.update(
            {
                "steps": _steps(
                    "确认店铺类型：semi-managed / full-managed / local-native",
                    "确认用途：product-inventory（商品库存）或 order-shipping（订单发货）",
                    "运行 temu_token_guide.py 获取对应步骤",
                    "复制 token 后执行 save_temu_access_token.py",
                ),
                "usageHint": "请指定 shopType 与 tokenPurpose，见 references/access-token.md",
            }
        )

    guide["cautions"] = [
        "子账号可能无法进入服务市场，需主账号授权",
        "多站点需分别登录对应后台获取 Token",
        "Token 有有效期，过期后需重新在后台复制",
        "调用网关前需联系紫鸟配置 IP 白名单",
    ]
    return guide

#!/usr/bin/env python3
"""Shared helpers for LinkFox Temu API skill scripts."""

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

from _temu_token_store import get_token

BASE_URL = os.environ.get("TEMU_API_BASE_URL") or os.environ.get(
    "STORE_API_BASE_URL", "https://tool-gateway.linkfox.com"
)
BASE_URL = BASE_URL.rstrip("/")
PROXY_URL = f"{BASE_URL}/temu/proxy"
FILE_DOWNLOAD_URL = f"{BASE_URL}/temu/fileDownload"

VALID_SITES = frozenset({"cn", "partner", "us", "global", "eu"})
VALID_MANAGEMENT_TYPES = frozenset({"full-managed", "semi-managed"})

# LinkFox 用户 Token（网关鉴权），勿与 Body 中的 Temu accessToken 混淆
LINKFOX_TOKEN_PARAM_KEYS = ("token", "linkfoxToken", "linkfox_token")

def get_linkfox_token(params=None) -> str:
    """
    LinkFox 用户鉴权 Token，与 linkfox-amazon-store-auth 一致。
    优先级：请求 JSON 中的 token / linkfoxToken > 环境变量 LINKFOXAGENT_API_KEY。
    """
    if params:
        for key in LINKFOX_TOKEN_PARAM_KEYS:
            value = params.get(key)
            if value is not None and str(value).strip():
                return str(value).strip()
    key = os.environ.get("LINKFOXAGENT_API_KEY")
    if not key:
        print(
            "LinkFox user Token not configured. Same as linkfox-amazon-store-auth:\n"
            "1. Visit https://skill.linkfox.com/linkfoxskills/guide.htm to obtain your Key\n"
            "2. export LINKFOXAGENT_API_KEY=your-key-here\n"
            "   Or pass \"token\" in the JSON parameters of proxy/fileDownload scripts.",
            file=sys.stderr,
        )
        sys.exit(1)
    return key

def build_gateway_headers(linkfox_token: str) -> dict:
    """网关鉴权：Authorization（全站通用）+ Token（TEMU_API_SPEC 约定）。"""
    return {
        "Authorization": linkfox_token,
        "Token": linkfox_token,
        "Content-Type": "application/json",
        "User-Agent": "LinkFox-Skill/1.0",
    }

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

def require_text(params: dict, key: str, label=None) -> str:
    value = params.get(key)
    if value is None or not str(value).strip():
        print(f"Error: '{label or key}' is required.", file=sys.stderr)
        sys.exit(1)
    return str(value).strip()

def validate_site(site: str) -> str:
    if site not in VALID_SITES:
        print(
            f"Error: invalid site '{site}'. Must be one of: {', '.join(sorted(VALID_SITES))}",
            file=sys.stderr,
        )
        sys.exit(1)
    return site

def validate_management_type(management_type: str) -> str:
    if management_type not in VALID_MANAGEMENT_TYPES:
        print(
            "Error: invalid managementType. Must be: full-managed, semi-managed",
            file=sys.stderr,
        )
        sys.exit(1)
    return management_type

def call_temu_api(
    url: str,
    body: dict,
    timeout: int = 60,
    linkfox_params=None,
) -> dict:
    """调用 Temu 网关接口；必须先具备 LinkFox 用户 Token。"""
    linkfox_token = get_linkfox_token(linkfox_params)
    data = json.dumps(body, ensure_ascii=False).encode("utf-8")
    req = Request(
        url,
        data=data,
        headers=build_gateway_headers(linkfox_token),
        method="POST",
    )
    try:
        with urlopen(req, timeout=timeout) as response:
            return json.loads(response.read().decode("utf-8"))
    except HTTPError as e:
        raw = e.read().decode("utf-8") if e.fp else ""
        try:
            return json.loads(raw)
        except json.JSONDecodeError:
            return {"error": f"HTTP {e.code}: {e.reason}", "details": raw}
    except URLError as e:
        return {"error": f"Connection failed: {e.reason}"}

def is_linkfox_auth_error(result: dict) -> bool:
    msg = str(result.get("message") or result.get("error") or "")
    return "无法识别当前用户" in msg or "重新登录" in msg

def resolve_access_token(params: dict) -> str:
    """Temu 店铺 accessToken：直接传入或从本地 storeKey 读取。"""
    if params.get("accessToken"):
        return str(params["accessToken"]).strip()
    store_key = params.get("storeKey")
    if not store_key:
        print(
            "Error: provide Temu 'accessToken' or 'storeKey' (+ site, managementType).",
            file=sys.stderr,
        )
        sys.exit(1)
    site = validate_site(require_text(params, "site"))
    management_type = validate_management_type(require_text(params, "managementType"))
    token_purpose = str(params.get("tokenPurpose", "default")).strip() or "default"
    token = get_token(str(store_key).strip(), site, management_type, token_purpose)
    if not token:
        print(
            f"Error: no Temu token for storeKey={store_key}, site={site}, "
            f"managementType={management_type}, tokenPurpose={token_purpose}. "
            "Run temu_token_guide.py and save_temu_access_token.py first.",
            file=sys.stderr,
        )
        sys.exit(1)
    return token

def parse_nested_body(result: dict) -> dict:
    """If gateway returns a JSON string in body, parse it into temuBody."""
    body = result.get("body")
    if isinstance(body, str) and body.strip():
        try:
            result["temuBody"] = json.loads(body)
        except json.JSONDecodeError:
            pass
    return result

#!/usr/bin/env python3
"""Global / Partner Ads API helpers; local gateway helpers."""

import json

from _temu_common import (
    FILE_DOWNLOAD_URL,
    PROXY_URL,
    call_temu_api,
    load_json_arg,
    parse_nested_body,
    resolve_access_token,
)

# Partner Global 广告：默认全球站半托管 + 卖家助手 token
DEFAULT_SITE = "global"
DEFAULT_MANAGEMENT_TYPE = "semi-managed"
DEFAULT_TOKEN_PURPOSE = "product-inventory"

RESERVED_KEYS = frozenset(
    {
        "token",
        "linkfoxToken",
        "linkfox_token",
        "accessToken",
        "storeKey",
        "site",
        "managementType",
        "tokenPurpose",
        "type",
        "params",
    }
)

def extract_business_params(params: dict) -> dict:
    if isinstance(params.get("params"), dict):
        return dict(params["params"])
    return {k: v for k, v in params.items() if k not in RESERVED_KEYS}

def build_global_proxy_body(params: dict, api_type: str, business=None) -> dict:
    site = str(params.get("site", DEFAULT_SITE)).strip().lower() or DEFAULT_SITE
    management_type = (
        str(params.get("managementType", DEFAULT_MANAGEMENT_TYPE)).strip().lower()
        or DEFAULT_MANAGEMENT_TYPE
    )
    if "tokenPurpose" not in params and params.get("storeKey") and not params.get("accessToken"):
        params = dict(params)
        params.setdefault("tokenPurpose", DEFAULT_TOKEN_PURPOSE)
        params.setdefault("site", site)
        params.setdefault("managementType", management_type)

    body = {
        "site": site,
        "managementType": management_type,
        "accessToken": resolve_access_token(params),
        "type": api_type,
    }
    biz = business if business is not None else extract_business_params(params)
    if biz:
        body["params"] = biz
    return body

def global_proxy_call(params: dict, api_type: str, business=None, timeout: int = 60) -> dict:
    body = build_global_proxy_body(params, api_type, business)
    return call_temu_api(PROXY_URL, body, timeout=timeout, linkfox_params=params)

def global_file_download_call(params: dict, timeout: int = 120) -> dict:
    if "tokenPurpose" not in params and params.get("storeKey") and not params.get("accessToken"):
        params = dict(params)
        params.setdefault("tokenPurpose", DEFAULT_TOKEN_PURPOSE)

    from _temu_common import require_text  # noqa: E402

    site = str(params.get("site", DEFAULT_SITE)).strip().lower() or DEFAULT_SITE
    management_type = (
        str(params.get("managementType", DEFAULT_MANAGEMENT_TYPE)).strip().lower()
        or DEFAULT_MANAGEMENT_TYPE
    )
    body = {
        "site": site,
        "managementType": management_type,
        "accessToken": resolve_access_token(params),
        "url": require_text(params, "url"),
    }
    return call_temu_api(FILE_DOWNLOAD_URL, body, timeout=timeout, linkfox_params=params)

def run_and_print(params: dict, api_type: str, business=None) -> None:
    result = parse_nested_body(global_proxy_call(params, api_type, business))
    print(json.dumps(result, indent=2, ensure_ascii=False))

#!/usr/bin/env python3
"""Local Temu accessToken storage (manual authorization from seller console)."""

from __future__ import annotations

import json
import os
from datetime import datetime, timezone

DEFAULT_STORE_PATH = os.path.expanduser("~/.linkfox/temu-access-tokens.json")

def store_path() -> str:
    return os.environ.get("TEMU_TOKEN_STORE_PATH", DEFAULT_STORE_PATH)

def _utc_now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

def _load_store() -> dict:
    path = store_path()
    if not os.path.isfile(path):
        return {"stores": []}
    with open(path, encoding="utf-8") as f:
        data = json.load(f)
    if "stores" not in data or not isinstance(data["stores"], list):
        return {"stores": []}
    return data

def _save_store(data: dict) -> None:
    path = store_path()
    parent = os.path.dirname(path)
    if parent:
        os.makedirs(parent, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)
        f.write("\n")

def _find_store(data: dict, store_key: str):
    for store in data["stores"]:
        if store.get("storeKey") == store_key:
            return store
    return None

def _token_key(site: str, management_type: str, token_purpose: str) -> tuple:
    return site, management_type, token_purpose or "default"

def save_token(
    store_key: str,
    site: str,
    management_type: str,
    access_token: str,
    token_purpose: str = "default",
    label: str | None = None,
) -> dict:
    data = _load_store()
    store = _find_store(data, store_key)
    if store is None:
        store = {"storeKey": store_key, "label": label or store_key, "tokens": []}
        data["stores"].append(store)
    elif label:
        store["label"] = label

    key = _token_key(site, management_type, token_purpose)
    entry = {
        "site": site,
        "managementType": management_type,
        "tokenPurpose": token_purpose,
        "accessToken": access_token,
        "updatedAt": _utc_now(),
    }
    replaced = False
    for i, item in enumerate(store["tokens"]):
        if _token_key(
            item.get("site", ""),
            item.get("managementType", ""),
            item.get("tokenPurpose", "default"),
        ) == key:
            store["tokens"][i] = entry
            replaced = True
            break
    if not replaced:
        store["tokens"].append(entry)

    _save_store(data)
    return {
        "storeKey": store_key,
        "site": site,
        "managementType": management_type,
        "tokenPurpose": token_purpose,
        "storePath": store_path(),
        "updatedAt": entry["updatedAt"],
        "replaced": replaced,
    }

def get_token(
    store_key: str,
    site: str,
    management_type: str,
    token_purpose: str = "default",
):
    data = _load_store()
    store = _find_store(data, store_key)
    if not store:
        return None
    key = _token_key(site, management_type, token_purpose)
    for item in store.get("tokens", []):
        if _token_key(
            item.get("site", ""),
            item.get("managementType", ""),
            item.get("tokenPurpose", "default"),
        ) == key:
            return item.get("accessToken")
    return None

def list_stores(mask: bool = True) -> dict:
    data = _load_store()
    stores = []
    for store in data.get("stores", []):
        tokens = []
        for item in store.get("tokens", []):
            token = item.get("accessToken", "")
            if mask and token:
                shown = token[:6] + "..." + token[-4:] if len(token) > 12 else "***"
            else:
                shown = token
            tokens.append(
                {
                    "site": item.get("site"),
                    "managementType": item.get("managementType"),
                    "tokenPurpose": item.get("tokenPurpose", "default"),
                    "accessToken": shown,
                    "updatedAt": item.get("updatedAt"),
                }
            )
        stores.append(
            {
                "storeKey": store.get("storeKey"),
                "label": store.get("label"),
                "tokens": tokens,
            }
        )
    return {"storePath": store_path(), "stores": stores}

#!/usr/bin/env python3
"""
Check LinkFox user Token before calling Temu gateway APIs.
Same role as LINKFOXAGENT_API_KEY in linkfox-amazon-store-auth.

Usage:
  python check_linkfox_token.py
  python check_linkfox_token.py '{"token": "your-linkfox-key"}'
"""

import json
import sys

from _temu_common import PROXY_URL, call_temu_api, get_linkfox_token, is_linkfox_auth_error, load_json_arg

def main():
    params = load_json_arg(sys.argv) if len(sys.argv) >= 2 else {}
    linkfox_token = get_linkfox_token(params)
    masked = (
        linkfox_token[:6] + "..." + linkfox_token[-4:]
        if len(linkfox_token) > 12
        else "***"
    )

    # 最小探测请求：LinkFox Token 有效时应返回业务/参数类 1002，而非「无法识别当前用户」
    probe_body = {
        "site": "cn",
        "managementType": "full-managed",
        "accessToken": "probe",
        "type": "probe",
    }
    result = call_temu_api(PROXY_URL, probe_body, timeout=30, linkfox_params=params)

    if is_linkfox_auth_error(result):
        print(
            json.dumps(
                {
                    "ok": False,
                    "reason": "linkfox_token_invalid",
                    "message": result.get("message") or result,
                    "hint": "Set LINKFOXAGENT_API_KEY or pass token in JSON.",
                },
                indent=2,
                ensure_ascii=False,
            )
        )
        sys.exit(1)

    print(
        json.dumps(
            {
                "ok": True,
                "linkfoxToken": masked,
                "gatewayReachable": True,
                "note": "LinkFox user Token accepted by gateway.",
            },
            indent=2,
            ensure_ascii=False,
        )
    )

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Get Temu accessToken from local store.

Usage:
  python get_temu_access_token.py '{
    "storeKey": "my-shop",
    "site": "cn",
    "managementType": "semi-managed",
    "tokenPurpose": "product-inventory"
  }'
"""

import json
import sys

from _temu_common import load_json_arg, require_text, validate_management_type, validate_site
from _temu_token_store import get_token

def main():
    if len(sys.argv) < 2:
        print("Usage: get_temu_access_token.py '<JSON parameters>'", file=sys.stderr)
        sys.exit(1)

    params = load_json_arg(sys.argv)
    store_key = require_text(params, "storeKey")
    site = validate_site(require_text(params, "site"))
    management_type = validate_management_type(require_text(params, "managementType"))
    token_purpose = str(params.get("tokenPurpose", "default")).strip() or "default"

    token = get_token(store_key, site, management_type, token_purpose)
    if not token:
        print(
            json.dumps(
                {
                    "found": False,
                    "storeKey": store_key,
                    "site": site,
                    "managementType": management_type,
                    "tokenPurpose": token_purpose,
                    "hint": "Run temu_token_guide.py then save_temu_access_token.py",
                },
                indent=2,
                ensure_ascii=False,
            )
        )
        sys.exit(1)

    print(
        json.dumps(
            {
                "found": True,
                "storeKey": store_key,
                "site": site,
                "managementType": management_type,
                "tokenPurpose": token_purpose,
                "accessToken": token,
            },
            indent=2,
            ensure_ascii=False,
        )
    )

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""temu.searchrec.ad.create — 创建广告."""
import _global_ads_script as m

if __name__ == "__main__":
    m.run_cli(
        "temu.searchrec.ad.create",
        "global_ads_searchrec_ad_create.py '<JSON>'",
    )

#!/usr/bin/env python3
"""temu.searchrec.ad.detail.query — 广告详情查询."""
import _global_ads_script as m

if __name__ == "__main__":
    m.run_cli(
        "temu.searchrec.ad.detail.query",
        "global_ads_searchrec_ad_detail_query.py '<JSON>'",
    )

#!/usr/bin/env python3
"""temu.searchrec.ad.goods.create.query — 广告可创建商品查询."""
import _global_ads_script as m

if __name__ == "__main__":
    m.run_cli(
        "temu.searchrec.ad.goods.create.query",
        "global_ads_searchrec_ad_goods_create_query.py '<JSON>'",
    )

#!/usr/bin/env python3
"""temu.searchrec.ad.log.query — 广告操作日志查询."""
import _global_ads_script as m

if __name__ == "__main__":
    m.run_cli(
        "temu.searchrec.ad.log.query",
        "global_ads_searchrec_ad_log_query.py '<JSON>'",
    )

#!/usr/bin/env python3
"""temu.searchrec.ad.modify — 修改广告."""
import _global_ads_script as m

if __name__ == "__main__":
    m.run_cli(
        "temu.searchrec.ad.modify",
        "global_ads_searchrec_ad_modify.py '<JSON>'",
    )

#!/usr/bin/env python3
"""temu.searchrec.ad.reports.mall.query — 店铺广告报表查询."""
import _global_ads_script as m

if __name__ == "__main__":
    m.run_cli(
        "temu.searchrec.ad.reports.mall.query",
        "global_ads_searchrec_ad_reports_mall_query.py '<JSON>'",
    )

#!/usr/bin/env python3
"""temu.searchrec.ad.roas.pred — 广告 ROAS 预测."""
import _global_ads_script as m

if __name__ == "__main__":
    m.run_cli(
        "temu.searchrec.ad.roas.pred",
        "global_ads_searchrec_ad_roas_pred.py '<JSON>'",
    )

#!/usr/bin/env python3
"""
List locally saved Temu accessTokens (masked).

Usage:
  python list_temu_access_tokens.py
  python list_temu_access_tokens.py '{"mask": false}'
"""

import json
import sys

from _temu_token_store import list_stores

def main():
    mask = True
    if len(sys.argv) >= 2:
        try:
            params = json.loads(sys.argv[1])
            mask = params.get("mask", True)
        except json.JSONDecodeError as e:
            print(f"Invalid parameter format: {e}", file=sys.stderr)
            sys.exit(1)

    print(json.dumps(list_stores(mask=mask), 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())