跨境/店群必看！Open Claw 1688接口批量选品监控，低成本放大收益

# Open Claw 1688接口批量选品监控（跨境/店群专属版）
import requests
import time
import pandas as pd

# 配置参数（新手替换成自己的信息，批量素材可直接添加到列表）
API_KEY = "你的接口Key"
API_SECRET = "你的接口Secret"
# 批量素材：跨境/店群可根据需求，添加多个图片URL、关键词
BATCH_IMG_URLS = [
    "https://img.example.com/bao-kuan-1.jpg",
    "https://img.example.com/bao-kuan-2.jpg",
    "https://img.example.com/bao-kuan-3.jpg"
]
BATCH_KEYWORDS = [
    "跨境家居收纳", "速卖通爆款饰品", "店群低价T恤", "亚马逊厨房工具"
]
MONITOR_INTERVAL = 3600  # 批量监控间隔，1小时一次
SAVE_PATH = "跨境店群批量选品数据.xlsx"  # 批量保存路径

# 构造批量请求函数
def batch_request(items, request_type="img"):
    """
    批量请求接口：request_type="img"按图搜款，request_type="keyword"关键词搜款
    """
    all_products = []
    for item in items:
        try:
            # 构造请求链接（按图搜款/关键词搜款切换）
            if request_type == "img":
                url = f"https://api-gw.xxx.cn/1688/item_search_img/?key={API_KEY}&secret={API_SECRET}&imgid={item}&cache=yes&result_type=json"
            else:
                url = f"https://api-gw.xxx.cn/1688/item_search/?key={API_KEY}&secret={API_SECRET}&q={item}&cache=yes&result_type=json"
            
            headers = {"Accept-Encoding": "gzip", "Connection": "close"}
            response = requests.get(url, headers=headers, timeout=15)
            data = response.json()
            
            # 调用成功，筛选优质货源
            if data.get("error_code") == "0000":
                for product in data["items"]["item"]:
                    # 跨境/店群筛选条件（可自定义修改）
                    # 跨境筛选：支持跨境物流、销量>50、发货地为广州/深圳
                    # 店群筛选：支持一件代发、价格<50、销量>100、相似度>0.85
                    if (
                        # 跨境筛选条件（注释掉可切换为店群筛选）
                        product.get("is_cross_border", "no") == "yes"
                        and int(product.get("sales", 0)) > 50
                        and product.get("location", "").startswith(("广州", "深圳"))
                        
                        # 店群筛选条件（需要时取消注释，注释跨境条件）
                        # product.get("is_drop_shipping", "no") == "yes"
                        # and float(product.get("price", 0)) < 50
                        # and int(product.get("sales", 0)) > 100
                        # and float(product.get("similarity", 0)) > 0.85
                    ):
                        product_info = {
                            "商品标题": product["title"],
                            "拿货价": product["price"],
                            "销量": product["sales"],
                            "相似度": product.get("similarity", "无"),
                            "商品链接": product["detail_url"],
                            "店铺名称": product["shop_name"],
                            "发货地": product.get("location", "无"),
                            "支持跨境": product.get("is_cross_border", "no"),
                            "支持一件代发": product.get("is_drop_shipping", "no")
                        }
                        all_products.append(product_info)
            time.sleep(5)  # 批量请求间隔，避免高频触发风控
        except Exception as e:
            print(f"请求{item}失败：{str(e)}")
    return all_products

# 批量保存数据到Excel
def save_batch_data(products):
    if not products:
        print("未获取到优质货源")
        return
    df = pd.DataFrame(products)
    df.to_excel(SAVE_PATH, index=False, encoding="utf-8-sig")
    print(f"批量选品完成，共获取{len(products)}款优质货源，已保存到：{SAVE_PATH}")

# 批量监控函数
def batch_monitor():
    print("开始批量监控选品...（按Ctrl+C停止）")
    while True:
        print(f"\n{time.strftime('%Y-%m-%d %H:%M:%S')} - 正在批量获取选品数据...")
        # 批量按图搜款 + 批量关键词搜款
        img_products = batch_request(BATCH_IMG_URLS, request_type="img")
        keyword_products = batch_request(BATCH_KEYWORDS, request_type="keyword")
        # 合并数据，去重
        all_products = img_products + keyword_products
        all_products = [dict(t) for t in {tuple(d.items()) for d in all_products}]  # 去重
        # 保存数据
        save_batch_data(all_products)
        print(f"下次批量监控时间：{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time() + MONITOR_INTERVAL))}")
        time.sleep(MONITOR_INTERVAL)

# 执行程序（跨境/店群卖家，修改筛选条件后直接运行）
if __name__ == "__main__":
    # 新手可先执行单次批量搜款，测试数据（注释下面一行，开启监控）
    # img_products = batch_request(BATCH_IMG_URLS, request_type="img")
    # keyword_products = batch_request(BATCH_KEYWORDS, request_type="keyword")
    # all_products = img_products + keyword_products
    # save_batch_data(all_products)
    
    # 测试通过后，开启批量监控
    batch_monitor()