1688 作為 B2B 電商核心平臺(tái)，其商品詳情包含批發(fā)價(jià)、起訂量、供應(yīng)商資質(zhì)等 B 端特色數(shù)據(jù)，接口架構(gòu)與 C 端平臺(tái)差異顯著。本文聚焦 1688 商品詳情接口的技術(shù)實(shí)現(xiàn)，重點(diǎn)解決平臺(tái)加密參數(shù)破解、多接口數(shù)據(jù)協(xié)同、供應(yīng)商信息提取等核心問題，提供一套合規(guī)、可落地的 B 端電商數(shù)據(jù)采集方案，嚴(yán)格遵循平臺(tái)規(guī)則與數(shù)據(jù)安全規(guī)范。

一、1688 詳情接口架構(gòu)與合規(guī)要點(diǎn)

1688 商品詳情數(shù)據(jù)分散在基礎(chǔ)信息接口、價(jià)格庫存接口、供應(yīng)商接口和規(guī)格參數(shù)接口中，需多接口協(xié)同獲取。實(shí)現(xiàn)前需明確以下合規(guī)邊界，確保通過 CSDN 審核且符合平臺(tái)規(guī)則：

數(shù)據(jù)范圍合規(guī)：僅采集公開的商品信息（批發(fā)價(jià)、起訂量、規(guī)格等），不涉及平臺(tái)私有 API 或用戶交易數(shù)據(jù)；

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

使用場景合規(guī)：數(shù)據(jù)僅用于市場調(diào)研、供應(yīng)鏈分析等合法場景，不得用于惡意比價(jià)、商業(yè)競爭；

協(xié)議遵循：嚴(yán)格遵守 1688 robots.txt 協(xié)議，不爬取 disallow 標(biāo)記的路徑（如 /trade/ 交易相關(guān)頁面）。

核心技術(shù)流程如下：

plaintext

商品ID解析 → 多接口參數(shù)生成 → 分布式請求調(diào)度 → 數(shù)據(jù)清洗與融合 → 結(jié)構(gòu)化存儲(chǔ)

點(diǎn)擊獲取key和secre

二、核心技術(shù)實(shí)現(xiàn)：多接口協(xié)同采集與解析

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

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

python

運(yùn)行

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端標(biāo)準(zhǔn)URL：https://detail.1688.com/offer/1234567890.html

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

# 匹配移動(dòng)端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解析失敗時(shí)，從頁面元數(shù)據(jù)提取offerId"""

try:

response = requests.get(

product_url,

headers=self.headers,

timeout=15,

allow_redirects=True

)

response.encoding = "utf-8"

# 從meta標(biāo)簽提?。?688頁面特色）

tree = etree.HTML(response.text)

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

if offer_id_meta:

return offer_id_meta[0]

# 從腳本標(biāo)簽提?。ㄟm配動(dòng)態(tài)渲染頁面）

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):

"""統(tǒng)一入口：先URL解析，失敗則頁面解析"""

offer_id = self.parse_from_url(product_url)

if offer_id:

return offer_id

return self.parse_from_page(product_url)

2. 多接口參數(shù)生成器（適配 1688 加密規(guī)則）

1688 詳情接口需動(dòng)態(tài)生成 sign、timestamp 等加密參數(shù)，不同接口參數(shù)規(guī)則差異較大，需針對性處理：

python

運(yùn)行

import time

import random

import hashlib

import json

class AlibabaParamsGenerator:

"""1688多接口參數(shù)生成器"""

def __init__(self):

self.app_key = "12574478" # 1688公開應(yīng)用標(biāo)識(shí)

self.secret = "6383d13959f142e59ac4a3d938826101" # 模擬密鑰（實(shí)際需動(dòng)態(tài)獲?。?/p>

self.platform = "pc"

def generate_base_params(self):

"""生成基礎(chǔ)公共參數(shù)"""

return {

"appKey": self.app_key,

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

"format": "json",

"v": "2.0",

"signMethod": "md5",

"partnerId": "apidoc",

"session": "" # 無需登錄時(shí)留空

}

def generate_sign(self, params):

"""生成1688標(biāo)準(zhǔn)簽名（MD5加密）"""

# 按參數(shù)名ASCII排序

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

# 拼接參數(shù)+密鑰

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

# MD5加密并轉(zhuǎn)為大寫

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

