一、故事开场:运营小姐姐的周五 17:30
义乌某电商公司,每周要汇报「1688 女式 T 恤批发价」:
- 人工滑屏 200 条 → 眼花了
- Excel 手填 → 错了
- 周五熬夜 → 秃了
- 后端同事甩给她一个 60 行 Python 脚本:
- 输入关键词 → 自动抓 500 条
- 一键导出 CSV → Excel 透视图直接出
- 周五 17:30 下班 → 老板夸「高效」
- 今天把完整思路开源,复制即可跑,Jupyter 也能直接跑。
二、技术选型:为什么选「官方 API + HTML」双通道
方案 | 封号风险 | 维护成本 | 数据完整度 | 本文选用 |
---|---|---|---|---|
模拟登录+滑块 | 高 | 极高 | 100% | ❌ |
无头浏览器 | 中 | 高 | 95% | ❌ |
官方 API | 0 | 低 | 99% | ✅ 首推 |
游客 HTML | 极低 | 中 | 80% | ⚠️ 备选 |
结论:能调 API 就别硬刚页面,把精力留给变现。
三、5 分钟开账号:把 1688 开放平台「白嫖」到极致
- https://open.1688.com → 注册「个人开发者」→ 创建「自用型」应用
- 能力市场搜索 offer.search 或 1688.item.get → 秒过
- 拿到:AppKey、AppSecret、日配额 5k~2w 次(免费)
四、环境准备:一条命令装好依赖
bash
pip install requests pandas tenacity tqdm
五、核心算法:TOP 签名 20 行搞定
规则:sign = md5(secret + 字典序(参数) + secret)
六、官方 API 版:搜索 + 详情 + CSV 一条龙(60 行)
Python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import time, csv, json
import requests
from hashlib import md5
from tenacity import retry, stop_after_attempt
from tqdm import tqdm
API = "https://eco.1688.com/router/rest"
APP_KEY = "YOUR_APP_KEY"
APP_SECRET = "YOUR_APP_SECRET"
FIELDS = "offerId,title,price,picUrl,detailUrl"
# ---------- 签名 ----------
def sign(params: dict) -> str:
params = {k: v for k, v in params.items() if v is not None and k != "sign"}
base = APP_SECRET + ''.join(f"{k}{v}" for k, v in sorted(params.items())) + APP_SECRET
return md5(base.encode()).hexdigest().upper()
# ---------- API 调用 ----------
@retry(stop=stop_after_attempt(3))
def api_search(keyword: str, page: int = 1, page_size: int = 20):
params = dict(method="cn.alibaba.open:offer.search", app_key=APP_KEY,
timestamp=time.strftime("%Y-%m-%d %H:%M:%S"), format="json", v="2.0",
sign_method="md5", keywords=keyword, pageNo=page, pageSize=page_size, fields=FIELDS)
params["sign"] = sign(params)
r = requests.get(API, params=params, timeout=15)
r.raise_for_status()
data = r.json()
if "error_response" in data:
raise RuntimeError(data["error_response"]["msg"])
return data["offer_search_response"]["result"]["offers"]
# ---------- 主函数 ----------
def main():
keyword = input("请输入关键词:")
pages = int(input("抓取页数:") or "1")
csv_file = f"1688_{keyword}_{time.strftime('%Y%m%d')}.csv"
with open(csv_file, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["offerId", "title", "price", "picUrl", "detailUrl", "updated"])
for p in range(1, pages + 1):
offers = api_search(keyword, p)
for it in tqdm(offers, desc=f"Page{p}"):
writer.writerow([it["offerId"], it["title"], it["price"],
it["picUrl"], it["detailUrl"], time.strftime("%Y-%m-%d %H:%M:%S")])
time.sleep(1) # 礼貌限速
print("已导出 →", csv_file)
if __name__ == "__main__":
main()
运行示例:
复制
$ python api1688.py
请输入关键词:女式T恤
抓取页数:2
Page1: 100%|████████| 20/20 [00:02<00:00, 8.12it/s]
Page2: 100%|████████| 20/20 [00:02<00:00, 8.50it/s]
已导出 → 1688_女式T恤_20250929.csv
Excel 打开 → 透视图 3 分钟搞定。
七、备用通道:HTML 解析版(无密钥也能跑)
页面结构:.sm-offer-item → 标题、价格、销量、链接
Python
# html1688.py
import requests, csv, time
from bs4 import BeautifulSoup
from tqdm import tqdm
BASE = "https://s.1688.com"
HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
def search_html(keyword, page=1):
url = f"{BASE}/selloffer/offer_search.htm?keywords={keyword}&page={page}"
r = requests.get(url, headers=HEADERS, timeout=15)
r.raise_for_status()
soup = BeautifulSoup(r.text, "lxml")
items = []
for it in soup.select(".sm-offer-item"):
title = it.select_one(".offer-title")
price = it.select_one(".price")
sales = it.select_one(".sale-num")
link = it.select_one("a.offer-title")
items.append({
"title": title.get_text(strip=True) if title else "",
"price": price.get_text(strip=True) if price else "",
"sales": sales.get_text(strip=True) if sales else "",
"link": link["href"] if link else ""
})
return items
def main():
keyword = input("关键词:")
pages = int(input("页数:") or "1")
csv_file = f"1688html_{keyword}_{time.strftime('%Y%m%d')}.csv"
with open(csv_file, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=["title", "price", "sales", "link"])
writer.writeheader()
for p in tqdm(range(1, pages + 1), desc="Pages"):
rows = search_html(keyword, p)
writer.writerows(rows)
time.sleep(1)
print("已导出 →", csv_file)
if __name__ == "__main__":
main()
结构变化时 → 用浏览器 F12 更新 select 选择器即可
八、提速 & 稳速:并发 + 重试 + 礼貌
技巧 | 实现 | 效果 |
---|---|---|
并发 | asyncio + aiohttp | 60条≈5秒 |
重试 | tenacity @retry | 成功率99% |
限速 | asyncio.Semaphore(10) | 单IP安全 |
代理 | aiohttp.ProxyConnector | 分布式再+代理 |
进阶:把 api_search 换成 async def + aiohttp,即可一键并发。
九、三行代码看行情(Excel 透视)
- 均价 =AVERAGE(C:C)
- 最低价格排序 → 低价收货
- 销量/价格散点图 → 找「高销低价」爆款
十、把脚本升级成「副业现金流」
需求 | 工具 | 成本 |
---|---|---|
定时 | Linux cron / Win 计划任务 | 0元 |
通知 | 企业微信机器人 + requests.post | 0元 |
前端 | Streamlit / Flask 自建页 | 0元 |
订阅 | 知识星球 / 小报童 | 29元/月 已有读者把「1688 行情」做成星球,200 会员 × 50 元 = 月入 1w |
十一、合规再提醒:只挖公开矿,不碰隐私矿
- 不登录、不破解、不存储商业机密
- 不公开商家手机号、地址、SKU 级库存
- 对外展示请脱敏 offerId 后三位,避免黄牛精准狙击
- 商用前阅读《1688 数据使用协议》,必要时购买正式流量包
十二、结语:让技术「温柔」地赚钱
今天这篇软文,没有对抗、没有炫技,只有:
- 官方 API → 低风险、可商用
- 60 行代码 → 虚拟主机可复制
- CSV 输出 → Excel 能落地
- 把脚本丢给 cron,每天一杯咖啡的时间,就能生成一份「1688 行情报告」。
当别人还在手动滑屏,你已经用 Python 把 1688 变成了「躺赚」的副业提款机。