/ Python 数据采集工具开发实战：从零到一的完整记录 /
1
前言
在电商运营和市场调研中，商品数据采集是一个重要需求。本文记录了我开发一个基于 Python 的商品数据采集工具的完整历程，包括最初的技术选型、开发过程中遇到的无数挫折、反复尝试的解决方案，以及最终的完整实现。
2
第一章：技术选型的纠结
2.1
1.1 最初的困惑
刚开始做这个项目时，我面临一个艰难的选择：到底用什么技术方案？我花了整整两天时间在调研和对比，因为每个方案都有明显的优缺点，实在难以抉择。
2.2
1.2 三大方案反复比较
方案一：API 接口
我一开始觉得 API 是最优解，因为：
速度快，几秒就能获取大量数据
稳定，接口返回结构化数据
资源占用少，跑在后台就行
于是我尝试了各种方式找 API：
使用 Chrome 开发者工具抓包
使用 Fiddler 抓取 HTTP 请求
研究手机 APP 的接口
查阅网上的技术文章
但是很快遇到了大问题：
问题一：接口加密很多请求都加密了，根本看不懂返回内容。我花了一周时间研究签名算法，结果发现用的是复杂的加密方式，短时间内根本无法破解。
问题二：接口频繁变动好不容易找到一些可用的接口，用了两天就失效了。对方平台在不断更新接口版本，我的程序总是报错。
问题三：需要逆向才能用有些接口参数加密、签名验证、设备指纹等，需要逆向 APP 才能用。这对于我这种只懂 Python 的开发者来说，太难了。
我尝试了各种逆向工具：
JADX（Android 逆向）
Charles 抓包
各种解密工具
结果都是以失败告终，浪费了我整整三周时间。
方案二：现成爬虫框架
API 走不通，我转向了 Scrapy、PySpider 等现成爬虫框架。
优势：
简单快速，几行代码就能用
社区活跃，遇到问题容易找到答案
内置了很多功能（去重、代理、限流等）
于是我用 Scrapy 重写了一遍代码，大概用了一周时间完成。
但是很快遇到了大坑：
问题一：被封杀严重第一天还好，第二天就频繁遇到：
403 Forbidden 错误
IP 被临时封禁
账号被锁定
我尝试了各种方法：
结果都不管用，还是会被封。
问题二：页面结构复杂有些页面是单页应用（SPA），用传统的 HTTP 请求根本拿不到数据。我研究了好几天 JavaScript 渲染，最后放弃了。
问题三：维护成本高平台一更新，我的爬虫就失效。每次都要重新分析、重写代码，太累了。
Scrapy 方案让我白白浪费了快一个月的时间。
方案三：Selenium 浏览器自动化
最后实在没办法，我选择了 Selenium 这个笨办法。
一开始我很不情愿，因为：
速度慢，打开浏览器要几秒
资源占用大，内存 CPU 都吃紧
不如 API 优雅
但我没得选了，因为前两个方案都失败了。
2.3
1.3 最终的技术选择
为什么只能选 Selenium：
真实浏览器环境：模拟真实用户行为，不容易被识别
支持 JavaScript：动态加载的内容也能获取
不需要逆向：直接在浏览器里操作，无需破解 API
可视化调试：可以实时看到页面，方便排错
技术栈确定：
核心：Selenium 4.x + Python 3.8+语言：Python浏览器：Edge WebDriver（Windows系统自带）配置：PyYAML数据处理：Pandas（可选）
这个选择过程，我花了整整一个月时间，走了很多弯路。
3
第二章：环境搭建的折磨
3.1
2.1 WebDriver 版本地狱
第一个就踩大坑：WebDriver 版本不匹配。
问题现象：
selenium.common.exceptions.SessionNotCreatedException: Message: session not created: This version of MSEdgeDriver only supports MSEdge version 120.0.xxx or higher
我当时的崩溃：我刚装好环境，第一次运行就报错。我完全不懂这个错误是什么意思，花了一下午研究。
解决过程：
先检查 Edge 浏览器版本
打开 Edge 浏览器
右上角三个点 → 设置
点击左侧"关于 Microsoft Edge"
记录版本号：120.0.2210.91
去下载 WebDriver
尝试了 5-6 个版本
最后才发现要精确匹配
这个问题折磨了我整整两天时间，当时真的很想放弃。
最终解决方案：
# 写个版本检查函数，自动提示用户def check_webdriver_compatibility(): """检查Edge和WebDriver版本兼容性""" import subprocess # 获取Edge版本 try: result = subprocess.run(['reg', 'query', 'HKEY_CURRENT_USER\\Software\\Microsoft\\Edge\\BLBeacon', '/v', 'version'], capture_output=True, text=True) edge_version = result.stdout.split('\\n')[0].strip() except: print("无法获取Edge版本") return False # 获取WebDriver版本 try: result = subprocess.run(['msedgedriver', '--version'], capture_output=True, text=True) driver_version = result.stdout.strip() except: print("无法获取WebDriver版本") return False # 比较主版本 edge_major = int(edge_version.split('.')[0]) driver_major = int(driver_version.split('.')[0]) if edge_major != driver_major: print("="*60) print("⚠️ 版本不匹配！") print(f"Edge版本: {edge_version}") print(f"WebDriver版本: {driver_version}") print("") print("请下载对应版本的WebDriver:") print("https://developer.microsoft.com/en-us/microsoft-edge/tools/webdriver/") print("="*60) return False return True# 在程序启动时调用if not check_webdriver_compatibility(): input("请修复版本问题后按回车继续...") sys.exit(1)
3.2
2.2 依赖地狱
Python 依赖管理也让我踩了不少坑。
问题一：Selenium 版本冲突
一开始我直接用 pip install selenium 安装，结果安装了最新版 4.15，但是代码里用的是 3.x 的 API。
现象：
AttributeError: module 'selenium' has no attribute 'webdriver'
解决：
# 卸载最新版pip uninstall selenium# 安装指定版本pip install selenium==4.11.0
问题二：openpyxl 依赖问题
添加 Excel 导出功能时，openpyxl 报错缺少 lxml，安装了 lxml 又报错版本冲突。
解决：
# 使用requirements.txt统一管理依赖pip install -r requirements.txt
3.3
2.3 目录权限问题
Windows 系统还有文件权限问题，比如某些目录创建失败、文件写入被拒绝。
现象：
PermissionError: [Errno 13] Permission denied: './output'
解决：
# 以管理员权限运行# 或提前创建目录并设置权限import osos.makedirs("./output", exist_ok=True)os.makedirs("./logs", exist_ok=True)
这些问题虽然都很基础，但都让我花了不少时间排查。
4
第三章：登录功能的痛苦实现
4.1
3.1 扫码登录的第一次尝试
登录功能是我遇到的第一个大难题。
最初的思路： 直接用 selenium 模拟扫码
尝试过程：
获取二维码图片
qr_element = driver.find_element(By.CLASS_NAME, "qrcode")qr_image = qr_element.screenshot_as_png()
本地识别二维码
import qrcodefrom PIL import Image# 用qrcode库识别img = Image.open(qr_image)qr_data = qrcode.decode(img)
模拟扫码登录
结果：完全失败！
失败原因：
二维码识别率低，经常识别失败
识别出了 URL，但不知道怎么处理
模拟扫码的登录流程太复杂，涉及到加密、验证码等
这个问题让我卡了一周，查阅了大量资料，还是解决不了。
4.2
3.2 Cookie 管理的第一次尝试
扫码走不通，我转向 Cookie 方案。
思路： 手动登录一次，保存 Cookie，之后直接用 Cookie
第一次实现：
# 最简单的Cookie保存def save_cookie_simple(): cookies = driver.get_cookies() with open('cookie.txt', 'w') as f: for cookie in cookies: f.write(f"{cookie['name']}={cookie['value']}\n")
问题一：Cookie 无法加载
第二次运行时：
# 尝试加载Cookiewith open('cookie.txt', 'r') as f: for line in f: name, value = line.strip().split('=') driver.add_cookie({'name': name, 'value': value})
现象：
selenium.common.exceptions.InvalidCookieDomainException: Message: Cookie is only valid for the domain that it was set for
研究过程：
修改：
# 先访问网站driver.get("https://www.example.com/")# 再加载Cookiefor cookie in cookies: driver.add_cookie(cookie)
问题二：Cookie 不生效
Cookie 加载成功了，但还是显示未登录。
现象：
# 检查页面print(driver.page_source) # 还是登录页面的HTML
尝试的方案：
刷新页面
driver.refresh()
清除所有 Cookie 后重新添加
driver.delete_all_cookies()for cookie in cookies: driver.add_cookie(cookie)
检查 Cookie 的 domain 是否匹配
# 发现Cookie的domain是.goofish.com# 但实际访问的URL可能是h5.goofish.com
问题三：Cookie 有效期未知
我根本不知道 Cookie 能保存多久，可能明天就失效，也可能能用一个月。
解决：
# 添加过期时间检查cookie_data = { 'cookies': cookies, 'saved_at': datetime.now().isoformat(), 'expires_at': (datetime.now() + timedelta(days=7)).isoformat() # 设置7天有效期}
4.3
3.3 最终的 Cookie 管理方案
经过无数次失败，我终于搞定了：
完整的 Cookie 管理流程：
class CookieManager: def save(self): """保存Cookie到文件""" cookies = self.driver.get_cookies() cookie_data = { 'cookies': cookies, 'saved_at': datetime.now().strftime('%Y-%m-%dT%H:%M:%S'), 'expires_at': (datetime.now() + timedelta(days=7)).strftime('%Y-%m-%dT%H:%M:%S') } with open(self.cookie_file, 'w', encoding='utf-8') as f: json.dump(cookie_data, f, ensure_ascii=False, indent=2) logger.info("Cookie已保存") def load(self): """从文件加载Cookie""" if not os.path.exists(self.cookie_file): logger.info("Cookie文件不存在") return False with open(self.cookie_file, 'r', encoding='utf-8') as f: cookie_data = json.load(f) # 检查过期 expires_at = datetime.fromisoformat(cookie_data['expires_at']) if datetime.now() > expires_at: logger.info("Cookie已过期") return False # 检查域名 current_url = self.driver.current_url or '' if 'example.com' not in current_url: logger.warning("当前未访问目标域名") return False # 添加Cookie success_count = 0 for cookie in cookie_data['cookies']: try: self.driver.add_cookie(cookie) success_count += 1 except: continue logger.info(f"成功加载 {success_count}/{len(cookie_data['cookies'])} 个Cookie") return success_count > 0 def check_login(self): """验证登录状态""" # 刷新页面 self.driver.refresh() time.sleep(3) # 检查是否有登录按钮 try: login_btn = self.driver.find_element(By.CSS_SELECTOR, "a[href*='login']") if login_btn.is_displayed(): return False except: pass # 检查URL current_url = self.driver.current_url if 'login' in current_url.lower() or 'auth' in current_url.lower(): return False return True
完整的登录流程：
def login_with_persistence(self): """持久化登录""" # 第一步：先访问网站（建立域名上下文） self.driver.get("https://www.example.com/") time.sleep(2) # 第二步：尝试加载Cookie if self._load_cookie(): # 第三步：刷新页面使Cookie生效 self.driver.refresh() time.sleep(3) # 第四步：验证登录状态 if self._check_login(): logger.info("✅ 使用保存的Cookie登录成功") return True # 第五步：Cookie失效，重新扫码登录 logger.info("⚠️ Cookie已失效，请重新登录") self.login_by_qrcode() # 第六步：保存新Cookie self._save_cookie() logger.info("✅ Cookie已保存（有效期7天）") return True
这个登录功能，我前后花了快一个月时间才搞定！
关键突破点：
必须先访问域名再添加 Cookie
Cookie 加载后必须刷新页面
需要验证 Cookie 是否真的生效
设置合理的过期时间（7 天）
5
第四章：数据解析的深度优化
5.1
4.1 第一次解析：简单粗暴
最初的解析逻辑非常简单粗暴：
def parse_simple(self, element): """最简单的解析""" try: # 提取标题 title = element.find_element(By.CSS_SELECTOR, "[class*='title']").text # 提取价格 price_text = element.find_element(By.CSS_SELECTOR, "[class*='price']").text price = float(re.search(r'(\d+)', price_text).group(1)) # 提取链接 link = element.find_element(By.TAG_NAME, 'a').get_attribute('href') return { 'title': title, 'price': price, 'link': link } except: return None
问题： 漏洞百出
有些元素找不到就直接报错
价格提取不准确
没有容错机制
5.2
4.2 多价格问题的漫长调试
这是我最头疼的问题。
问题现象：有些商品有多个价格显示：
¥999（原价）¥199（活动价）
我的代码只能提取到第一个价格 999，漏掉了实际的成交价 199。
调试过程：
尝试 1：换正则表达式
# 试试多种正则patterns = [ r'(\d+\.?\d*)', # 简单的数字 r'¥\s*(\d+\.?\d*)', # 带¥符号 r'￥\s*(\d+\.?\d*)', # 中文符号 r'(?:¥|￥)\s*(\d+\.?\d*)', # 混合]for pattern in patterns: match = re.search(pattern, text) if match: price = float(match.group(1)) break
问题： 还是只能取到一个
尝试 2：提取所有数字再筛选
# 提取所有数字all_numbers = re.findall(r'(\d+\.?\d*)', text)print(f"所有数字: {all_numbers}") # ['999', '199']# 然后取最小值if all_numbers: prices = [float(n) for n in all_numbers] main_price = min(prices)
问题： 有时会提取到不相关的数字（比如浏览量、时间等）
尝试 3：查找所有价格元素
# 找所有包含price的元素price_elements = element.find_elements(By.CSS_SELECTOR, "[class*='price']")for elem in price_elements: text = elem.text.strip() print(f"价格元素: {text}")
输出示例：
价格元素1: ¥999价格元素2: ¥199价格元素3: 已卖出5件
终于找到思路了！
最终解决方案：
def extract_all_prices(self, element): """提取所有价格""" # 1. 查找所有价格元素 price_elements = element.find_elements(By.CSS_SELECTOR, "[class*='price']") prices = [] for elem in price_elements: text = elem.text.strip() if not text: continue # 2. 提取数字 match = re.search(r'(\d+\.?\d*)', text) if match: price_value = float(match.group(1)) # 3. 过滤掉明显不是价格的数字 if price_value > 0 and price_value < 100000: # 价格范围 0-100000 prices.append(price_value) # 4. 去重并排序 unique_prices = sorted(set(prices)) # 5. 构建返回数据 return { 'main_price': unique_prices[0] if unique_prices else 0.0, # 最低价 'all_prices': ','.join(map(str, unique_prices)), # 所有价格 'price_text': ' '.join([f"¥{p}" for p in unique_prices]) # 原始文本 }
效果对比：
修改前：price: 999.0price_text: ¥999修改后：main_price: 199.0all_prices: 199,999price_text: ¥999 ¥199
这个问题的解决，我花了整整两周时间调试，尝试了无数种方案。
5.3
4.3 去重逻辑的崩溃
这是另一个让我崩溃的问题。
问题现象：
获取 160 个商品去重: 移除了 160 个重复商品去重后剩余 0 个商品
原因分析：
我之前的数据结构有id字段，是商品的唯一标识。但我后来重构 CSV 字段时，把id字段删除了！
# 旧的CSV字段fields = ['id', 'title', 'price', ...]# 新的CSV字段（删除了id）fields = ['title', 'price', ...] # ❌ 没有id了
去重代码还在用 id：
def deduplicate_by_id(self, products): for product in products: product_id = product.get('id', '') # ❌ id字段已删除，始终为空 if product_id not in seen_ids: seen_ids.add(product_id) unique.append(product)
结果： 所有商品的 id 都是空，都被当成重复删除了！
发现问题时，我差点吐血：
解决方案：
def deduplicate_by_link(self, products): """使用链接去重（而不是id）""" seen_links = set() unique = [] for product in products: link = product.get('link', '') # ✅ 使用链接 if link and link not in seen_links: seen_links.add(link) unique.append(product) return unique
去重效果对比：
修改前（用id）：获取 160 个商品去重: 移除了 160 个重复商品（按id）去重后剩余 0 个商品修改后（用link）：获取 160 个商品去重: 移除了 0 个重复商品（按link）去重后剩余 160 个商品
这个问题虽然简单，但让我排查了三天，因为：
一开始根本没意识到是字段问题
以为是去重逻辑写错了
重写了三次去重代码都不管用
最后才想到检查一下返回值里到底有没有 id 字段
5.4
4.4 实时保存功能的实现
另一个问题是程序中断导致数据丢失。
问题场景：
程序运行了10分钟，爬取了300个商品程序意外中断（网络问题、电脑卡死、手动停止等）检查CSV文件：只保存了前50个
原因：我的代码是最后统一保存：
all_products = []for keyword in keywords: products = search(keyword) all_products.extend(products)# 最后才保存一次save_to_csv(all_products)
程序中途中断，前面的数据就全部丢了！
解决方案：实时追加
def main(): timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') csv_filename = f"products_{timestamp}.csv" csv_exporter = CSVExporter(config) # 维护全局链接集合，用于跨关键词去重 all_links = set() for i, keyword in enumerate(keywords, 1): print(f"[{i}/{len(keywords)}] 搜索关键词: {keyword}") # 搜索商品 products = search(keyword) # 添加关键词标签 for product in products: product['keywords'] = [keyword] # 实时保存到CSV for product in products: link = product.get('link', '') if link not in all_links: all_links.add(link) csv_exporter.append(product, csv_filename) # ✅ 立即保存 print(f"已保存 {len(products)} 个商品到 {csv_filename}") print(f"\n✅ 完成！共采集 {len(all_links)} 个商品")
实时保存的优势：
程序中断不丢失已爬取的数据
可以实时看到数据增长
便于调试和监控
可以随时查看 CSV 文件
6
第五章：反爬与稳定性优化
6.1
5.1 第一次被封：完全懵了
第一次被封杀，我完全懵了。
现象：
程序正常运行突然报错：403 Forbidden
我当时的反应：
为什么我都没怎么操作就被封了？
是不是平台检测到我是机器人了？
怎么办啊，账号是不是废了？
排查过程：
检查请求频率
# 统计请求间隔import timelast_request_time = time.time()request_count = 0def make_request(): global last_request_time, request_count now = time.time() interval = now - last_request_time request_count += 1 print(f"请求间隔: {interval:.2f}秒，总请求数: {request_count}") last_request_time = now
发现： 我每秒请求 1 次，10 分钟就是 600 次请求，确实太频繁了。
第一次优化：降低频率
time.sleep(3) # 每个请求间隔3秒
结果： 还是被封，只是慢了一点。
6.2
5.2 浏览器特征隐藏
我发现平台会检测 WebDriver 的自动化特征。
检测方式：
// 平台检测的代码（推测）navigator.webdriver // 如果是自动化工具，这个属性不是undefinedwindow.chrome // Chrome相关的对象window.navigator.plugins.length === 0
解决方案：
def setup_stealth_browser(self): """配置隐蔽浏览器""" edge_options = Options() # 1. 移除自动化特征 edge_options.add_argument('--disable-blink-features=AutomationControlled') edge_options.add_experimental_option("excludeSwitches", ["enable-automation"]) edge_options.add_experimental_option('useAutomationExtension', False) # 2. 修改navigator.webdriver self.driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})") # 3. 添加真实的User-Agent edge_options.add_argument('--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') return edge_options
6.3
5.3 随机延迟与人类行为模拟
进一步优化，模拟人类行为：
import randomdef human_like_delay(min_seconds=1, max_seconds=3): """人类行为的随机延迟""" delay = random.uniform(min_seconds, max_seconds) time.sleep(delay)def random_scroll(): """随机滚动（模拟人类浏览行为）""" # 随机滚动距离 scroll_distance = random.randint(300, 800) self.driver.execute_script(f"window.scrollBy(0, {scroll_distance});") time.sleep(random.uniform(0.5, 2))def mouse_hover_simulation(self, element): """模拟鼠标悬停""" from selenium.webdriver.common.action_chains import ActionChains actions = ActionChains(self.driver) actions.move_to_element(element).perform() time.sleep(random.uniform(0.3, 1))
6.4
5.4 异常重试机制
from functools import wrapsimport logginglogger = logging.getLogger(__name__)def retry(max_retries=3, delay=2): """重试装饰器""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): for i in range(max_retries): try: return func(*args, **kwargs) except Exception as e: if i == max_retries - 1: logger.error(f"重试{max_retries}次后仍失败: {e}") raise logger.warning(f"第{i+1}次重试: {e}") time.sleep(delay) return wrapper return decorator@retry(max_retries=3, delay=2)def fetch_page(self, url): """带重试的页面请求""" self.driver.get(url)
6.5
5.5 断点续传功能
为了避免程序中断导致的数据丢失，我实现了断点续传：
# 保存已处理的链接processed_links_file = "processed_links.json"def load_processed_links(): """加载已处理的链接""" if os.path.exists(processed_links_file): with open(processed_links_file, 'r') as f: return set(json.load(f)) return set()def save_processed_links(links): """保存已处理的链接""" with open(processed_links_file, 'w') as f: json.dump(list(links), f)def main(): # 加载已处理的链接 processed = load_processed_links() # 获取新数据 new_products = [p for p in all_products if p['link'] not in processed] # 保存新处理的链接 for product in new_products: processed.add(product['link']) save_processed_links(processed)
7
第六章：最终项目架构
7.1
6.1 完整的目录结构
data_collector/├── core/ # 核心模块│ ├── selenium_collector.py # 浏览器采集器（含Cookie管理）│ └── __init__.py├── storage/ # 存储模块│ ├── csv_exporter.py # CSV导出│ ├── excel_exporter.py # Excel导出│ └── data_manager.py # 数据管理├── filter/ # 过滤模块│ └── deduplication.py # 去重逻辑├── config/ # 配置文件│ └── settings.yaml # 配置项├── cookie/ # Cookie目录│ └── session.json # 登录Cookie（自动生成）├── main.py # 主程序入口├── requirements.txt # 依赖列表├── output/ # 输出目录│ └── *.csv # 商品数据├── logs/ # 日志目录│ └── app.log # 程序日志├── README.md # 使用文档├── QUICKSTART.md # 快速开始├── CSV_FIELDS.md # 字段说明└── CHANGELOG.md # 更新日志
7.2
6.2 核心代码实现
6.2.1 浏览器采集器
class SeleniumCollector: """浏览器采集器""" def __init__(self, config): self.config = config self.driver = None self.is_logged_in = False self.cookie_dir = "./cookie" self.cookie_file = os.path.join(self.cookie_dir, "session.json") os.makedirs(self.cookie_dir, exist_ok=True) def init_driver(self, headless=False): """初始化浏览器""" edge_options = Options() if headless: edge_options.add_argument('--headless') # 反爬优化 edge_options.add_argument('--no-sandbox') edge_options.add_argument('--disable-dev-shm-usage') edge_options.add_argument('--disable-blink-features=AutomationControlled') edge_options.add_experimental_option("excludeSwitches", ["enable-automation"]) edge_options.add_experimental_option('useAutomationExtension', False) edge_options.add_argument('--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') self.driver = webdriver.Edge(options=edge_options) self.driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})") logger.info("浏览器初始化成功") return True def login_with_persistence(self): """持久化登录""" # 1. 先访问网站 self.driver.get("https://www.example.com/") time.sleep(2) # 2. 尝试加载Cookie if self._load_cookie(): self.driver.refresh() time.sleep(3) if self._check_login(): logger.info("✅ 使用保存的Cookie登录成功") self.is_logged_in = True return True # 3. 重新登录 logger.info("⚠️ Cookie已失效，请重新登录") if not self.login_by_qrcode(): logger.error("登录失败") return False # 4. 保存新Cookie self._save_cookie() self.is_logged_in = True return True def search_products(self, keyword, scroll_count=1): """搜索商品""" if not self.is_logged_in: logger.error("未登录，请先登录") return [] search_url = f"https://www.example.com/search?q={keyword}" self.driver.get(search_url) products = [] for i in range(scroll_count): logger.info(f"滚动加载... ({i+1}/{scroll_count})") # 滚动 self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);") time.sleep(2) # 解析当前页 page_products = self._parse_current_page() products.extend(page_products) if len(page_products) == 0 and i > 0: logger.info("未获取到新商品，可能已加载完成") break return self._deduplicate(products)
6.2.2 CSV 导出器
class CSVExporter: """CSV导出器""" def __init__(self, config): self.config = config self.output_dir = config['storage']['output_dir'] os.makedirs(self.output_dir, exist_ok=True) def _get_fieldnames(self): return [ 'title', # 商品标题 'link', # 商品链接 'main_price', # 主要价格（最低价） 'all_prices', # 所有价格（逗号分隔） 'price_text', # 价格文本（原始格式） 'location', # 位置（卖家地址） 'want_count', # 想要数 'sold_count', # 卖出件数 'keywords', # 关键词 'crawl_time', # 爬取时间 ] def _prepare_row(self, product): return { 'title': product.get('title', ''), 'link': product.get('link', ''), 'main_price': product.get('main_price', 0), 'all_prices': product.get('all_prices', ''), 'price_text': product.get('price_text', ''), 'location': product.get('location', ''), 'want_count': product.get('want_count', 0), 'sold_count': product.get('sold_count', 0), 'keywords': '|'.join(product.get('keywords', [])), 'crawl_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), } def append(self, product, filename): """追加数据到CSV（实时保存）""" file_exists = os.path.exists(os.path.join(self.output_dir, filename)) with open(os.path.join(self.output_dir, filename), 'a', encoding='utf-8-sig', newline='') as f: writer = csv.DictWriter(f, fieldnames=self._get_fieldnames()) if not file_exists: writer.writeheader() row = self._prepare_row(product) writer.writerow(row)
6.2.3 去重模块
class Deduplicator: """去重器""" def deduplicate(self, products): """基于链接去重""" seen_links = set() unique = [] for product in products: link = product.get('link', '') if link and link not in seen_links: seen_links.add(link) unique.append(product) return unique
7.3
6.3 商品数据字段说明
字段名 类型 说明 示例──────────────────────────────────────────────────────title 文本 商品标题 商品标题示例link 文本 商品链接（完整URL） https://www.example.com/item/123main_price 数值 主要价格（最低价） 199.0all_prices 文本 所有价格（逗号分隔） 199,999price_text 文本 价格文本（原始格式） ¥999 ¥199location 文本 位置（卖家地址） 广东发货极快want_count 整数 想要数 58sold_count 整数 卖出件数 5keywords 文本 关键词（|分隔） 关键词1|关键词2crawl_time 文本 爬取时间 2026-02-27 10:30:00
8
第七章：使用指南
8.1
7.1 使用流程
首次运行需要扫码登录
7 天内运行自动使用 Cookie
可自定义关键词和搜索页数
实时保存数据到 CSV
生成可视化图表
9
第八章：总结与反思
9.1
8.1 开发时间统计
总计： 约 2.5 个月
9.2
8.2 遇到的困难
WebDriver 版本不匹配 - 解决方案：精确版本匹配
Cookie 无法加载 - 解决方案：先访问域名再添加
多价格提取不完整 - 解决方案：提取所有并去重
去重逻辑删除所有商品 - 解决方案：改用 link 字段
程序中断数据丢失 - 解决方案：实时追加保存
频繁被封杀 - 解决方案：降低频率、隐藏特征
元素选择器失效 - 解决方案：多种选择器备选
9.3
8.3 技术收获
Selenium 深度应用
Cookie 管理和会话保持
动态内容处理
浏览器特征隐藏
数据处理
问题解决
系统化的问题排查方法
多种方案尝试验证
失败后的优化迭代
9.4
8.4 项目特色
Cookie 持久化（7 天有效期）
实时 CSV 保存（防数据丢失）
多价格提取（完整记录）
基于链接去重（跨关键词）
隐蔽浏览器（反爬优化）
详细日志记录（便于调试）
9.5
8.5 后续优化方向
功能扩展
支持更多电商平台
添加商品详情页深度采集
实现定时自动任务
添加数据清洗和标准化
技术升级
异步采集提升速度
分布式采集支持
数据可视化增强
Web 管理界面
数据库存储
10
结语
这个项目的开发过程，让我深刻体会到：
1. 技术选型的重要性不要一开始就选最复杂的方案，要综合考虑稳定性、开发难度、维护成本。
2. 问题导向的开发每个问题都是优化机会，不要怕遇到问题。
3. 用户体验优先Cookie 持久化、实时保存等功能虽然增加了复杂度，但极大提升了使用体验。
4. 代码可维护性模块化设计、清晰的职责分离，让后续维护和扩展变得容易。
5. 持续学习的态度技术领域变化很快，要保持学习的态度，不断优化和改进。
整个开发过程虽然曲折，但也让我成长了很多。希望本文能对有类似需求的朋友有所帮助。
声明： 本文仅供技术交流和学习使用。实际应用时请遵守相关平台的使用协议和法律法规，不要用于商业用途或大规模采集。