def generate_basic_params(self, offer_id):

"""生成基礎(chǔ)信息接口參數(shù)（商品名稱、主圖等）"""

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):

"""生成價(jià)格/起訂量接口參數(shù)（B端核心數(shù)據(jù)）"""

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):

"""生成供應(yīng)商信息接口參數(shù)（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):

"""生成規(guī)格參數(shù)接口參數(shù)（多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. 分布式請求調(diào)度器（應(yīng)對 B 端反爬）

1688 對 B 端數(shù)據(jù)接口反爬嚴(yán)格，需實(shí)現(xiàn)代理池輪換、請求間隔動(dòng)態(tài)調(diào)整、會(huì)話保持等策略：

python

運(yùn)行

import time

import random

import requests

from fake_useragent import UserAgent

from concurrent.futures import ThreadPoolExecutor, as_completed

class AlibabaRequestScheduler:

"""1688多接口請求調(diào)度器（分布式架構(gòu)）"""

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

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

self.proxy_pool = proxy_pool or []

self.ua = UserAgent()

self.max_workers = max_workers # 并發(fā)數(shù)（B端接口建議≤3）

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

def _init_session_pool(self):

"""初始化會(huì)話池（每個(gè)會(huì)話獨(dú)立Cookie）"""

session_pool = []

for _ in range(self.max_workers):

session = requests.Session()

# 配置基礎(chǔ) headers

session.headers.update({

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

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

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

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

})

# 預(yù)訪問首頁獲取基礎(chǔ)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

# 隨機(jī)選擇代理并驗(yàn)證

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):

"""根據(jù)接口類型動(dòng)態(tài)調(diào)整間隔（B端接口間隔更長）"""

# 基礎(chǔ)信息接口：20-25秒

# 價(jià)格/供應(yīng)商接口：25-30秒（敏感數(shù)據(jù)反爬更嚴(yán)）

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):

"""發(fā)送單接口請求"""

self._dynamic_sleep(interface_type)

proxy = self._get_proxy()

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

# 從會(huì)話池隨機(jī)選擇一個(gè)會(huì)話

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}被攔截，更換會(huì)話與代理")

# 重置會(huì)話池

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):

"""批量發(fā)送多接口請求（并發(fā)調(diào)度）"""

results = {}

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

# 構(gòu)建任務(wù)：(接口類型, 未來對象)

future_tasks = {

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

for interface_type, params in params_list.items()

}

# 收集結(jié)果

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}任務(wù)異常: {str(e)}")

return results

def _is_blocked(self, response_text):

"""判斷是否被反爬攔截（1688特色攔截標(biāo)識(shí)）"""

blocked_indicators = [

"請輸入驗(yàn)證碼",

"訪問頻率過高",

"系統(tǒng)繁忙",

"403 Forbidden",

"login required"

]

return any(indicator in response_text for indicator in blocked_indicators)

4. 多源數(shù)據(jù)融合解析器（B 端數(shù)據(jù)特色處理）

1688 數(shù)據(jù)分散在多個(gè)接口，需融合解析并處理 B 端特色字段（如起訂量、批發(fā)價(jià)區(qū)間、供應(yīng)商資質(zhì)等）：

python

運(yùn)行

import json

from datetime import datetime

class AlibabaDataMerger:

"""1688多接口數(shù)據(jù)融合解析器"""

def __init__(self):

pass

def parse_basic_data(self, basic_json):

"""解析基礎(chǔ)信息接口數(shù)據(jù)"""

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

return None

result = {}

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

# 基礎(chǔ)商品信息

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端多級(jí)分類）

result["category"] = {

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

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

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

}

# 供應(yīng)商ID（用于后續(xù)調(diào)用供應(yīng)商接口）

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

return result

def parse_price_data(self, price_json):

"""解析價(jià)格/起訂量數(shù)據(jù)（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"), # 批發(fā)價(jià)/零售價(jià)

"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):

"""解析供應(yīng)商信息（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", "未評級(jí)"), # 誠信通等級(jí)

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

"is_verified": "verified" in data # 是否企業(yè)認(rèn)證

}

def parse_spec_data(self, spec_json):

"""解析規(guī)格參數(shù)與多SKU數(shù)據(jù)"""

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

return None

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

spec_groups = []

# 解析規(guī)格組（如顏色、尺寸）

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", [])]

})

#

審核編輯 黃宇