㊗️本期内容已收录至专栏《Python爬虫实战》,持续完善知识体系与项目实战,建议先订阅收藏,后续查阅更方便~ ㊙️本期爬虫难度指数:⭐⭐⭐☆☆(进阶级) 🉐福利: 一次订阅后,专栏内的所有文章可永久免费看,持续更新中,保底1000+(篇)硬核实战内容。
全文目录:
-
-
- 🌟 开篇语
- 0️⃣ 前言(Preface)
- 1️⃣ 摘要(Abstract)
- 2️⃣ 背景与需求(Why)
- 3️⃣ 合规与注意事项(必写)
- 4️⃣ 技术选型与整体流程(What/How)
- 5️⃣ 环境准备与依赖安装(可复现)
- 6️⃣ 核心实现:请求层(Fetcher)
- 7️⃣ 核心实现:解析层(Parser)
- 8️⃣ 数据存储与导出(Storage)
- 9️⃣ 运行方式与结果展示(必写)
- 🔟 常见问题与排错(强烈建议写)
- 1️⃣1️⃣ 进阶优化(可选但加分)
- 1️⃣2️⃣ 总结与延伸阅读
- 🌟 文末
-
- ✅ 专栏持续更新中|建议收藏 + 订阅
- ✅ 互动征集
- ✅ 免责声明
-
🌟 开篇语
哈喽,各位小伙伴们你们好呀~我是【喵手】。 运营社区: C站 / 掘金 / 腾讯云 / 阿里云 / 华为云 / 51CTO 欢迎大家常来逛逛,一起学习,一起进步~🌟
我长期专注 Python 爬虫工程化实战,主理专栏👉 《Python爬虫实战》:从采集策略到反爬对抗,从数据清洗到分布式调度,持续输出可复用的方法论与可落地案例。内容主打一个“能跑、能用、能扩展”,让数据价值真正做到——抓得到、洗得净、用得上。
📌 专栏食用指南(建议收藏)
- ✅ 入门基础:环境搭建 / 请求与解析 / 数据落库
- ✅ 进阶提升:登录鉴权 / 动态渲染 / 反爬对抗
- ✅ 工程实战:异步并发 / 分布式调度 / 监控与容错
- ✅ 项目落地:数据治理 / 可视化分析 / 场景化应用
📣 专栏推广时间:如果你想系统学爬虫,而不是碎片化东拼西凑,欢迎订阅专栏👉《Python爬虫实战》👈,一次订阅后,专栏内的所有文章可永久免费阅读,持续更新中。 💕订阅后更新会优先推送,按目录学习更高效💯~
0️⃣ 前言(Preface)
这篇文章将带你手搓一个基于 Python 和 Requests/BeautifulSoup 的增量爬虫系统,专门用于盯防“修补服务活动日历”,最终产出一个自带去重逻辑的本地 SQLite 数据库及新增活动控制台提醒器。 读完这篇硬核长文,你将掌握:
1️⃣ 摘要(Abstract)
本文系统讲解了如何使用 Python 的 Requests 库结合 SQLite 数据库,对修补服务类活动日历进行增量数据采集与状态快照保存,最终实现并模拟新活动上线的自动化提醒。 读者通过本文将获得:
2️⃣ 背景与需求(Why)
为什么我们要盯上“修补服务活动日历”?在循环经济和可持续发展的大潮下,各类社区修补活动(如家电维修进社区、旧衣物改造快闪店)层出不穷。但这些活动往往档期短、名额有限。纯靠人工刷新页面不仅费时费力,还容易错失良机。通过爬虫实现信息聚合与自动化提醒,是数据赋能生活的典型案例。
目标站点假设与字段清单: 假设目标是一个社区活动聚合平台(如某某同城活动网)。 我们需要精准抽取的字段如下:
- 活动名 (Event_Name):如“周末家电义诊大放送”。
- 服务类型 (Service_Type):如“电子产品维修”、“衣物缝补”。
- 时间 (Event_Time):活动的起止时间。
- 地点 (Location):详细地址或经纬度。
- 状态 (Status):如“报名中”、“已满额”、“已结束”。
- 链接 (Source_URL):溯源抓取详情的原始链接。
3️⃣ 合规与注意事项(必写)
作为老手,每次动手前必须默念“爬虫规范”。技术无罪,但使用技术的方式必须克制。
- 遵循 robots.txt:在目标网站根目录查看该文件,明确哪些路径是站长允许抓取的。如果明确 Disallow,请寻找其他合法数据源或公开 API。
- 频率控制(Rate Limiting):坚决反对攻击式并发。我们的目的是获取数据,不是做压力测试。代码中必须强制加入 time.sleep() 或随机延时,单 IP 访问频率建议控制在 1-3秒/次。
- 数据边界:仅采集公开展示的活动日历信息,绝不尝试绕过登录鉴权窃取用户个人隐私(如报名者的手机号、姓名),绝不触碰付费墙背后的特权数据。
4️⃣ 技术选型与整体流程(What/How)
修补活动日历通常是一个列表+详情的结构。本篇实战我们采用 API(抓列表) + 静态解析(抓详情) 的混合模式。
流程图解: 请求列表API ➡️ 提取活动ID/链接 ➡️ 比对本地库(增量判断) ➡️ (若存在)跳过 / (若不存在) 请求详情页 ➡️ 解析清洗字段 ➡️ 落库存储 ➡️ 触发新增提醒。
为什么选 Requests + BeautifulSoup4 + SQLite? 对于日更量级在千百级别的活动日历,上 Scrapy 属于高射炮打蚊子,太重了。Playwright 虽然能对抗强 JS 混淆,但开销极大。Requests + BS4 轻量、敏捷,配合 SQLite 作为本地轻量级数据库完美实现增量比对状态机,是性价比最高的黄金组合。
5️⃣ 环境准备与依赖安装(可复现)
请确保你的开发环境为 Python 3.8+(推荐 3.10,对类型提示支持更好)。
在终端执行以下命令安装依赖:
pip install requests beautifulsoup4 lxml loguru fake-useragent
(注:lxml 是 BS4 的底层解析引擎,比自带的 html.parser 速度快几倍;loguru 是极简但强大的日志库;fake-useragent 用于伪造浏览器指纹)
推荐项目结构(工程化思维):
repair_calendar_scraper/
│
├── config.py # 全局配置(请求头、重试次数、数据库名)
├── db_manager.py # SQLite 数据库操作层
├── fetcher.py # 网络请求层(负责下载源码/JSON)
├── parser.py # 页面解析层(负责抽取那6个核心字段)
├── main.py # 业务主逻辑(组装上述模块)
└── data/ # 数据存储目录(存放 SQLite 文件或 CSV)
6️⃣ 核心实现:请求层(Fetcher)
不要直接用 requests.get(url)!网络是脆弱的,你需要重试机制、超时控制和浏览器伪装。我们将这一层封装在 fetcher.py 中。
# fetcher.py
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from fake_useragent import UserAgent
from loguru import logger
import time
import random
class RobustFetcher:
def __init__(self):
self.session = requests.Session()
self.ua = UserAgent()
# 定义重试策略:总共重试3次,退避系数为1秒,针对特定的HTTP状态码重试
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["HEAD", "GET", "OPTIONS"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
self.session.mount("https://", adapter)
self.session.mount("http://", adapter)
def get_html(self, url, referer=None):
"""
带健壮性处理的 GET 请求
"""
headers = {
"User-Agent": self.ua.random,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Connection": "keep-alive"
}
if referer:
headers["Referer"] = referer
# 礼貌性延时
time.sleep(random.uniform(1.0, 2.5))
try:
logger.info(f"正在抓取: {url}")
# 设置 timeout 避免死等:连接3秒,读取10秒
response = self.session.get(url, headers=headers, timeout=(3, 10))
response.raise_for_status() # 非 200 抛出异常
# 处理编码问题,防止乱码
response.encoding = response.apparent_encoding
return response.text
except requests.exceptions.RequestException as e:
logger.error(f"抓取失败 {url}: {e}")
return None
实战解析: 这里引入了 urllib3 的 Retry。当遇到 502 Bad Gateway 或者 429 Too Many Requests 时,它会自动等待并重试,极大地提高了爬虫的存活率。同时通过 response.apparent_encoding 动态推断网页编码,彻底消灭中文乱码。
7️⃣ 核心实现:解析层(Parser)
假设我们从接口拿到了一堆详情页 URL(列表提取逻辑会在 Main 中体现),现在要在 parser.py 中写详情页抽取的逻辑。
这里考验的是容错能力。页面结构经常变,某个活动可能没有填写“地点”,你的代码不能因此崩溃。
# parser.py
from bs4 import BeautifulSoup
from loguru import logger
import re
class EventParser:
@staticmethod
def parse_detail(html_content, source_url):
"""
使用 BeautifulSoup 解析 HTML,提取核心字段。
采用容错写法,防止页面结构残缺导致崩溃。
"""
if not html_content:
return None
soup = BeautifulSoup(html_content, 'lxml')
event_data = {
"Source_URL": source_url,
"Event_Name": "未知",
"Service_Type": "通用修补",
"Event_Time": "待定",
"Location": "待定",
"Status": "未知"
}
try:
# 1. 活动名 (假设在一个 h1 标签,class 为 event-title)
title_node = soup.find('h1', class_='event-title')
if title_node:
event_data["Event_Name"] = title_node.get_text(strip=True)
# 2. 服务类型 (假设在 class 为 tag-service 的 span 中)
type_node = soup.find('span', class_='tag-service')
if type_node:
event_data["Service_Type"] = type_node.get_text(strip=True)
# 3. 时间 (假设使用正则从一段文本中提取)
# 例如: <p class="time-info"><i class="icon-clock"></i> 活动时间:2023-10-25 14:00至18:00 </p>
time_node = soup.find('p', class_='time-info')
if time_node:
raw_time = time_node.get_text(strip=True)
# 使用正则清洗掉无用的前缀
time_match = re.search(r'时间:(.+)', raw_time)
if time_match:
event_data["Event_Time"] = time_match.group(1)
# 4. 地点
loc_node = soup.find('div', class_='address-detail')
if loc_node:
event_data["Location"] = loc_node.get_text(strip=True)
# 5. 状态 (假设根据 class 的不同显示不同状态)
status_node = soup.find('div', class_=re.compile('^status-btn'))
if status_node:
event_data["Status"] = status_node.get_text(strip=True)
return event_data
except Exception as e:
logger.error(f"解析页面报错 {source_url}: {str(e)}")
# 即使报错,也将尽力解析出的残缺数据返回,避免彻底丢失
return event_data
8️⃣ 数据存储与导出(Storage)
增量抓取的关键在于有一个“参照物”。SQLite 是最完美的参照物载体。 我们将根据 Source_URL 或者活动的唯一 ID 生成一个哈希值作为主键,利用数据库的 UNIQUE 约束来实现绝对的去重。
数据模型结构说明 (Database Schema):
| id | VARCHAR | PRIMARY KEY | URL hash (MD5) |
| event_name | VARCHAR | NOT NULL | Name of the repair event |
| service_type | VARCHAR | E.g., Electronics, Clothing | |
| event_time | VARCHAR | Time range | |
| location | VARCHAR | Detailed address | |
| status | VARCHAR | Open, Closed, etc. | |
| source_url | TEXT | UNIQUE | Original detail link |
| crawl_time | DATETIME | DEFAULT CURRENT_TIMESTAMP | Snapshot creation time |
# db_manager.py
import sqlite3
import hashlib
from loguru import logger
class DBManager:
def __init__(self, db_path="data/repair_events.db"):
self.db_path = db_path
self._create_table()
def _get_connection(self):
return sqlite3.connect(self.db_path)
def _create_table(self):
query = """
CREATE TABLE IF NOT EXISTS events (
id VARCHAR(32) PRIMARY KEY,
event_name VARCHAR(255) NOT NULL,
service_type VARCHAR(100),
event_time VARCHAR(100),
location VARCHAR(255),
status VARCHAR(50),
source_url TEXT UNIQUE,
crawl_time DATETIME DEFAULT CURRENT_TIMESTAMP
)
"""
with self._get_connection() as conn:
conn.execute(query)
def generate_id(self, url):
"""利用 URL 生成唯一 MD5 作为主键"""
return hashlib.md5(url.encode('utf-8')).hexdigest()
def is_event_exists(self, url):
"""【增量核心逻辑】查询此链接是否已存在于库中"""
event_id = self.generate_id(url)
query = "SELECT 1 FROM events WHERE id = ?"
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute(query, (event_id,))
return cursor.fetchone() is not None
def insert_event(self, data):
"""插入新活动数据"""
event_id = self.generate_id(data['Source_URL'])
query = """
INSERT INTO events (id, event_name, service_type, event_time, location, status, source_url)
VALUES (?, ?, ?, ?, ?, ?, ?)
"""
try:
with self._get_connection() as conn:
conn.execute(query, (
event_id,
data.get('Event_Name'),
data.get('Service_Type'),
data.get('Event_Time'),
data.get('Location'),
data.get('Status'),
data.get('Source_URL')
))
return True
except sqlite3.IntegrityError:
logger.warning(f"数据重复拦截: {data['Event_Name']}")
return False
9️⃣ 运行方式与结果展示(必写)
我们将所有的模块组装在 main.py 中。为了保证代码可直接运行,这里我们会模拟一段“假设从列表页抓取到的链接合集”。
# main.py (或称为 repair_calendar_scraper.py)
import os
from fetcher import RobustFetcher
from parser import EventParser
from db_manager import DBManager
from loguru import logger
def main():
# 确保存储目录存在
os.makedirs("data", exist_ok=True)
# 实例化各个组件
fetcher = RobustFetcher()
parser = EventParser()
db = DBManager()
logger.info("=== 修补服务活动增量抓取启动 ===")
# 【实战模拟】假设我们请求了列表页的 API,解析出了以下 3 个详情页链接
# 其中 2 个是真实存在的常见结构,第 3 个是干扰项/死链以测试鲁棒性
mock_list_urls = [
"https://example-community-site.com/events/repair-101",
"https://example-community-site.com/events/clothes-202",
"https://example-community-site.com/events/already-in-db", # 假设这篇已被抓过
]
new_events_count = 0
for url in mock_list_urls:
# 1. 增量判断:检查数据库
if db.is_event_exists(url):
logger.info(f"⏭️ [增量跳过] 库中已存在,跳过抓取: {url}")
continue
logger.info(f"✨ [发现新活动] 准备抓取详情: {url}")
# 2. 请求网络 (为了演示,实际运行这里请求真实网址,如果没有网则会触发报错机制被我们捕获)
# html_content = fetcher.get_html(url)
# ====== 本地 Mock HTML 用于确保代码绝对可跑 ======
html_content = f"""
<html><body>
<h1 class="event-title">秋季家电维修进社区 ({url[–3:]})</h1>
<span class="tag-service">电子维修</span>
<p class="time-info">时间:2023-11-01 09:00至12:00</p>
<div class="address-detail">朝阳区某某街道中心广场</div>
<div class="status-btn-active">报名中</div>
</body></html>
"""
# ==================================================
# 3. 解析清洗
parsed_data = parser.parse_detail(html_content, url)
if parsed_data:
# 4. 落库存储
if db.insert_event(parsed_data):
new_events_count += 1
# 【新增提醒模拟】
print(f"\\n🔔🔔 【档期上新提醒】 🔔🔔")
print(f"活动: {parsed_data['Event_Name']}")
print(f"类型: {parsed_data['Service_Type']}")
print(f"时间: {parsed_data['Event_Time']}")
print(f"直达链接: {parsed_data['Source_URL']}\\n")
logger.success(f"=== 抓取任务完成,本次新增记录: {new_events_count} 条 ===")
if __name__ == "__main__":
main()
运行方式: 在终端运行 python main.py (如果文件名为 repair_calendar_scraper.py,则运行该文件名)。
预期控制台输出结果:
2023-10-27 10:00:00.123 | INFO | __main__:main:16 – === 修补服务活动增量抓取启动 ===
2023-10-27 10:00:00.125 | INFO | __main__:main:29 – ✨ [发现新活动] 准备抓取详情: https://example-community-site.com/events/repair-101
🔔🔔 【档期上新提醒】 🔔🔔
活动: 秋季家电维修进社区 (101)
类型: 电子维修
时间: 2023-11-01 09:00至12:00
直达链接: https://example-community-site.com/events/repair-101
2023-10-27 10:00:00.130 | INFO | __main__:main:29 – ✨ [发现新活动] 准备抓取详情: https://example-community-site.com/events/clothes-202
…
🔟 常见问题与排错(强烈建议写)
在真实的修补活动日历抓取战场上,你一定会遇到以下几个大坑,请熟记这份排雷指南:
403 Forbidden / 429 Too Many Requests 怎么办?
- 排错:这是你被反爬策略盯上了。
- 解法:首先检查 Headers,很多网站严格校验 Referer 和浏览器专属的 Header(如 sec-ch-ua)。其次,降低抓取频率。如果封了 IP,就需要引入代理池(Proxy Pool),在 requests 中通过 proxies={"http": "http://ip:port", "https": "http://ip:port"} 挂载。
HTML 抓到空壳怎么办?(源码里找不到数据)
- 排错:数据是 JS 动态渲染的,或者前端用了 Vue/React 单页应用(SPA)。
- 解法:打开浏览器开发者工具 (F12) -> Network -> 过滤 XHR/Fetch。你大概率能找到一个返回 JSON 的接口。直接抓那个接口,连 BeautifulSoup 解析都省了。如果接口加密极其复杂,再考虑上 Playwright 驱动无头浏览器硬刚。
解析经常报错(AttributeError: ‘NoneType’ object has no attribute ‘get_text’)
- 排错:这就是典型的页面结构变化,或者部分活动字段为空(比如没填地点)。
- 解法:像上面 parser.py 里演示的一样,永远不要对 soup.find() 出来的结果直接调用 .text,一定要先做 if node: 判断,保证代码的容错性。
1️⃣1️⃣ 进阶优化(可选但加分)
对于一个追求卓越的极客来说,上述代码只是 1.0 版本。如果日历数据量达到万级别,我们需要做如下进阶:
- 并发加速:使用 Python 内置的 concurrent.futures.ThreadPoolExecutor,开 5-10 个线程去并发请求详情页。注意:多线程写 SQLite 时要开启跨线程共享(check_same_thread=False)或者使用队列(Queue)集中落库。
- 自动化监控与真·消息推送:我们现在是控制台 Print。你可以接入钉钉/飞书机器人 webhook,或者使用 Server酱推送到微信。配合 Linux 的 crontab -e 或者 GitHub Actions,设置每天早中晚各跑一次,实现真正的“全自动抢档期提醒”。
1️⃣2️⃣ 总结与延伸阅读
复盘一下:我们从零搭建了一套针对“修补服务活动”的定制化爬虫系统。我们并没有盲目地全量乱爬,而是通过 SQLite 的主键约束设计了一套优雅的“增量跳过”机制;我们封装了自带重试和伪装的请求层,以及高容错的解析层。这已经是一个准企业级的数据清洗流水线雏形了。
下一步可以做什么? 如果你对这套流程已经游刃有余,建议你去了解以下进阶技术:
爬虫的世界浩瀚无垠,增量快照只是其中一颗璀璨的星。祝你在数据挖掘的旅途中玩得开心,收获满满!如果有任何模块想深入探讨,随时来找我!
🌟 文末
好啦~以上就是本期的全部内容啦!如果你在实践过程中遇到任何疑问,欢迎在评论区留言交流,我看到都会尽量回复~咱们下期见!
小伙伴们在批阅的过程中,如果觉得文章不错,欢迎点赞、收藏、关注哦~ 三连就是对我写作道路上最好的鼓励与支持! ❤️🔥
✅ 专栏持续更新中|建议收藏 + 订阅
墙裂推荐订阅专栏 👉 《Python爬虫实战》,本专栏秉承着以“入门 → 进阶 → 工程化 → 项目落地”的路线持续更新,争取让每一期内容都做到:
✅ 讲得清楚(原理)|✅ 跑得起来(代码)|✅ 用得上(场景)|✅ 扛得住(工程化)
📣 想系统提升的小伙伴:强烈建议先订阅专栏 《Python爬虫实战》,再按目录大纲顺序学习,效率十倍上升~
✅ 互动征集
想让我把【某站点/某反爬/某验证码/某分布式方案】等写成某期实战?
评论区留言告诉我你的需求,我会优先安排实现(更新)哒~
⭐️ 若喜欢我,就请关注我叭~(更新不迷路) ⭐️ 若对你有用,就请点赞支持一下叭~(给我一点点动力) ⭐️ 若有疑问,就请评论留言告诉我叭~(我会补坑 & 更新迭代)
✅ 免责声明
本文爬虫思路、相关技术和代码仅用于学习参考,对阅读本文后的进行爬虫行为的用户本作者不承担任何法律责任。
使用或者参考本项目即表示您已阅读并同意以下条款:
- 合法使用: 不得将本项目用于任何违法、违规或侵犯他人权益的行为,包括但不限于网络攻击、诈骗、绕过身份验证、未经授权的数据抓取等。
- 风险自负: 任何因使用本项目而产生的法律责任、技术风险或经济损失,由使用者自行承担,项目作者不承担任何形式的责任。
- 禁止滥用: 不得将本项目用于违法牟利、黑产活动或其他不当商业用途。
- 使用或者参考本项目即视为同意上述条款,即 “谁使用,谁负责” 。如不同意,请立即停止使用并删除本项目。!!!

