拼多多批量发布工具， 拼多多批量上架软件，拼多多一键上传发布工具【python】

文章附件下载：https://www.pan38.com/dow/share.php?code=JCnzE 提取密码：3187

该工具主要功能特点：
完整的API签名生成机制，确保接口调用安全
支持从CSV文件批量读取商品数据并自动转换格式
实现商品基础信息、图片、库存等全字段上传
包含错误处理和状态反馈机制

import requests
import pandas as pd
import json
import time
from datetime import datetime

class PDDProductUploader:
def init(self, app_key, app_secret, access_token):
self.base_url = "https://api.pinduoduo.com/api"
self.app_key = app_key
self.app_secret = app_secret
self.access_token = access_token
self.session = requests.Session()

def _generate_sign(self, params):
    """生成API签名"""
    params_str = ''.join([f'{k}{v}' for k,v in sorted(params.items())])
    sign_str = f"{self.app_secret}{params_str}{self.app_secret}"
    return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()

def _call_api(self, method, params):
    """基础API调用方法"""
    params.update({
        'type': method,
        'client_id': self.app_key,
        'access_token': self.access_token,
        'timestamp': str(int(time.time())),
        'data_type': 'JSON'
    })
    params['sign'] = self._generate_sign(params)

    try:
        response = self.session.post(self.base_url, data=params)
        return response.json()
    except Exception as e:
        print(f"API调用失败: {str(e)}")
        return None

def upload_products(self, csv_file):
    """批量上传商品主方法"""
    df = pd.read_csv(csv_file)
    success_count = 0

    for index, row in df.iterrows():
        product_data = {
            'outer_id': row['商品编码'],
            'name': row['商品名称'],
            'goods_type': 1,  # 普通商品
            'cat_id': int(row['类目ID']),
            'price': int(float(row['价格'])*100),  # 单位:分
            'market_price': int(float(row['市场价'])*100),
            'cost_price': int(float(row['成本价'])*100),
            'is_pre_sale': 1 if row['是否预售'] else 0,
            'weight': int(float(row['重量'])*1000),  # 单位:克
            'quantity': int(row['库存']),
            'sold_quantity': 0,
            'image_url': row['主图URL'],
            'carousel_gallery': row['轮播图URLs'].split('|'),
            'detail_gallery': row['详情图URLs'].split('|'),
            'description': row['商品描述'],
            'is_refundable': 1 if row['支持退货'] else 0
        }

        # 调用商品创建接口
        result = self._call_api('pdd.goods.add', product_data)

        if result and result.get('error_code') == 0:
            print(f"商品上传成功: {row['商品名称']} (ID: {result['goods_id']})")
            success_count += 1
        else:
            error_msg = result.get('error_msg', '未知错误') if result else 'API请求失败'
            print(f"商品上传失败: {row['商品名称']} - {error_msg}")

    return success_count

def batch_update_inventory(self, inventory_data):
    """批量更新库存"""
    return self._call_api('pdd.goods.quantity.update', {
        'goods_quantity_list': json.dumps(inventory_data)
    })

if name == "main":

# 配置开发者账号信息
UPLOADER = PDDProductUploader(
    app_key="YOUR_APP_KEY",
    app_secret="YOUR_APP_SECRET",
    access_token="YOUR_ACCESS_TOKEN"
)

# 执行批量上传
result = UPLOADER.upload_products("products.csv")
print(f"批量上传完成，成功{result}个商品")