1688 平台商品详情接口技术实现：从多接口协同到结构化数据融合全方

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

})

#

审核编辑 黄宇