文章附件下载：https://www.pan38.com/dow/share.php?code=JCnzE 提取密码：7315
核心功能：实现淘宝商品批量上传，包含登录认证、商品数据处理、图片处理和API调用
多线程处理：使用ThreadPoolExecutor实现并发上传，提高效率
图片处理：自动缩放和转换商品图片，保持宽高比
数据读取：支持从Excel文件读取商品信息，包含标题、价格、库存等关键字段
错误处理：完善的异常捕获和错误提示机制
配置管理：通过配置文件管理API密钥等敏感信息
分类映射：内置常见商品分类ID映射表
import os
import time
import requests
import pandas as pd
from PIL import Image
from hashlib import md5
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor
class TaobaoUploader:
def init(self, config_file='config.ini'):
self.session = requests.Session()
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
'Referer': 'https://www.taobao.com'
}
self.load_config(config_file)
self.login()
def load_config(self, config_file):
"""加载配置文件"""
if not os.path.exists(config_file):
raise FileNotFoundError("配置文件不存在")
config = {}
with open(config_file, 'r', encoding='utf-8') as f:
for line in f:
if '=' in line:
key, value = line.strip().split('=', 1)
config[key.strip()] = value.strip()
self.api_url = config.get('api_url', '')
self.app_key = config.get('app_key', '')
self.app_secret = config.get('app_secret', '')
self.session_key = config.get('session_key', '')
self.max_workers = int(config.get('max_workers', '5'))
self.image_dir = config.get('image_dir', 'images')
def login(self):
"""模拟登录获取token"""
params = {
'method': 'taobao.user.seller.get',
'app_key': self.app_key,
'session': self.session_key,
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'format': 'json',
'v': '2.0',
'sign_method': 'md5'
}
sign_str = self.app_secret + ''.join([f'{k}{v}' for k,v in sorted(params.items())]) + self.app_secret
params['sign'] = md5(sign_str.encode()).hexdigest()
try:
resp = self.session.get(self.api_url, params=params, headers=self.headers)
if resp.json().get('error_response'):
raise Exception("登录失败: " + resp.json()['error_response']['msg'])
print("登录成功")
except Exception as e:
print(f"登录异常: {str(e)}")
raise
def process_image(self, image_path, target_size=(800, 800)):
"""处理商品图片"""
if not os.path.exists(image_path):
return None
try:
img = Image.open(image_path)
if img.mode != 'RGB':
img = img.convert('RGB')
# 保持宽高比缩放
img.thumbnail(target_size)
# 生成新文件名
new_name = f"processed_{os.path.basename(image_path)}"
save_path = os.path.join('processed_images', new_name)
os.makedirs('processed_images', exist_ok=True)
img.save(save_path, quality=95)
return save_path
except Exception as e:
print(f"图片处理失败: {str(e)}")
return None
def read_product_data(self, excel_file):
"""读取商品数据Excel"""
try:
df = pd.read_excel(excel_file)
required_cols = ['title', 'price', 'stock', 'description', 'category']
if not all(col in df.columns for col in required_cols):
raise ValueError("Excel缺少必要列")
products = []
for _, row in df.iterrows():
product = {
'title': str(row['title']),
'price': float(row['price']),
'stock': int(row['stock']),
'desc': str(row['description']),
'category': str(row['category']),
'images': []
}
# 处理图片列
if 'images' in df.columns:
img_str = str(row['images'])
if img_str:
product['images'] = [img.strip() for img in img_str.split(',')]
products.append(product)
return products
except Exception as e:
print(f"读取商品数据失败: {str(e)}")
return []
def upload_product(self, product):
"""上传单个商品"""
try:
# 处理图片
processed_images = []
for img in product['images']:
img_path = os.path.join(self.image_dir, img)
processed_path = self.process_image(img_path)
if processed_path:
processed_images.append(processed_path)
# 构造API参数
params = {
'method': 'taobao.item.add',
'app_key': self.app_key,
'session': self.session_key,
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'format': 'json',
'v': '2.0',
'sign_method': 'md5',
'title': product['title'],
'price': product['price'],
'num': product['stock'],
'desc': product['desc'],
'cid': self.get_category_id(product['category']),
'image': ','.join(processed_images) if processed_images else ''
}
sign_str = self.app_secret + ''.join([f'{k}{v}' for k,v in sorted(params.items())]) + self.app_secret
params['sign'] = md5(sign_str.encode()).hexdigest()
# 调用API
resp = self.session.post(self.api_url, data=params, headers=self.headers)
result = resp.json()
if 'error_response' in result:
return False, result['error_response']['msg']
else:
return True, result['item_add_response']['item']['num_iid']
except Exception as e:
return False, str(e)
def get_category_id(self, category_name):
"""获取分类ID"""
# 这里简化处理，实际应该调用淘宝API获取
category_map = {
'女装': '162104',
'男装': '162103',
'数码': '1512',
'家电': '1512',
'美妆': '1801'
}
return category_map.get(category_name, '162104')
def batch_upload(self, excel_file, max_workers=5):
"""批量上传商品"""
products = self.read_product_data(excel_file)
if not products:
return False
success_count = 0
failed_items = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = []
for product in products:
futures.append(executor.submit(self.upload_product, product))
for future in futures:
success, result = future.result()
if success:
success_count += 1
print(f"上传成功，商品ID: {result}")
else:
failed_items.append(result)
print(f"上传失败: {result}")
print(f"\n上传完成，成功: {success_count}, 失败: {len(failed_items)}")
if failed_items:
print("失败商品:")
for item in failed_items:
print(f"- {item}")
return success_count > 0
if name == 'main':
try:
uploader = TaobaoUploader()
uploader.batch_upload('products.xlsx', max_workers=5)
except Exception as e:
print(f"程序异常: {str(e)}")
==2.31.0
pandas==2.1.0
Pillow==10.0.0
openpyxl==3.1.2
