前言：采购决策背后的数据支撑
在硬件开发、工业控制、科研工程等多个领域，电子元器件的选型和采购过程日趋复杂。工程技术人员需要依据参数精确比对型号，采购方则关注供货周期与供货稳定性。然而，手动逐一在各平台检索信息，不仅耗时，也存在数据碎片化问题。
为提升效率与可视性，构建一个针对电子行业的自动采集系统，从主流分销平台中提取关键参数，已成为众多企业推动采购流程数字化的方向之一。本文将以 Digikey 平台为例，介绍一个模块化、可扩展的信息抓取方案，涵盖搜索、结构提取、代理配置与数据归档等环节。
系统结构拆解
我们设计的整体架构由五个主要组件构成，各司其职、互相解耦：
关键词输入
↓
请求构建器（配置IP代理、用户信息、浏览器标识）
↓
页面解析器（提取产品字段）
↓
数据存储器（保存至数据库）
↓
定时调度器（每日增量采集）
模块功能说明
请求构建器
- 构造搜索URL
- 接入代理IP服务（示例中使用亿牛云代理）
- 配置用户与浏览器伪装头
页面解析器
- 解析搜索结果页面HTML结构
- 获取字段：产品编号、制造商、制造商产品编号、描述、原厂交货周期等
- 支持翻页与冗余信息去重
数据存储器
- 将数据写入本地数据库（如 SQLite）
- 记录抓取时间、搜索关键词
- 为后续比对与趋势分析提供支撑
定时调度器
- 通过定时任务框架实现周期运行
- 可每日自动执行更新流程
实现示例
下面是简化版实现代码，用于演示如何结合代理、请求头伪装及HTML结构解析，完成核心数据提取并归档入库。
import requests
from bs4 import BeautifulSoup
import sqlite3
import time
from datetime import datetime
# 代理与请求头配置（参考亿牛云示例）
proxies = {
"http": "http://16YUN:16IP@proxy.16yun.cn:3100",
"https": "http://16YUN:16IP@proxy.16yun.cn:3100"
}
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/122.0",
"Cookie": "your_cookie_if_needed=abc123;"
}
# 初始化数据库结构
def init_db():
conn = sqlite3.connect("digikey_products.db")
c = conn.cursor()
c.execute("""
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
crawl_time TEXT,
keyword TEXT,
product_number TEXT,
manufacturer TEXT,
mfr_product_number TEXT,
description TEXT,
lead_time TEXT
)
""")
conn.commit()
conn.close()
# 抓取函数：根据关键词爬取页面并提取字段
def scrape_products(keyword):
print(f"正在采集关键词: {keyword}")
url = f"https://www.digikey.cn/zh/products/result?s=N&keywords={keyword}"
try:
response = requests.get(url, headers=headers, proxies=proxies, timeout=10)
response.raise_for_status()
except Exception as e:
print(f"请求失败: {e}")
return
soup = BeautifulSoup(response.text, "html.parser")
table = soup.find("table", class_="product-table")
if not table:
print("未获取到产品表格结构")
return
rows = table.find_all("tr", class_="product-table-row")
crawl_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
conn = sqlite3.connect("digikey_products.db")
c = conn.cursor()
for row in rows:
cols = row.find_all("td")
if len(cols) < 6:
continue
try:
product_number = cols[0].get_text(strip=True)
manufacturer = cols[1].get_text(strip=True)
mfr_product_number = cols[2].get_text(strip=True)
description = cols[3].get_text(strip=True)
lead_time = cols[4].get_text(strip=True)
c.execute("""
INSERT INTO products (
crawl_time, keyword, product_number, manufacturer,
mfr_product_number, description, lead_time
) VALUES (?, ?, ?, ?, ?, ?, ?)
""", (
crawl_time, keyword, product_number, manufacturer,
mfr_product_number, description, lead_time
))
except Exception as e:
print(f"数据入库失败: {e}")
conn.commit()
conn.close()
print(f"关键词'{keyword}'采集完毕。")
# 主流程入口
if __name__ == "__main__":
init_db()
keywords = ["STM32", "贴片电阻", "电容", "连接器"]
for kw in keywords:
scrape_products(kw)
time.sleep(5) # 避免访问频率过高被封锁
信息采集流程图
输入关键词或型号
↓
发起网页请求（带代理、伪装头）
↓
获取HTML响应 → 定位结果表格
↓
提取产品字段并写入数据库
↓
支持周期性采集与增量更新
