sign = MD5( secret + "key1value1key2value2..." + secret ).upper()

pip install requests pandas

import hashlib
import time
import json
import base64
import requests
import pandas as pd

# ========= 1. 配置区（替换成你自己的密钥）=========
APP_KEY = "your_app_key"
APP_SECRET = "your_app_secret"
GATEWAY = "https://api.1688.com/router/rest"


# ========= 2. 签名生成 =========
def build_sign(params: dict, secret: str) -> str:
    """按阿里开放平台规则生成签名：字典排序 -> 拼接 -> 首尾包 secret -> MD5 大写"""
    sorted_keys = sorted(params.keys())
    sign_str = "".join(f"{k}{params[k]}" for k in sorted_keys)
    sign_str = secret + sign_str + secret
    return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()


# ========= 3. 本地图片转 Base64（可选）=========
def image_to_base64(image_path: str) -> str:
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode("utf-8")
    # 注意：不要加 data:image/jpg;base64, 前缀
    return b64.replace("\n", "").replace("\r", "").strip()


# ========= 4. 以图搜货主函数 =========
def search_by_image(img_url: str = None,
                    img_base64: str = None,
                    search_type: int = 1,
                    page: int = 1,
                    page_size: int = 20,
                    cat: str = None) -> dict:
    if not img_url and not img_base64:
        raise ValueError("必须提供 img_url 或 img_base64 其中之一")

    params = {
        "method": "1688.item_search_img",
        "app_key": APP_KEY,
        "timestamp": str(int(time.time() * 1000)),  # 毫秒时间戳
        "v": "2.0",
        "format": "json",
        "sign_method": "md5",
        "imgid": img_url if img_url else img_base64,
        "search_type": search_type,
        "page": page,
        "page_size": page_size,
    }
    if cat:
        params["cat"] = cat

    params["sign"] = build_sign(params, APP_SECRET)

    resp = requests.post(GATEWAY, data=params, timeout=15)
    resp.raise_for_status()
    return resp.json()


# ========= 5. 解析结果 =========
def parse_items(data: dict, similarity_threshold: float = 0.7) -> pd.DataFrame:
    items = (data.get("items") or {}).get("item") or []
    rows = []
    for it in items:
        sim = float(it.get("similarity", 0) or 0)
        rows.append({
            "num_iid": it.get("num_iid"),
            "title": it.get("title"),
            "price": it.get("price"),
            "moq": it.get("moq"),
            "sales": it.get("sales"),
            "similarity": sim,
            "seller_name": it.get("seller_name"),
            "detail_url": it.get("detail_url"),
            "tag": "同款" if sim >= 0.8 else ("相似" if sim >= 0.6 else "存疑"),
        })
    df = pd.DataFrame(rows)
    if not df.empty:
        df = df[df["similarity"] >= similarity_threshold]
        df = df.sort_values("similarity", ascending=False).reset_index(drop=True)
    return df


# ========= 6. 主流程 =========
if __name__ == "__main__":
    # 方式一：直接用公网图片 URL
    result = search_by_image(
        img_url="https://cbu01.alicdn.com/img/ibank/example.jpg",
        search_type=1,       # 1=同款，0=相似款
        page=1,
        page_size=20,
    )

    # 方式二：本地图片转 Base64（二选一即可）
    # b64 = image_to_base64("./sample.jpg")
    # result = search_by_image(img_base64=b64, search_type=1)

    print("返回状态:", result.get("code"), result.get("msg"))

    df = parse_items(result, similarity_threshold=0.7)
    print(df.to_string(index=False))

    # 导出比价表
    if not df.empty:
        df.to_csv("1688_pailitao_result.csv", index=False, encoding="utf-8-sig")
        print("已导出 1688_pailitao_result.csv")

返回状态: 200 success
       num_iid                        title  price  moq  sales  similarity     seller_name tag
0  728510689123          2025夏新款真丝连衣裙 128.00    2    327        0.92  杭州XX服饰工厂  同款
1  728510776655              真丝雪纺连衣裙批 95.50    3    189        0.85  绍兴XX纺织    同款
2  728510812345            仿真丝连衣裙 一件代发 68.00    1     56        0.73  义乌XX服饰    相似

def get_item_detail(num_iid: str) -> dict:
    params = {
        "method": "1688.item_get",          # 详情接口
        "app_key": APP_KEY,
        "timestamp": str(int(time.time() * 1000)),
        "v": "2.0",
        "format": "json",
        "sign_method": "md5",
        "num_iid": num_iid,
    }
    params["sign"] = build_sign(params, APP_SECRET)
    resp = requests.post(GATEWAY, data=params, timeout=15)
    return resp.json()


# 对搜到的 Top N 商品批量拉详情
for iid in df["num_iid"].head(5):
    detail = get_item_detail(iid)
    # detail 中可拿到：阶梯批发价、SKU、库存、重量、材质、发货时效等

{
"code"
:
200
,
"msg"
:
"success"
,
"request_id"
:
"12345abcde"
,
"items"
:
{
"page"
:
"1"
,
"real_total_results"
:
670
,
"total_results"
:
670
,
"pagecount"
:
14
,
"page_size"
:
"50"
,
"item"
:
[
"..."
]
}
}