1688 作為 B2B 電商核心平臺，其商品詳情包含批發價、起訂量、供應商資質等 B 端特色數據，接口架構與 C 端平臺差異顯著。本文聚焦 1688 商品詳情接口的技術實現，重點解決平臺加密參數破解、多接口數據協同、供應商信息提取等核心問題，提供一套合規、可落地的 B 端電商數據采集方案，嚴格遵循平臺規則與數據安全規范。

一、1688 詳情接口架構與合規要點

1688 商品詳情數據分散在基礎信息接口、價格庫存接口、供應商接口和規格參數接口中，需多接口協同獲取。實現前需明確以下合規邊界，確保通過 CSDN 審核且符合平臺規則：

數據范圍合規：僅采集公開的商品信息（批發價、起訂量、規格等），不涉及平臺私有 API 或用戶交易數據；

請求行為合規：單 IP 請求間隔不低于 20 秒，單商品詳情采集流程（含多接口）總耗時控制在 60 秒以上；

使用場景合規：數據僅用于市場調研、供應鏈分析等合法場景，不得用于惡意比價、商業競爭；

協議遵循：嚴格遵守 1688 robots.txt 協議，不爬取 disallow 標記的路徑（如 /trade/ 交易相關頁面）。

核心技術流程如下：

plaintext

商品ID解析 → 多接口參數生成 → 分布式請求調度 → 數據清洗與融合 → 結構化存儲

點擊獲取key和secre

二、核心技術實現：多接口協同采集與解析

1. 1688 商品 ID 解析器（適配 B 端 URL 特色）

1688 商品 URL 格式多樣（含 PC 端、移動端、短鏈等），需針對性解析商品 ID（offerId）：

python

運行

import re

import requests

from lxml import etree

class AlibabaOfferIdParser:

"""1688商品ID（offerId）解析器"""

def __init__(self):

self.headers = {

"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",

"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",

"Referer": "https://www.1688.com/"

}

def parse_from_url(self, product_url):

"""從URL直接提取offerId（適配多種URL格式）"""

# 匹配PC端標準URL：https://detail.1688.com/offer/1234567890.html

pc_pattern = r"offer/(d+).html"

# 匹配移動端URL：https://m.1688.com/offer/1234567890.html

m_pattern = r"m.1688.com/offer/(d+).html"

# 匹配短鏈：https://s.1688.com/selloffer/offer_view.htm?offerId=1234567890

short_pattern = r"offerId=(d+)"

for pattern in [pc_pattern, m_pattern, short_pattern]:

match = re.search(pattern, product_url)

if match:

return match.group(1)

return None

def parse_from_page(self, product_url):

"""URL解析失敗時，從頁面元數據提取offerId"""

try:

response = requests.get(

product_url,

headers=self.headers,

timeout=15,

allow_redirects=True

)

response.encoding = "utf-8"

# 從meta標簽提取（1688頁面特色）

tree = etree.HTML(response.text)

offer_id_meta = tree.xpath('//meta[@name="offerId"]/@content')

if offer_id_meta:

return offer_id_meta[0]

# 從腳本標簽提取（適配動態渲染頁面）

script_tags = tree.xpath('//script[contains(text(), "offerId")]/text()')

for script in script_tags:

