Superbuy 模式淘宝 / 1688 代购系统搭建指南

一、Superbuy 模式代购系统核心概述

1. 模式定义

Superbuy 模式本质是一站式跨境 / 国内代购服务系统：用户提交淘宝 / 1688 商品链接 / 代购需求，系统自动抓取商品信息（价格、规格、库存），计算「商品价 + 代购费 + 国内运费 + 国际运费（跨境场景）+ 服务费」等综合费用，用户支付后，系统 / 人工完成采购、验货、打包、发货，最终实现物流跟踪全流程管理。

2. 核心业务流程

二、系统架构设计

1. 整体架构（轻量化版）

2. 核心功能模块

三、技术选型（零基础友好版）

1. 核心技术栈（优先轻量化，降低搭建成本）

四、核心功能实现（Python 版）

1. 环境准备

bash

# 安装核心依赖 pip install fastapi uvicorn playwright mysql-connector-python requests python-dotenv # 安装Playwright浏览器驱动（首次执行） playwright install chromium

2. 核心模块实现

（1）商品抓取模块（淘宝 / 1688 通用）

核心：模拟浏览器访问商品链接，解析页面获取关键信息（抗反爬优于纯接口抓取）

python

from playwright.sync_api import sync_playwright import re import time def crawl_taobao_1688(url): """ 抓取淘宝/1688商品信息 :param url: 商品链接（淘宝https://item.taobao.com/ 或1688 https://detail.1688.com/） :return: 商品信息字典 """ # 基础配置（规避反爬） 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" goods_info = { "title": "", "price": 0.0, "original_price": 0.0, "sku_list": [], "stock": "未知", "shop_name": "", "error": "" } try: with sync_playwright() as p: # 启动浏览器（无头模式，生产环境用；调试时设为headless=False） browser = p.chromium.launch( headless=True, args=["--no-sandbox", "--disable-blink-features=AutomationControlled"] ) context = browser.new_context( user_agent=user_agent, viewport={"width": 1920, "height": 1080} ) # 禁用webdriver标识（抗反爬关键） context.add_init_script(""" Object.defineProperty(navigator, 'webdriver', { get: () => undefined }) """) page = context.new_page() # 访问商品页（添加等待，避免页面加载不完整） page.goto(url, timeout=30000) time.sleep(2) # 等待动态内容加载 # 1. 解析商品标题 title_elem = page.locator("h1[class*='title']") or page.locator(".tb-detail-hd h1") goods_info["title"] = title_elem.inner_text().strip() if title_elem else "未知" # 2. 解析价格（适配淘宝/1688不同的价格标签） price_elem = page.locator(".price J_price") or page.locator(".price-price") if price_elem: price_text = price_elem.inner_text().strip().replace("¥", "").replace(",", "") goods_info["price"] = float(price_text) if price_text else 0.0 # 3. 解析店铺名称 shop_elem = page.locator(".shop-name") or page.locator(".seller-name") goods_info["shop_name"] = shop_elem.inner_text().strip() if shop_elem else "未知" # 4. 解析库存（简化版，复杂SKU需单独处理） stock_elem = page.locator(".stock") or page.locator(".quantity") goods_info["stock"] = stock_elem.inner_text().strip() if stock_elem else "未知" browser.close() return goods_info except Exception as e: goods_info["error"] = f"抓取失败：{str(e)}" return goods_info # 测试调用 if __name__ == "__main__": # 替换为淘宝/1688商品链接 test_url = "https://item.taobao.com/item.htm?id=699999999999" result = crawl_taobao_1688(test_url) print("商品抓取结果：", result)

（2）订单管理模块（数据库交互）

核心：创建订单、更新订单状态、查询订单（基于 MySQL）

python

