首页 > 应用开发 > IT综合 > 电商软件开发软件实录:京东图片搜索商品列表
import base64
from PIL import Image
import io
def process_image(image_path):
"""处理图片为京东图片搜索接口要求的格式"""
try:
# 打开图片并转换为RGB模式(避免CMYK报错)
with Image.open(image_path) as img:
if img.mode == "CMYK":
img = img.convert("RGB")
# 压缩图片至2MB以内
max_size = 2 * 1024 * 1024 # 2MB
img_byte_arr = io.BytesIO()
quality = 95
while True:
img_byte_arr.seek(0)
img.save(img_byte_arr, format="JPEG", quality=quality)
if img_byte_arr.tell() <= max_size or quality <= 10:
break
quality -= 5
# 转换为base64,去除前缀和换行
img_base64 = base64.b64encode(img_byte_arr.getvalue()).decode("utf-8").replace("\n", "")
return img_base64
except Exception as e:
print(f"图片处理失败: {e}")
return None
import hashlib
import time
import urllib.parse
def generate_image_search_sign(params, app_secret):
"""生成京东图片搜索接口的签名"""
# 1. 过滤空值并按参数名ASCII排序
sorted_params = sorted([(k, v) for k, v in params.items() if v is not None], key=lambda x: x[0])
# 2. 拼接为key=value&key=value格式(注意对值进行URL编码)
query_str = "&".join([
f"{k}={urllib.parse.quote(str(v), safe='')}"
for k, v in sorted_params
])
# 3. 首尾加app_secret,MD5加密后转大写
sign_str = f"{app_secret}{query_str}{app_secret}"
return hashlib.md5(sign_str.encode()).hexdigest().upper()
# 使用示例
img_base64 = process_image("test.jpg")
params = {
"method": "jd.image.search",
"app_key": "your_app_key",
"timestamp": str(int(time.time())),
"image_base64": img_base64,
"page": 1,
"page_size": 20
}
params["sign"] = generate_image_search_sign(params, "your_app_secret")
import requests
def filter_similar_products(response_data, min_similarity=60):
"""过滤相似度低于阈值的商品"""
products = response_data.get("result", {}).get("products", [])
filtered = []
for p in products:
# 提取相似度(不同接口版本字段可能不同,需注意)
similarity = p.get("similarity", 0)
if similarity >= min_similarity:
filtered.append({
"sku_id": p.get("sku_id"),
"name": p.get("name"),
"price": p.get("price"),
"similarity": similarity,
"image_url": p.get("image_url")
})
return filtered
# 调用示例
response = requests.post("https://api.jd.com/image/search", data=params)
filtered_products = filter_similar_products(response.json())
import time
from collections import deque
class ImageSearchLimiter:
def __init__(self, max_calls=5, period=60):
self.max_calls = max_calls
self.period = period
self.calls = deque()
def allow(self):
now = time.time()
# 移除过期的调用记录
while self.calls and now - self.calls[0] > self.period:
self.calls.popleft()
if len(self.calls) < self.max_calls:
self.calls.append(now)
return True
return False
# 使用示例
limiter = ImageSearchLimiter()
if limiter.allow():
response = requests.post(api_url, data=params)
else:
print("触发限流,等待1分钟后再试")
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/70026917/viewspace-3094342/ ,如需转载,请注明出处,否则将追究法律责任。