match = re.search(r'offerIds*[:=]s*["']?(d+)["']?', script)

if match:

return match.group(1)

return None

except Exception as e:

print(f"頁面提取offerId失敗: {str(e)}")

return None

def get_offer_id(self, product_url):

"""統一入口：先URL解析，失敗則頁面解析"""

offer_id = self.parse_from_url(product_url)

if offer_id:

return offer_id

return self.parse_from_page(product_url)

2. 多接口參數生成器（適配 1688 加密規則）

1688 詳情接口需動態生成 sign、timestamp 等加密參數，不同接口參數規則差異較大，需針對性處理：

python

運行

import time

import random

import hashlib

import json

class AlibabaParamsGenerator:

"""1688多接口參數生成器"""

def __init__(self):

self.app_key = "12574478" # 1688公開應用標識

self.secret = "6383d13959f142e59ac4a3d938826101" # 模擬密鑰（實際需動態獲取）

self.platform = "pc"

def generate_base_params(self):

"""生成基礎公共參數"""

return {

"appKey": self.app_key,

"timestamp": str(int(time.time() * 1000)),

"format": "json",

"v": "2.0",

"signMethod": "md5",

"partnerId": "apidoc",

"session": "" # 無需登錄時留空

}

def generate_sign(self, params):

"""生成1688標準簽名（MD5加密）"""

# 按參數名ASCII排序

sorted_params = sorted(params.items(), key=lambda x: x[0])

# 拼接參數+密鑰

sign_str = "".join([f"{k}{v}" for k, v in sorted_params]) + self.secret

# MD5加密并轉為大寫

return hashlib.md5(sign_str.encode()).hexdigest().upper()

def generate_basic_params(self, offer_id):

"""生成基礎信息接口參數（商品名稱、主圖等）"""

params = self.generate_base_params()

params.update({

"method": "alibaba.offer.get",

"offerId": offer_id,

"fields": "offerId,title,picUrl,detailUrl,cateId,cateName"

})

params["sign"] = self.generate_sign(params)

return params

def generate_price_params(self, offer_id):

"""生成價格/起訂量接口參數（B端核心數據）"""

params = self.generate_base_params()

params.update({

"method": "alibaba.offer.price.get",

"offerId": offer_id,

"fields": "priceRange,moq,unit,priceType,promotionPrice"

})

params["sign"] = self.generate_sign(params)

return params

def generate_supplier_params(self, supplier_id):

"""生成供應商信息接口參數（B端特色）"""

params = self.generate_base_params()

params.update({

"method": "alibaba.member.get",

"memberId": supplier_id,

"fields": "memberId,companyName,mainProduct,creditLevel,startYear"

})

params["sign"] = self.generate_sign(params)

return params

def generate_spec_params(self, offer_id):

"""生成規格參數接口參數（多SKU適配）"""

params = self.generate_base_params()

params.update({

"method": "alibaba.offer.spec.get",

"offerId": offer_id,

"fields": "specId,specName,specValues,skuList"

})

params["sign"] = self.generate_sign(params)

return params

3. 分布式請求調度器（應對 B 端反爬）

1688 對 B 端數據接口反爬嚴格，需實現代理池輪換、請求間隔動態調整、會話保持等策略：

python

運行

import time

import random

import requests

from fake_useragent import UserAgent

from concurrent.futures import ThreadPoolExecutor, as_completed

class AlibabaRequestScheduler:

"""1688多接口請求調度器（分布式架構）"""

def __init__(self, proxy_pool=None, max_workers=3):

self.api_domain = "https://gw.open.1688.com/openapi/api" # 1688開放平臺入口

self.proxy_pool = proxy_pool or []

self.ua = UserAgent()

self.max_workers = max_workers # 并發數（B端接口建議≤3）

self.session_pool = self._init_session_pool() # 多會話池避免單一會話被封

def _init_session_pool(self):

"""初始化會話池（每個會話獨立Cookie）"""

session_pool = []

for _ in range(self.max_workers):

session = requests.Session()

# 配置基礎 headers

session.headers.update({

"User-Agent": self.ua.random,

"Accept": "application/json,text/plain,*/*",

"Referer": "https://www.1688.com/",

"Origin": "https://www.1688.com"

})

# 預訪問首頁獲取基礎Cookie

session.get("https://www.1688.com", timeout=10)

session_pool.append(session)

return session_pool

def _get_proxy(self):

"""從代理池獲取可用代理（帶健康檢測）"""

if not self.proxy_pool:

return None

# 隨機選擇代理并驗證

proxy = random.choice(self.proxy_pool)

try:

requests.get("https://www.1688.com", proxies={"https": proxy}, timeout=5)

return proxy

except:

# 移除無效代理

self.proxy_pool.remove(proxy)

print(f"移除無效代理: {proxy}")

return self._get_proxy() if self.proxy_pool else None

def _dynamic_sleep(self, interface_type):

"""根據接口類型動態調整間隔（B端接口間隔更長）"""

# 基礎信息接口：20-25秒

# 價格/供應商接口：25-30秒（敏感數據反爬更嚴）

interval_map = {

"basic": random.uniform(20, 25),

"price": random.uniform(25, 30),

"supplier": random.uniform(25, 30),

"spec": random.uniform(22, 27)

}

sleep_time = interval_map.get(interface_type, 25)

print(f"接口請求間隔: {sleep_time:.1f}秒")

time.sleep(sleep_time)

def send_request(self, params, interface_type):

"""發送單接口請求"""

self._dynamic_sleep(interface_type)

proxy = self._get_proxy()

proxies = {"https": proxy} if proxy else None

# 從會話池隨機選擇一個會話

session = random.choice(self.session_pool)

try:

response = session.get(

self.api_domain,

params=params,

proxies=proxies,

timeout=20

)

# 檢查反爬攔截

if self._is_blocked(response.text):

print(f"接口{interface_type}被攔截，更換會話與代理")

# 重置會話池

self.session_pool = self._init_session_pool()

return None

return response.json()

except Exception as e:

print(f"接口{interface_type}請求異常: {str(e)}")

return None

def send_batch_requests(self, params_list):

"""批量發送多接口請求（并發調度）"""

results = {}

with ThreadPoolExecutor(max_workers=self.max_workers) as executor:

# 構建任務：(接口類型, 未來對象)

future_tasks = {

executor.submit(self.send_request, params, interface_type): interface_type

for interface_type, params in params_list.items()

}

# 收集結果

for future in as_completed(future_tasks):

interface_type = future_tasks[future]

try:

result = future.result()

results[interface_type] = result

print(f"接口{interface_type}請求完成")

except Exception as e:

results[interface_type] = None

print(f"接口{interface_type}任務異常: {str(e)}")

return results

def _is_blocked(self, response_text):

"""判斷是否被反爬攔截（1688特色攔截標識）"""

blocked_indicators = [

"請輸入驗證碼",

"訪問頻率過高",

"系統繁忙",

"403 Forbidden",

"login required"

]

return any(indicator in response_text for indicator in blocked_indicators)

4. 多源數據融合解析器（B 端數據特色處理）

1688 數據分散在多個接口，需融合解析并處理 B 端特色字段（如起訂量、批發價區間、供應商資質等）：

python

運行

import json

from datetime import datetime

class AlibabaDataMerger:

"""1688多接口數據融合解析器"""

def __init__(self):

pass

def parse_basic_data(self, basic_json):

"""解析基礎信息接口數據"""

if not basic_json or basic_json.get("errorCode") != 0:

return None

result = {}

data = basic_json.get("result", {})

# 基礎商品信息

result["offer_id"] = data.get("offerId", "")

result["title"] = data.get("title", "").strip()

result["main_image"] = data.get("picUrl", "")

result["detail_url"] = data.get("detailUrl", "")

# 分類信息（B端多級分類）

result["category"] = {

"id": data.get("cateId", ""),

"name": data.get("cateName", ""),

"full_path": self._parse_category_path(data.get("catePath", ""))

}

# 供應商ID（用于后續調用供應商接口）

result["supplier_id"] = data.get("memberId", "")

return result

def parse_price_data(self, price_json):

"""解析價格/起訂量數據（B端核心）"""

if not price_json or price_json.get("errorCode") != 0:

return None

data = price_json.get("result", {})

return {

"price_range": {

"min": float(data.get("priceRange", {}).get("minPrice", 0)),

"max": float(data.get("priceRange", {}).get("maxPrice", 0)),

"unit": data.get("unit", "件")

},

"moq": int(data.get("moq", 1)), # 最小起訂量（B端特色）

"price_type": data.get("priceType", "wholesale"), # 批發價/零售價

"promotion": {

"has_promo": "promotionPrice" in data,

"price": float(data.get("promotionPrice", 0)) if "promotionPrice" in data else 0

}

}

def parse_supplier_data(self, supplier_json):

"""解析供應商信息（B端特色）"""

if not supplier_json or supplier_json.get("errorCode") != 0:

return None

data = supplier_json.get("result", {})

return {

"id": data.get("memberId", ""),

"company_name": data.get("companyName", ""),

"main_product": data.get("mainProduct", "").split(";") if data.get("mainProduct") else [],

"credit_level": data.get("creditLevel", "未評級"), # 誠信通等級

"establishment_year": data.get("startYear", "未知"), # 成立年份

"is_verified": "verified" in data # 是否企業認證

}

def parse_spec_data(self, spec_json):

"""解析規格參數與多SKU數據"""

if not spec_json or spec_json.get("errorCode") != 0:

return None

data = spec_json.get("result", {})

spec_groups = []

# 解析規格組（如顏色、尺寸）

for spec in data.get("specList", []):

spec_groups.append({

"spec_id": spec.get("specId", ""),

"spec_name": spec.get("specName", ""),

"values": [v.get("specValueName", "") for v in spec.get("specValueList", [])]

})

#

審核編輯 黃宇