import mysql.connector from dotenv import load_dotenv import os from datetime import datetime # 加载环境变量（存储数据库配置，避免硬编码） load_dotenv() # 数据库连接配置 DB_CONFIG = { "host": os.getenv("DB_HOST"), "user": os.getenv("DB_USER"), "password": os.getenv("DB_PASSWORD"), "database": os.getenv("DB_NAME") } # 初始化数据库（首次执行） def init_db(): conn = mysql.connector.connect(**DB_CONFIG) cursor = conn.cursor() # 创建订单表 cursor.execute(""" CREATE TABLE IF NOT EXISTS orders ( order_id VARCHAR(32) PRIMARY KEY, user_id VARCHAR(32) NOT NULL, goods_url VARCHAR(512) NOT NULL, goods_title VARCHAR(512) NOT NULL, goods_price FLOAT NOT NULL, service_fee FLOAT NOT NULL, -- 代购服务费 domestic_shipping FLOAT NOT NULL, -- 国内运费 international_shipping FLOAT NOT NULL, -- 国际运费（跨境场景） total_amount FLOAT NOT NULL, -- 总费用 order_status ENUM('pending_pay', 'paid', 'purchasing', 'shipped', 'completed', 'cancelled') NOT NULL DEFAULT 'pending_pay', purchase_order_no VARCHAR(64) DEFAULT '', -- 淘宝/1688采购单号 logistics_no VARCHAR(64) DEFAULT '', -- 物流单号 create_time DATETIME NOT NULL, update_time DATETIME NOT NULL ) """) conn.commit() cursor.close() conn.close() print("数据库初始化完成") # 创建订单 def create_order(order_id, user_id, goods_info, fee_calculation): """ 创建代购订单 :param order_id: 唯一订单号（可生成UUID） :param user_id: 用户ID :param goods_info: 商品信息字典 :param fee_calculation: 费用计算结果字典 """ conn = mysql.connector.connect(**DB_CONFIG) cursor = conn.cursor() now = datetime.now() sql = """ INSERT INTO orders ( order_id, user_id, goods_url, goods_title, goods_price, service_fee, domestic_shipping, international_shipping, total_amount, order_status, create_time, update_time ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ values = ( order_id, user_id, fee_calculation["goods_url"], goods_info["title"], goods_info["price"], fee_calculation["service_fee"], fee_calculation["domestic_shipping"], fee_calculation["international_shipping"], fee_calculation["total_amount"], "pending_pay", now, now ) cursor.execute(sql, values) conn.commit() cursor.close() conn.close() print(f"订单 {order_id} 创建成功") # 更新订单状态 def update_order_status(order_id, status, purchase_order_no="", logistics_no=""): conn = mysql.connector.connect(**DB_CONFIG) cursor = conn.cursor() now = datetime.now() sql = """ UPDATE orders SET order_status = %s, purchase_order_no = %s, logistics_no = %s, update_time = %s WHERE order_id = %s """ values = (status, purchase_order_no, logistics_no, now, order_id) cursor.execute(sql, values) conn.commit() cursor.close() conn.close() print(f"订单 {order_id} 状态更新为：{status}") # 测试调用 if __name__ == "__main__": # 先创建.env文件，配置DB_HOST、DB_USER、DB_PASSWORD、DB_NAME init_db() # 模拟创建订单 test_order_id = "ORD20260105001" test_user_id = "USER001" test_goods_info = {"title": "测试商品", "price": 100.0} test_fee = { "goods_url": "https://item.taobao.com/test", "service_fee": 10.0, "domestic_shipping": 5.0, "international_shipping": 50.0, "total_amount": 165.0 } create_order(test_order_id, test_user_id, test_goods_info, test_fee) # 更新订单状态为已支付 update_order_status(test_order_id, "paid")

（3）费用计算模块（核心规则）

核心：按「商品价 + 代购服务费（比例 / 固定） + 国内运费 + 国际运费 + 打包费」计算总费用

python

