EastMallBuy模式淘宝1688代购系统搭建指南-平芜编程栈

一、核心概述

1. 模式定义

EastMallBuy是轻资产无货源代购方案，核心逻辑：用户在自有平台下单→系统对接淘宝/1688采购→同步物流/售后至自有平台，通过自定义加价实现盈利，适配1688代发、淘宝代购等场景。

2. 核心业务流程

用户下单→系统算代购价→订单解析→淘宝/1688采购→同步订单号→定时拉取物流→推送用户→完成/售后

二、核心模块开发（关键代码）

前置准备

1. 注册淘宝/1688开放平台，创建应用获取appkey/appsecret，申请商品/订单/物流权限；2. 安装依赖：pip install requests django selenium apscheduler；添加Taobaoapi2014加V获取演示站。

1. 商品采集（对接1688 API）

import requests import hmac import hashlib import time from dotenv import load_dotenv import os load_dotenv() APP_KEY = os.getenv("1688_APP_KEY") APP_SECRET = os.getenv("1688_APP_SECRET") API_URL = "https://gw.open.1688.com/openapi/param2/1/portals.open/" # 1688 API签名 def sign_params(params): sorted_params = sorted(params.items(), key=lambda x: x[0]) sign_str = APP_SECRET + ''.join([f"{k}{v}" for k, v in sorted_params]) + APP_SECRET return hmac.new(APP_SECRET.encode(), sign_str.encode(), hashlib.sha1).digest().hex().upper() # 获取商品详情 def get_1688_item_detail(item_id): params = { "method": "alibaba.item.get", "app_key": APP_KEY, "timestamp": str(int(time.time()*1000)), "format": "json", "v": "2.0", "item_id": item_id, "sign_method": "hmac" } params["sign"] = sign_params(params) try: resp = requests.get(API_URL, params=params, timeout=10) resp.raise_for_status() result = resp.json() if "error_response" in result: print(f"失败：{result['error_response']['msg']}") return None item = result["item_get_response"]["item"] return {"item_id": item.get("item_id"), "title": item.get("title"), "original_price": item.get("price"), "stock": item.get("stock"), "main_img": item.get("image"), "specs": item.get("sku_infos", {}).get("sku_info", [])} except Exception as e: print(f"采集失败：{str(e)}") return None

2. 加价规则（盈利核心）

def calculate_agent_price(original_price, rule_type="ratio", value=0.1): """支持比例（默认10%）、固定、阶梯加价""" try: original = float(original_price) if rule_type == "ratio": return round(original * (1 + value), 2) elif rule_type == "fixed": return round(original + value, 2) elif rule_type == "step": # 阶梯：≤50加8，50-200加15，>200加10% return round(original + 8 if original ≤50 else (original+15 if 50<original≤200 else original*1.1), 2) return original except ValueError: print("价格格式错误") return None

3. 订单下单与物流同步

（1）官方API下单（推荐）：复用签名函数，调用alibaba.trade.order.create接口，传入商品ID、规格、收货信息，返回1688订单号；

（2）物流同步（定时任务）：

from apscheduler.schedulers.background import BackgroundScheduler # 拉取物流 def get_1688_logistics(order_id): params = {"method": "taobao.logistics.trace.search", "app_key": APP_KEY, "timestamp": str(int(time.time()*1000)), "format": "json", "v": "2.0", "tid": order_id, "sign_method": "hmac"} params["sign"] = sign_params(params) try: resp = requests.get(API_URL, params=params, timeout=10) result = resp.json() if "error_response" in result: return None logistics = result["logistics_trace_search_response"] return {"express_company": logistics.get("company_name"), "express_no": logistics.get("mail_no"), "trace": [{"time": t.get("accept_time"), "content": t.get("accept_address")} for t in logistics.get("trace_list", [])]} except Exception as e: print(f"物流拉取失败：{str(e)}") return None # 定时同步（每10分钟） def sync_all_logistics(): # 伪代码：查询待发货订单，遍历拉取物流并更新 # orders = Order.objects.filter(status="待发货") print("物流同步完成") scheduler = BackgroundScheduler() scheduler.add_job(sync_all_logistics, "interval", minutes=10) scheduler.start()