def calculate_fee(goods_price, goods_weight=0.5, service_fee_ratio=0.08, domestic_shipping=5.0, international_shipping_base=30.0): """ 计算代购综合费用 :param goods_price: 商品价格（元） :param goods_weight: 商品重量（kg，默认0.5kg） :param service_fee_ratio: 代购服务费比例（默认8%） :param domestic_shipping: 国内运费（淘宝/1688到仓库，默认5元） :param international_shipping_base: 国际运费基础价（每kg，默认30元） :return: 费用计算结果字典 """ # 1. 代购服务费（最低5元） service_fee = max(goods_price * service_fee_ratio, 5.0) # 2. 国际运费（按重量计算） international_shipping = goods_weight * international_shipping_base # 3. 总费用 total_amount = goods_price + service_fee + domestic_shipping + international_shipping # 4. 打包费（固定2元） packaging_fee = 2.0 total_amount += packaging_fee return { "goods_price": goods_price, "service_fee": service_fee, "domestic_shipping": domestic_shipping, "international_shipping": international_shipping, "packaging_fee": packaging_fee, "total_amount": round(total_amount, 2) # 保留2位小数 } # 测试调用 if __name__ == "__main__": # 商品价格100元，重量1kg fee_result = calculate_fee(goods_price=100.0, goods_weight=1.0) print("费用计算结果：", fee_result) # 输出示例： # { # "goods_price": 100.0, # "service_fee": 8.0, # "domestic_shipping": 5.0, # "international_shipping": 30.0, # "packaging_fee": 2.0, # "total_amount": 145.0 # }

（4）物流跟踪模块（对接快递 100API）

核心：调用第三方物流 API，查询物流轨迹

python

import requests import json def track_logistics(logistics_no, logistics_company="auto"): """ 对接快递100API查询物流 :param logistics_no: 物流单号 :param logistics_company: 快递公司编码（auto=自动识别） :return: 物流轨迹列表 """ # 快递100API配置（需注册获取key，免费额度足够测试） API_KEY = os.getenv("KUAIDI100_KEY") API_URL = "https://www.kuaidi100.com/api" params = { "nu": logistics_no, "com": logistics_company, "key": API_KEY, "format": "json" } try: response = requests.get(API_URL, params=params, timeout=10) result = response.json() if result.get("status") == "200": return result.get("data", []) # 物流轨迹列表 else: return [{"time": "", "context": f"查询失败：{result.get('message')}"}] except Exception as e: return [{"time": "", "context": f"查询异常：{str(e)}"}] # 测试调用 if __name__ == "__main__": # 替换为真实物流单号 logistics_data = track_logistics(logistics_no="78901234567890") print("物流跟踪结果：", logistics_data)

3. 后端 API 封装（FastAPI）

核心：将上述模块封装为 HTTP 接口，供前端调用

python

from fastapi import FastAPI import uvicorn import uuid app = FastAPI(title="Superbuy代购系统API") # 1. 商品抓取接口 @app.get("/api/crawl-goods") def crawl_goods(url: str): result = crawl_taobao_1688(url) return {"code": 200 if not result["error"] else 500, "data": result} # 2. 费用计算接口 @app.get("/api/calculate-fee") def calculate_fee_api(goods_price: float, goods_weight: float = 0.5): result = calculate_fee(goods_price, goods_weight) return {"code": 200, "data": result} # 3. 创建订单接口 @app.post("/api/create-order") def create_order_api(user_id: str, goods_url: str, goods_price: float, goods_weight: float = 0.5): # 生成唯一订单号 order_id = f"ORD{datetime.now().strftime('%Y%m%d')}{str(uuid.uuid4())[:8]}" # 抓取商品信息 goods_info = crawl_taobao_1688(goods_url) if goods_info["error"]: return {"code": 500, "message": goods_info["error"]} # 计算费用 fee_result = calculate_fee(goods_price, goods_weight) fee_result["goods_url"] = goods_url # 创建订单 create_order(order_id, user_id, goods_info, fee_result) return {"code": 200, "data": {"order_id": order_id, "total_amount": fee_result["total_amount"]}} # 4. 物流跟踪接口 @app.get("/api/track-logistics") def track_logistics_api(logistics_no: str): result = track_logistics(logistics_no) return {"code": 200, "data": result} # 启动服务 if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)

五、系统部署

1. 服务器准备

• 购买云服务器：阿里云 / 腾讯云轻量应用服务器（2 核 4G，CentOS 7/8）；
• 配置安全组：开放 80/443/8000 端口（8000 为后端 API，80/443 为前端）；
• 安装依赖：Docker、Python3.9+、MySQL。

2. 部署步骤

bash

# 1. 服务器安装Docker（简化环境配置） curl -fsSL https://get.docker.com | bash -s docker # 2. 启动MySQL容器（持久化数据） docker run -d --name mysql -p 3306:3306 \ -e MYSQL_ROOT_PASSWORD=你的密码 \ -e MYSQL_DATABASE=superbuy_db \ -v /data/mysql:/var/lib/mysql \ mysql:8.0 # 3. 上传代码到服务器（如/root/superbuy） # 4. 创建.env文件，配置数据库和API密钥 cat > /root/superbuy/.env << EOF DB_HOST=localhost DB_USER=root DB_PASSWORD=你的密码 DB_NAME=superbuy_db KUAIDI100_KEY=你的快递100API密钥 EOF # 5. 启动后端服务 cd /root/superbuy nohup uvicorn main:app --host 0.0.0.0 --port 8000 & # 6. 部署前端（简化版：放置HTML文件到Nginx） # 安装Nginx yum install nginx -y # 将前端HTML/JS文件放到/usr/share/nginx/html # 启动Nginx systemctl start nginx && systemctl enable nginx

六、合规与风控要点

1. 爬虫合规

• 淘宝 / 1688 禁止商用爬取，需控制抓取频率（单 IP≤1 次 / 秒），使用 IP 代理池分散请求；
• 优先使用官方开放平台 API（如有），非官方抓取仅用于小规模测试，商用需获平台授权；
• 避免抓取用户隐私数据（如卖家联系方式）。

2. 代购业务合规

• 跨境代购需办理《跨境电子商务经营主体备案》《进出口权》等资质；
• 禁止代购违禁品（化妆品、食品需额外备案）；
• 明确收费规则，避免价格欺诈，保留交易凭证。

3. 支付与数据合规

• 支付接口需对接持牌第三方支付机构，禁止私收转账；
• 用户数据（手机号、地址）需加密存储，遵守《个人信息保护法》；
• 留存订单、物流、支付记录至少 3 年。

4. 反爬与风控

• 抓取模块使用 Playwright 模拟真人操作，避免被平台封禁 IP；
• 订单状态流转增加人工审核环节，避免恶意下单；
• 限制单用户单日下单数量，防范刷单 / 诈骗。

七、结语

1. Superbuy 模式淘宝 / 1688 代购系统的核心是「商品抓取 + 订单管理 + 费用计算 + 物流跟踪」，新手可先实现核心功能（商品抓取、费用计算），再逐步扩展支付、自动化采购等模块；
2. 合规性是系统长期运行的关键，尤其是爬虫、跨境资质、支付合规，切勿触碰平台规则和法律红线；
3. 生产环境需优化：增加接口鉴权（JWT）、日志监控、异常告警、IP 代理池、多线程 / 异步抓取，提升系统稳定性和抗风险能力；
4. 自动化采购（自动下单）模块建议初期用人工替代，待系统稳定后，再基于 Selenium/Playwright 实现半自动下单（避免平台风控）。

总结

1. Superbuy 代购系统核心业务流程为「用户需求→商品抓取→费用计算→订单支付→采购发货→物流跟踪」，Python+FastAPI+MySQL 是轻量化搭建的最优组合；
2. 商品抓取需用 Playwright 模拟浏览器（抗反爬），费用计算需按「商品价 + 服务费 + 运费」制定清晰规则，物流对接快递 100API 即可满足需求；
3. 合规风控是核心：爬虫需控频、跨境需资质、支付需合规，避免平台封禁或法律风险。

新手可先从「商品抓取 + 费用计算 + 简单订单管理」的最小可用版本入手，测试稳定后再扩展前端、支付、自动化采购等功能。