Python爬虫入门实战指南
·
作者：Albert 出处：https://blog.csdn.net/qq_33558698/article/details/157540605
欢迎转载，也请保留这段说明，谢谢🌹
1. requests 模块
1.1. 处理 get 请求
import requests
content = input('请输入你要检索的内容:')
url = f"https://www.sogou.com/web?query={content}"
headers = {
# 添加一个请求头信息 UA
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36"
}
response = requests.get(url, headers=headers)
print(response.text)
1.2. 处理 post 请求
import requests
url = "https://fanyi.baidu.com/sug"
data = {
"kw": input("请输入一个单词")
}
resp = requests.post(url, data=data)
resp.encoding = "utf-8"
# 此时拿到的是文本字符串
print(resp.text)
# 此时拿到的是 json 数据
print(resp.json())
1.3. 多参数 get 请求
import requests
url = "https://movie.douban.com/j/chart/top_list"
data = {
"type": "13",
"interval_id": "100:90",
"action": "",
"start": "0",
"limit": "20"
}
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0"
}
resp = requests.get(url, params=data, headers=headers)
# print(resp.text)
print(resp.json())
print(resp.request.url)
2. 正则表达式
2.1. 元字符
2.2. 量词
控制前面元字符出现的次数
2.3. 贪婪匹配 | 惰性匹配
3. 数据解析方式
3.1. re 解析
3.1.1. 基本使用
import re
result = re.findall("a", "我是一个abcdefga")
print(result)
# 输出：['a', 'a']
result = re.findall(r"\d+", "我今年18岁，我有20000000块")
print(result)
# 输出：['18', '20000000']
result = re.finditer(r"\d+", "我今年18岁，我有20000000块")
for item in result: # 从迭代器中拿到内容
print(item.group()) # 从匹配到的结果中拿到数据
# 输出：
# 18
# 20000000
# search 只会匹配到第一次匹配的内容
result = re.search(r"\d+", "我叫周杰伦，今年32岁，我的班级是3年2班")
print(result.group())
# 输出：32
# match 在匹配的时候，是从字符串的开头进行匹配的，类似在正则前面加上了^
result = re.match(r"\d+", "我叫周杰伦，今年32岁，我的班级是3年2班")
print(result)
3.1.2. 预加载
提前加载正则对象
# 预加载
# 提前把正则对象加载完毕
obj = re.compile(r"\d+")
# 直接把加载好的正则进行使用
result = obj.findall("我叫周杰伦，今年32岁，我的班级是3年2班")
print(result)
3.1.3. 提取分组数据
# 提取分组数据
# (?P<名字>正则)
# 提取数据的时候，需要group("名字")
s = """
<div class='西游记'><span id='10010'>中国联通</span>></div>
<div class='西游记'><span id='10086'>中国移动</span>></div>
"""
obj01 = re.compile(r"<span id='\d+'>.*?</span>")
obj02 = re.compile(r"<span id='(\d+)'>(.*?)</span>")
obj03 = re.compile(r"<span id='(?P<id>\d+)'>(?P<name>.*?)</span>")
result01 = obj01.findall(s)
result02 = obj02.findall(s)
result03 = obj03.finditer(s)
print(result01)
print(result02)
for item in result03:
id = item.group("id")
name = item.group("name")
print(id)
print(name)
3.1.4. 实操案例
爬取豆瓣 top250 电影信息
import csv
import re
import requests
# 使用with语句自动管理文件，结合csv.writer写入数据
# newline='' 避免写入时出现空行（Windows系统）
with open("top250.csv", mode="a+", encoding='utf-8-sig', newline='') as f:
# 创建csv写入对象
csv_writer = csv.writer(f)
# 移动文件指针到开头，判断是否为空文件（为空则写入表头）
f.seek(0)
if not f.readline():
csv_writer.writerow(["电影名称", "导演", "上映年份", "评分", "评价人数"])
for start in range(0, 250, 25):
url = f"https://movie.douban.com/top250?start={start}&filter="
headers = {
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0",
"Referer": "https://movie.douban.com/", # 模拟从豆瓣主页跳转，提升兼容性
}
try:
resp = requests.get(url, headers=headers, timeout=10)
if resp.status_code != 200:
print(f"第 {start // 25 + 1} 页请求失败！状态码：{resp.status_code}")
continue
resp.encoding = "utf-8"
pageSource = resp.text
# 正则表达式（保留你的原有逻辑）
obj = re.compile(
r'<div class="item">.*?'
r'<span class="title">(?P<name>.*?)</span>.*?'
r'<p>.*?导演: (?P<dao>.*?) .*?'
r'<br>(?P<year>.*?) .*?'
r'<span class="rating_num" property="v:average">(?P<score>.*?)</span>.*?'
r'<span>(?P<num>.*?)人评价</span>',
re.S
)
# 正则匹配并写入数据
result = obj.finditer(pageSource)
count = 0 # 统计当前页爬取到的电影数
for item in result:
data = item.groupdict() # 转为字典，方便处理
# 清洗数据（去除空格、多余字符）
data["name"] = data["name"].strip()
data["dao"] = data["dao"].strip()
data["year"] = data["year"].strip()
data["score"] = data["score"].strip()
data["num"] = data["num"].strip()
# 写入 CSV
csv_writer.writerow([data["name"], data["dao"], data["year"], data["score"], data["num"]])
count += 1
print(f"第 {start // 25 + 1} 页爬取完成，共 {count} 部电影")
except Exception as e:
print(f"第 {start // 25 + 1} 页爬取失败！错误信息：{str(e)}")
print("数据写入完成！")
爬取电影天堂电影信息
"""
1.提取到主页面中的每一个电影背后的url地址
1.1.拿到"2026必看热片"的HTML代码
1.2.从HTML代码中提取href值
2.访问子页面，提取电影的名称以及下载地址
2.1.拿到子页面源代码
2.2.数据提取
"""
import csv
import re
import requests
with open("2026必看热片.csv", mode="a+", encoding='utf-8-sig', newline='') as f:
# 创建csv写入对象
csv_writer = csv.writer(f)
# 移动文件指针到开头，判断是否为空文件（为空则写入表头）
f.seek(0)
if not f.readline():
csv_writer.writerow(["片名", "下载链接"])
url = "https://www.dytt8899.com/"
headers = {
"referer": "https://www.dytt8899.com/",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0"
}
try:
resp = requests.get(url, headers=headers, timeout=10)
resp.encoding = "gbk"
# 1.提取2026必看热片部分的HTML代码
obj01 = re.compile(r'2026必看热片.*?<ul>(?P<html>.*?)</ul>', re.S)
result01 = obj01.search(resp.text)
html = result01.group('html')
# 2.提取a标签中的href值
obj02 = re.compile(r"<li><a href='(?P<href>.*?)' title")
result02 = obj02.finditer(html)
obj03 = re.compile(r'<div id="Zoom">.*?◎片 名(?P<movie>.*?)<br />.*?'
r'<td style="WORD-WRAP: break-word" bgcolor="#fdfddf"><a href="(?P<download>.*?)">', re.S)
count = 0
for item in result02:
# 拼接子页面的url
child_url = url.strip('/') + item.group('href').strip()
child_resp = requests.get(child_url)
child_resp.encoding = 'gbk'
result03 = obj03.search(child_resp.text)
# 清洗数据
movie = result03.group('movie').strip()
download = result03.group('download').strip()
csv_writer.writerow([movie, download])
count += 1
print(f"第 {count} 条数据爬取完成")
except Exception as e:
print(f"电影数据爬取失败，错误信息：{str(e)}")
print("数据写入完成")
3.2. bs4 解析
3.2.1. 基本使用
from bs4 import BeautifulSoup
html = """
<ul>
<li><a href="zwj.com">张无忌</a></li>
<li id="abc"><a href="zxc.com">周星驰</a></li>
<li><a href="zbj.com">猪八戒</a></li>
<li><a href="wzt.com">武则天</a></li>
</ul>
"""
# 1.初始化BeautifulSoup对象
page = BeautifulSoup(html, "html.parser")
# page.find("标签名", attrs={"属性": "值"}) # 查找某个元素，只会找到一个结果
# page.find_all("标签名", attrs={"属性": "值"}) # 找到一堆结果
li = page.find("li", attrs={"id": "abc"})
print(li)
# 输出: <li id="abc"><a href="zxc.com">周星驰</a></li>
a = li.find("a")
print(a)
# 输出: <a href="zxc.com">周星驰</a>
print(a.text)
# 输出: 周星驰
print(a.get("href"))
# 输出: zxc.com
li_list = page.find_all("li")
for li in li_list:
a = li.find("a")
text = a.text
href = a.get("href")
print(text, href)
3.2.2. 实操案例
Boss 直聘热门职位
import csv
import requests
import urllib3
from bs4 import BeautifulSoup
with open("热门职位.csv", mode="a+", encoding='utf-8-sig', newline='') as f:
# 创建csv写入对象
csv_writer = csv.writer(f)
# 移动文件指针到开头，判断是否为空文件（为空则写入表头）
f.seek(0)
if not f.readline():
csv_writer.writerow(["职位", "薪酬", "描述"])
url = "https://www.zhipin.com/?ka=header-home-logo"
headers = {
"Referer": "https://www.zhipin.com/",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0"
}
# 抑制SSL警告（可选，避免控制台输出无关警告）
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
try:
resp = requests.get(url, headers=headers, timeout=10, verify=False)
resp.encoding = "utf-8"
# 初始化BS4对象
page = BeautifulSoup(resp.text, "html.parser")
hot_job = page.find("div", attrs={"class": "hot-job-box"})
job_list = hot_job.find_all("div", attrs={"class": "sub-li"})
for job in job_list:
name = job.find("p", attrs={"class": "name"}).text
salary = job.find("p", attrs={"class": "salary"}).text
job_text = job.find("p", attrs={"class": "job-text"})
desc_text = ""
if job_text: # 先判断job-text标签是否存在
desc_spans = job_text.find_all("span")
# 用列表推导式拼接，避免循环中重复赋值的问题
desc_list = [span.text.strip() for span in desc_spans if span.text.strip()]
desc_text = ",".join(desc_list) # 更高效的拼接方式
csv_writer.writerow([name, salary, desc_text])
print(f"已写入：职位={name}，薪酬={salary}，描述={desc_text}")
except Exception as e:
print(f"职位数据爬取失败，错误信息：{str(e)}")
print("数据写入完成")
3.3. xpath 解析
XPath 是一门在 XML 文档中查找信息的语言。
3.3.1. 基本使用
from lxml import etree
# xpath处理XML
xml = """
<library>
<author>吉多·范罗苏姆</author>
<book id="1001">
<title>Python编程：从入门到实践</title>
<author class="alick">埃里克·马瑟斯</author>
<price currency="CNY">89.00</price>
<publish_date>2020-07-01</publish_date>
</book>
<book id="1002">
<title>Java核心技术</title>
<author class="kai">凯·S·霍斯特曼</author>
<price currency="CNY">128.00</price>
<publish_date>2019-11-01</publish_date>
</book>
</library>
"""
et = etree.XML(xml)
result01 = et.xpath("/library") # 表示根节点
result02 = et.xpath("/library/book/title/text()")[0] # 得到子节点, text() 拿文本
result03 = et.xpath("/library/*/author/text()") # * 通配符，表示谁都行
result04 = et.xpath("/library/*/author[@class='kai']/text()") # [] 表示属性筛选，@属性名=值
result05 = et.xpath("/library/book/@id") # @属性，可以直接拿到属性值
print(result01)
print(result02)
print(result03)
print(result04)
print(result05)
# xpath处理HTML
html = """
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<title>Title</title>
</head>
<body>
<ul>
<li><a href="http://www.baidu.com">百度</a></li>
<li><a href="http://www.google.com">谷歌</a></li>
<li><a href="http://www.sogou.com">搜狗</a></li>
</ul>
<ol>
<li><a href="feiji">飞机</a></li>
<li><a href="dapao">大炮</a></li>
<li><a href="huoche">火车</a></li>
</ol>
<div class="job">李嘉诚</div>
<div class="common">胡辣汤</div>
</body>
</html>
"""
et = etree.XML(html)
li_list01 = et.xpath("/html/body/ul/li[2]/a/text()")
li_list02 = et.xpath("//li")
print(li_list01)
for li in li_list02:
href = li.xpath("./a/@href")[0] # ./表示当前节点
text = li.xpath("./a/text()")[0] # ./表示当前节点
print(href, text)
3.3.2. 实操案例
import csv
import requests
import urllib3
from lxml import etree
# 抑制SSL警告（可选，避免控制台输出无关警告）
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
DOMAIN_URL = "https://www.bilibili.com/video/"
CSV_FILE_PATH = "B站热搜.csv"
TARGET_URL = "https://www.remenla.com/hot/bilibili"
HEADERS = {
"Referer": "https://www.remenla.com/hot/zhihu",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0"
}
def extract_bv_from_url(url: str) -> str | None:
"""
从URL中提取BV号（适用于BV号在最后一个/之后的情况）
"""
# 按/分割字符串，取最后一部分
bv_code = url.split('/')[-1]
# 简单校验：确保提取的内容以BV开头
if bv_code.startswith('BV'):
return DOMAIN_URL + bv_code
else:
return None
def main():
with open(CSV_FILE_PATH, mode="a+", encoding='utf-8-sig', newline='') as f:
# 创建csv写入对象
csv_writer = csv.writer(f)
# 移动文件指针到开头，判断是否为空文件（为空则写入表头）
f.seek(0)
if not f.readline():
csv_writer.writerow(["标题", "描述", "链接", "热度"])
try:
resp = requests.get(TARGET_URL, headers=HEADERS, timeout=10, verify=False)
resp.raise_for_status()
resp.encoding = "utf-8"
et = etree.HTML(resp.text)
content_list = et.xpath("//div[@class='list-container']//div[@class='list-content']")
if not content_list:
raise Exception("未匹配到任何热搜条目，请检查XPath路径是否正确")
for content in content_list:
item_data = {
"title": "", # 标题
"desc": "", # 目标描述文本（可能为空）
"bv_url": "", # BV号链接
"hot_value": "" # 热度值
}
item_data["title"] = content.xpath(".//a/text()")[0].strip()
href = content.xpath(".//a/@href")[0].strip()
item_data["bv_url"] = extract_bv_from_url(href)
item_data["hot_value"] = content.xpath("//span/text()")[0].strip()
div_list = content.xpath('.//div')
for div in div_list:
# 检查当前div是否是热度div（包含icon-fire或format-number）
is_hot_div = bool(
div.xpath('./i[@class="iconfont icon-fire"]') or div.xpath('./span[@class="format-number"]'))
if not is_hot_div:
# 不是热度div，提取文本并清理
raw_text = div.xpath('./text()')[0].strip()
# 排除只有"-"的无效文本
if raw_text and raw_text != "-":
item_data["desc"] = raw_text
break # 找到有效描述就停止遍历
csv_row = [
item_data["title"],
item_data["desc"],
item_data["bv_url"],
item_data["hot_value"]
]
csv_writer.writerow(csv_row)
except Exception as e:
print(f"热搜数据爬取失败，错误信息：{str(e)}")
if __name__ == "__main__":
main()
3.4. pyquery 解析
3.4.1. 基本使用
from pyquery import PyQuery
html = """
<ul>
<li class="aaa"><a href="http://www.google.com">谷歌</a></li>
<li class="aaa"><a href="http://www.baidu.com">百度</a></li>
<li class="bbb" id="qq"><a href="http://www.qq.com">腾讯</a></li>
</ul>
"""
# 加载html内容
p = PyQuery(html)
a_list = p("a")
a = p("a").eq(0)
print(a_list)
print(a)
print("- - - - - - " * 10)
# 链式操作
a = p("ul")("li")("a")
a = p("ul li a")
print(a)
print("- - - - - - " * 10)
a = p(".aaa a") # class="aaa"
print(a)
print("- - - - - - " * 10)
a = p("#qq a") # id="qq"
print(a)
print("- - - - - - " * 10)
href = p("#qq a").attr("href")
text = p("#qq a").text()
print(text, href)
print("- - - - - - " * 10)
# 多个标签拿属性
it = p("ul li a").items()
for item in it:
href = item.attr("href")
text = item.text()
print(text, href)
"""
快速总结:
1. pyquery(选择器)
2. items() 当选择器选择的内容很多，需要一个一个处理时
3. attr(属性名) 获取属性信息
4. text() 获取文本
"""
3.4.2. 进阶使用
from pyquery import PyQuery
html = """
<HTML>
<div class="aaa">哒哒哒</div>
<div class="bbb">嘟嘟嘟</div>
</HTML>
"""
# 加载html内容
p = PyQuery(html)
# 在xxxx标签后面添加xxxxx新标签
p("div.aaa").after("""<div class="ccc">吼吼吼</div>""")
p("div.aaa").append("""<span>嘿嘿嘿</span>""")
p("div.bbb").attr("class", "aaa") # 修改/新增属性
p("div.ccc").attr("id", "12306")
p("div.ccc").remove_attr("id") # 删除属性
p("div.ccc").remove() # 删除标签
3.4.3. 实操案例
import csv
import requests
import urllib3
from pyquery import PyQuery
# 抑制SSL警告（可选，避免控制台输出无关警告）
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
CSV_FILE_PATH = "抖音热搜.csv"
TARGET_URL = "https://www.zhuanti.com.cn/top/z46"
HEADERS = {
"Referer": "https://m.zhuanti.com.cn/",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0"
}
def get_page_source(url: str) -> str | None:
resp = requests.get(url, headers=HEADERS, timeout=10, verify=False)
resp.raise_for_status()
resp.encoding = "utf-8"
return resp.text
def parse_page_source(html: str) -> None:
with open(CSV_FILE_PATH, mode="a+", encoding='utf-8-sig', newline='') as f:
# 创建csv写入对象
csv_writer = csv.writer(f)
# 移动文件指针到开头，判断是否为空文件（为空则写入表头）
f.seek(0)
if not f.readline():
csv_writer.writerow(["排名", "标题", "链接", "热度"])
doc = PyQuery(html)
item_list = doc(".item").items()
for item in item_list:
rank = item(".rank").text()
title = item(".titles").text()
href = item(".titles a").attr("href")
reading = item(".reading").text()
csv_writer.writerow([rank, title, href, reading])
def main():
try:
html = get_page_source(TARGET_URL)
parse_page_source(html)
except Exception as e:
print(f"热搜数据爬取失败，错误信息：{str(e)}")
if __name__ == "__main__":
main()
4. 模拟浏览器登录
import requests
# 会话
session = requests.session()
headers = {
"Content-Type": "application/json",
"Referer": "http://47.113.113.212/",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0"
}
data = {
"username": "albert",
"password": "123456"
}
# 登录
url = "http://47.113.113.212:8080/auth/login"
login_resp = session.post(url, json=data, timeout=10, headers=headers)
login_resp.raise_for_status()
login_resp.encoding = "utf-8"
token = login_resp.headers["Authorization"]
# 请求
headers = {
"Authorization": token
}
resp = session.get("http://47.113.113.212:8080/auth/me", timeout=10, headers=headers)
resp.raise_for_status()
print(resp.json().get("data"))
5. 防盗链
import requests
import urllib3
url = "https://www.pearvideo.com/video_1804596"
contId = url.split("_")[1]
videoStatusUrl = f"https://www.pearvideo.com/videoStatus.jsp?contId={contId}&mrd=0.5587364561383386"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0",
# 防盗链：溯源，当前请求的上一级
"Referer": "https://www.pearvideo.com/video_1804596",
}
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
resp = requests.get(videoStatusUrl, headers=headers, timeout=10, verify=False)
dic = resp.json()
srcUrl = dic["videoInfo"]["videos"]["srcUrl"]
systemTime = dic["systemTime"]
srcUrl = srcUrl.replace(systemTime, f"cont-{contId}")
with open("a.mp4", mode="wb") as f:
f.write(requests.get(srcUrl, timeout=10, verify=False).content)
6. 代理
import requests
url = "https://www.baidu.com/"
# 准备代理信息
proxy={
"http": "http://218.207.102.106:843",
"https": "https://218.207.102.106:843"
}
# proxies代理
resp = requests.get(url, proxies=proxy)
resp.encoding = "utf-8"
print(resp.text)
7. 多线程
# 线程，进程
# 进程是资源单位，每一个进程至少要有一个线程
# 线程是执行单位
# 单线程
# def func():
# for i in range(1000):
# print("func", i)
#
#
# if __name__ == '__main__':
# func()
# for i in range(1000):
# print("main", i)
# 多线程(写法一)
from threading import Thread
def func(name):
for i in range(1000):
print(name, i)
if __name__ == '__main__':
t1 = Thread(target=func, args=("aaa",)) # 创建线程并给线程安排任务
t1.start() # 多线程状态为可以开始工作状态，具体的执行时间由CPU决定
t2 = Thread(target=func, args=("bbb",))
t2.start()
for i in range(1000):
print("main", i)
# 多线程(写法二)
# class MyThread(Thread):
# def run(self): # 固定的 -> 当线程被执行时，被执行的就是run()
# for i in range(1000):
# print("子线程", i)
#
#
# if __name__ == '__main__':
# t1 = MyThread()
# t1.start() # 开启线程
# t2 = MyThread()
# t2.start() # 开启线程
#
# for i in range(1000):
# print("主线程", i)
8. 多进程
from multiprocessing import Process
def func():
for i in range(1000):
print("子进程", i)
if __name__ == '__main__':
p = Process(target=func)
p.start()
for i in range(1000):
print("主进程", i)
9. 线程池 | 进程池
# 线程池：一次性开辟一些线程，用户直接给线程池提交任务，线程任务的调度交给线程池来完成
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
# 任务函数
def fn(name):
# 循环10次
for i in range(10):
print(name, i)
if __name__ == '__main__':
# 创建线程池
with ThreadPoolExecutor(max_workers=50) as t:
for i in range(100):
# submit方法：提交任务到线程池，非阻塞（主程序继续执行下一次循环）
t.submit(fn, name=f"线程{i}")
# 等待线程池中的任务全部执行完毕，才继续执行(守护)
print("123")
# 创建进程池
with ProcessPoolExecutor(max_workers=50) as t:
for i in range(100):
t.submit(fn, name=f"进程{i}")
# 等待进程池中的任务全部执行完毕，才继续执行(守护)
print("123")
9.1. 实操案例
import csv
from concurrent.futures import ThreadPoolExecutor
import requests
import urllib3
from pyquery import PyQuery
# 抑制SSL警告（可选，避免控制台输出无关警告）
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
CSV_FILE_PATH = "豆瓣影评.csv"
HEADERS = {
"Referer": "https://movie.douban.com/chart",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0"
}
def get_page_source(url: str) -> str | None:
resp = requests.get(url, headers=HEADERS, timeout=10, verify=False)
resp.raise_for_status()
resp.encoding = "utf-8"
return resp.text
def parse_page_source(html: str) -> None:
with open(CSV_FILE_PATH, mode="a+", encoding='utf-8-sig', newline='') as f:
# 创建csv写入对象
csv_writer = csv.writer(f)
# 移动文件指针到开头，判断是否为空文件（为空则写入表头）
f.seek(0)
if not f.readline():
csv_writer.writerow(["影片名称", "用户", "标题", "链接", "内容", "点赞", "评论", "发布时间"])
doc = PyQuery(html)
item_list = doc(".main.review-item").items()
for item in item_list:
movie_title = item(".subject-img img").attr("title")
name = item(".name").text()
title = item(".main-bd h2 a").text()
href = item(".main-bd h2 a").attr("href")
content = item(".short-content").text().strip()
up = item(".action a.up").text()
comment = item(".action a.reply").text().replace('回应', '')
time_text = item(".main-meta").text()
csv_writer.writerow([movie_title, name, title, href, content, up, comment, time_text])
def download_one_page(url: str) -> None:
html = get_page_source(url)
parse_page_source(html)
print(url, "提取完毕")
if __name__ == "__main__":
try:
with ThreadPoolExecutor(max_workers=50) as t:
for i in range(0, 80, 20):
t.submit(download_one_page, f"https://movie.douban.com/review/best/?start={i}")
except Exception as e:
print(f"影评数据爬取失败，错误信息：{str(e)}")
10. 协程
import asyncio
import time
# async def func():
# print("你好，我是薇尔莉特")
#
#
# if __name__ == '__main__':
# g = func() # 此时的函数是异步协程函数，函数执行得到的是一个协程对象
# asyncio.run(g) # 协程程序运行需要asyncio模块的支持
async def func1():
print("我是甲")
await asyncio.sleep(3)
print("我是甲")
async def func2():
print("我是乙")
await asyncio.sleep(2)
print("我是乙")
async def func3():
print("我是丙")
await asyncio.sleep(4)
print("我是丙")
# 定义主异步函数，所有异步逻辑都放在这里
async def main():
await asyncio.gather(func1(), func2(), func3())
if __name__ == '__main__':
t1 = time.time()
# 一次性启动多个任务(协程)
asyncio.run(main())
t2 = time.time()
print(t2 - t1)
10.1. aiohttp 模块
import asyncio
import aiohttp
urls = [
"https://pixnio.com/free-images/2026/01/12/2026-01-12-04-50-14-576x384.jpg",
"https://pixnio.com/free-images/2026/01/01/2026-01-01-11-01-36-576x384.jpeg",
"https://pixnio.com/free-images/2025/08/18/2025-08-18-10-31-57-576x324.jpg"
]
async def aiodownload(url: str):
name = url.rsplit("/", 1)[1] # 从右边切，切一次，得到[1]位置的内容
try:
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0"
}
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers) as response:
response.raise_for_status()
with open(name, mode="wb") as f:
f.write(await response.content.read())
print(f"{name} 下载完成")
except Exception as e:
print(f"{name} 下载失败：{str(e)}")
async def main():
tasks = []
for url in urls:
tasks.append(aiodownload(url))
await asyncio.gather(*tasks) # *解包
if __name__ == '__main__':
asyncio.run(main())
10.2. 实操案例
# https://dushu.baidu.com/api/pc/getCatalog?data={"book_id":"4306063500"}
# https://dushu.baidu.com/api/pc/getChapterContent?data={"book_id":"4306063500","cid":"4306063500|1569782244","need_bookinfo":1}
"""
1. 同步操作：访问getCatalog 拿到所有章节的cid和名称
2. 异步操作：访问getChapterContent 下载所有的文章内容
"""
import asyncio
import json
import aiofiles
import aiohttp
HEADERS = {
"Referer": "https://dushu.baidu.com/",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0"
}
BOOK_ID = "4306063500"
async def getChapterContent(session, cid, b_id, title):
try:
data = json.dumps({
"book_id": b_id,
"cid": f"{b_id}|{cid}",
"need_bookinfo": 1
})
params = {"data": data}
# 复用ClientSession，减少连接开销
async with (session.get(
url="https://dushu.baidu.com/api/pc/getChapterContent",
params=params,
headers=HEADERS,
timeout=aiohttp.ClientTimeout(total=15))
as response):
dic = await response.json()
async with aiofiles.open(title, "w", encoding="utf-8") as f:
await f.write(dic["data"]["novel"]["content"])
print(f"成功下载：{title}")
except Exception as e:
print(f"下载失败 {title}：{str(e)}")
async def getCatalog():
url = f"https://dushu.baidu.com/api/pc/getCatalog?data={json.dumps({'book_id': BOOK_ID})}"
async with aiohttp.ClientSession() as session: # 创建全局session，复用连接
try:
async with session.get(url, headers=HEADERS, timeout=10) as response:
response.raise_for_status()
dic = await response.json()
tasks = []
items = dic["data"]["novel"]["items"]
for item in items:
title = item["title"]
cid = item["cid"]
# 准备异步任务
tasks.append(getChapterContent(session, cid, BOOK_ID, title))
await asyncio.gather(*tasks)
except Exception as e:
print(f"获取目录失败：{str(e)}")
if __name__ == '__main__':
asyncio.run(getCatalog())
print("\n所有章节下载任务执行完毕！")
11. m3u8 视频爬取合并
# <video src="不能播的视频.mp4"></video>
# 一般的视频网站是怎么做的？
# 用户上传 -> 转码(把视频做处理，2K，1080，标清) -> 切片处理(把单个的文件进行拆分) 60
# 用户在进行拉进进度条的时候
# ==============================================
# 需要一个文件记录：1.视频播放顺序，2.视频存放的路径.
# M3U8 txt json => 文本
import asyncio
import os
import re
import aiohttp
import requests
"""
想要抓取一个视频：
1. 找到m3u8（各种手段）
2. 通过m3u8下载到ts文件
3. 可以通过各种手段(不仅是编程手段) 把ts文件合并为一个mp4文件
"""
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0"
}
# 创建video文件夹（避免下载时找不到路径）
if not os.path.exists("video"):
os.makedirs("video")
def extract_m3u8_url_from_html(html):
pattern = re.compile(r'https:\\/\\/vip1\.lz-cdn5\.com\\/.*?\.m3u8', re.S)
match = pattern.search(html)
if match:
return match.group(0).replace('\\/', '/')
else:
print("未找到m3u8的url字段")
return None
def extract_m3u8_url_from_file(m3u8_url01):
with open("紫罗兰剧场版.m3u8", mode="r", encoding="utf-8") as f:
suffix = ""
lines = f.readlines()
for line in lines:
line = line.strip()
if line and line.endswith('.m3u8'):
suffix = line
break
if suffix:
return m3u8_url01.rsplit("/", 1)[0] + '/' + suffix
else:
return None
# 异步下载单个ts文件
async def download_single_ts(session, ts_url, save_path):
try:
async with session.get(ts_url, headers=headers) as resp:
if resp.status == 200:
with open(save_path, "wb") as f:
f.write(await resp.read())
print(f"{ts_url} 下载完成")
else:
print(f"{ts_url} 下载失败，状态码：{resp.status}")
except Exception as e:
print(f"{ts_url} 下载异常：{e}")
# 异步批量下载ts文件
async def download_ts_async(m3u8_url02):
# 读取m3u8文件，提取所有ts链接
ts_tasks = []
with open("紫罗兰剧场版.m3u8", mode="r", encoding="utf-8") as f:
lines = f.readlines()
for line in lines:
line = line.strip()
if line.startswith('#'):
continue
# 拼接完整ts下载地址
ts_url = m3u8_url02.rsplit("/", 1)[0] + '/' + line
save_path = f"video/{line}"
# 创建异步任务
ts_tasks.append((ts_url, save_path))
# 创建异步会话，并发下载
async with aiohttp.ClientSession() as session:
tasks = []
# 遍历ts_tasks，逐个创建任务并添加到列表
for url, path in ts_tasks:
task = asyncio.create_task(download_single_ts(session, url, path))
tasks.append(task)
# 等待所有任务完成
await asyncio.gather(*tasks)
def merge_ts():
lst = []
# 读取m3u8文件，提取ts文件名
with open("紫罗兰剧场版.m3u8", mode="r", encoding="utf-8") as f:
for line in f:
if line.startswith("#"): # 跳过注释行
continue
line = line.strip() # 去除换行/空格
# 坑，cmd路径要用反斜杠'\'
lst.append(f'"video\\{line}"') # 拼接ts文件路径
# 把列表用"+"连接（Windows copy /b需要用+拼接文件）
s = "+".join(lst)
# 执行Windows合并命令（/b表示二进制模式）
os.system(f'copy /b {s} "movie.mp4"')
print("搞定!")
def main():
url = "https://ifana.cc/play/6073-1-1.html"
# 从源码中提取m3u8_url
resp01 = requests.get(url, headers=headers)
resp01.encoding = "utf-8"
m3u8_url01 = extract_m3u8_url_from_html(resp01.text)
print(m3u8_url01)
resp01.close()
if not m3u8_url01:
return
# 提取真实m3u8_url后缀
resp02 = requests.get(m3u8_url01, headers=headers)
resp02.encoding = "utf-8"
with open("紫罗兰剧场版.m3u8", "wb") as f:
f.write(resp02.content)
resp02.close()
m3u8_url02 = extract_m3u8_url_from_file(m3u8_url01)
print(m3u8_url02)
if not m3u8_url02:
return
# 下载m3u8
resp03 = requests.get(m3u8_url02, headers=headers)
resp03.encoding = "utf-8"
with open("紫罗兰剧场版.m3u8", "wb") as f:
f.write(resp03.content)
resp03.close()
# 异步下载ts文件
asyncio.run(download_ts_async(m3u8_url02))
# 合并视频
merge_ts()
if __name__ == '__main__':
main()
12. Selenium
12.1. 基本操作
import time
from selenium.webdriver import Edge, Keys, ActionChains
from selenium.webdriver.common.by import By
from selenium.webdriver.edge.options import Options
# 创建Edge选项对象
edge_options = Options()
# 禁用自动关闭
edge_options.add_experimental_option("detach", True)
web = Edge(options=edge_options)
web.get("https://movie.douban.com/")
# 模拟遮挡广告去除
web.execute_script("""
const movieAnnualLink = document.querySelector('.movieannual');
if (movieAnnualLink) {
movieAnnualLink.remove();
}
""")
# 找到输入框，输入电影名称，搜索
web.find_element(By.XPATH, '//*[@id="inp-query"]').send_keys("紫罗兰永恒花园", Keys.ENTER)
# 等待搜索结果
time.sleep(1)
# 数据提取
items = web.find_elements(By.XPATH, '//div[contains(@class, "item-root")]')
for item in items:
title = item.find_element(By.XPATH, './div/div[1]/a')
# 在新标签页中打开
# 1.按住Ctrl键点击链接，在新标签页打开
ActionChains(web).key_down(Keys.CONTROL).click(title).key_up(Keys.CONTROL).perform()
# 2.通过JS设置链接的target属性为_blank
# web.execute_script("arguments[0].setAttribute('target', '_blank');", title)
# 此时，在浏览器这边. 我们看到的内容已经是详情页的内容了：
# 但是，在selenium的眼中. 我们依然在首页.
# 所以，必须得让selenium去调整它的视角
# 切换到最后一个窗口
web.switch_to.window(web.window_handles[-1])
h2 = web.find_element(By.XPATH, '//div[@class="related-info"]/h2/i')
print(h2.text)
time.sleep(0.25)
summary = web.find_element(By.XPATH, '//div[@id="link-report-intra"]/span')
print(summary.text)
time.sleep(0.25)
print('\n')
web.close()
web.switch_to.window(web.window_handles[0])
time.sleep(0.25)
web.quit() # 关闭浏览器
12.2. 切换 iframe
from selenium.webdriver import Edge
from selenium.webdriver.common.by import By
from selenium.webdriver.edge.options import Options
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.wait import WebDriverWait
edge_options = Options()
# 禁用自动关闭
edge_options.add_experimental_option("detach", True)
web = Edge(options=edge_options)
web.get("https://www.126.com/")
# 切换到iframe
iframe = WebDriverWait(web, 10).until(
EC.presence_of_element_located((By.XPATH, '//iframe[starts-with(@id, "x-URS-iframe")]'))
)
web.switch_to.frame(iframe)
# 等待iframe可用并直接切换（无需先定位再切换）
# WebDriverWait(web, 10).until(
# EC.frame_to_be_available_and_switch_to_it((By.XPATH, '//iframe[starts-with(@id, "x-URS-iframe")]'))
# )
account = web.find_element(By.NAME, 'email')
ac_placeholder = account.get_property('placeholder')
print(ac_placeholder)
pwd = web.find_element(By.NAME, 'password')
pwd_placeholder = pwd.get_property('placeholder')
print(pwd_placeholder)
# 跳出iframe
web.switch_to.parent_frame()
title = web.find_element(By.XPATH, '//*[@class="headerTitle"]')
print(title.text)
12.3. 切换下拉列表
from selenium.webdriver import Edge
from selenium.webdriver.common.by import By
from selenium.webdriver.edge.options import Options
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.select import Select
from selenium.webdriver.support.wait import WebDriverWait
# 配置无头信息
edge_options = Options()
edge_options.add_argument("--headless")
edge_options.add_argument("--disable-gpu")
web = Edge(options=edge_options)
web.get("https://www.endata.com.cn/BoxOffice/B0/Year/index.html")
sel = web.find_element(By.XPATH, '//*[@id="OptionDate"]')
sel_new = Select(sel)
print(len(sel_new.options)) # 所有的选项 0 1 2 3 4 5 6
for i in range(len(sel_new.options)):
sel_new.select_by_index(i) # 根据位置切换
WebDriverWait(web, 10).until(
EC.presence_of_element_located((By.XPATH, '//*[@id="TableList"]/table/tbody/tr'))
)
trs = web.find_elements(By.XPATH, '//*[@id="TableList"]/table/tbody/tr')
for tr in trs:
print(tr.text)
13. Scrapy
13.1. 工作原理
常规爬虫逻辑：
scrapy 框架工作逻辑：
整个工作流程：
13.2. 基本使用
控制台执行scrapy startproject 项目名称
这里的示例项目名称使用 tianya，并设置为 Resource Root
进入到 tianya目录，控制台执行 scrapy genspider ty "tianya.org"
流程总结：
要使用到 redis，所以需安装依赖：redis、scrapy-redis
settings.py⬇️
BOT_NAME = "tianya"
SPIDER_MODULES = ["tianya.spiders"]
NEWSPIDER_MODULE = "tianya.spiders"
LOG_LEVEL = "WARNING"
ADDONS = {}
ROBOTSTXT_OBEY = False
CONCURRENT_REQUESTS_PER_DOMAIN = 1
DOWNLOAD_DELAY = 1
ITEM_PIPELINES = {
"tianya.pipelines.TianyaPipeline": 300,
}
FEED_EXPORT_ENCODING = "utf-8"
items.py⬇️
import scrapy
class TianyaItem(scrapy.Item):
title = scrapy.Field() # 标题
author = scrapy.Field() # 作者
reply = scrapy.Field() # 回复数
like = scrapy.Field() # 点赞数
mark = scrapy.Field() # 收藏数
ty.py⬇️
import scrapy
from redis import Redis
from tianya.items import TianyaItem
# 有可能产生重复的数据
# 1. 使用python的set集合去重
# 2. 使用redis的set集合去重(推荐)
# 用redis有两种方案去重：
# 1. url，优点：简单，缺点：如果url内部进行了更新，可能会忽略掉一些数据
# 2. 数据，优点：准确性高，缺点：若数据集庞大，对redis来说会产生BigKey
class TySpider(scrapy.Spider):
name = "ty"
allowed_domains = ["tianya.org"]
start_urls = ["https://www.tianya.org/"]
def __init__(self, name=None, **kwargs):
self.red = Redis(
host='127.0.0.1',
port=6379,
db=0,
decode_responses=False,
encoding="utf-8",
)
# 让父类能初始化
super(TySpider, self).__init__(name, **kwargs)
def parse(self, resp, **kwargs):
# 进入详情页
li_list = resp.xpath("//ul[@class='list-unstyled threadlist mb-0']/li")
for li in li_list:
href = li.xpath("./@data-href").extract_first()
detail_url = resp.urljoin(href)
result = self.red.sismember("tianya:ty:detail:url", detail_url)
if result:
print(f"该url已经被抓取过{detail_url}")
else:
yield scrapy.Request(
url=detail_url,
callback=self.parse_detail,
**kwargs
)
def parse_detail(self, resp):
t = TianyaItem()
title = resp.xpath("//div[@class='card card-thread']/div/div/div/h4/text()").extract_first()
author = resp.xpath("//span[@class='username']/a/text()").extract_first()
func_num_list = resp.xpath("//div[contains(@class, 'thread_left_func_num')]/text()").extract()
t['title'] = title.strip() if title else None
t['author'] = author.strip() if author else None
t['reply'] = func_num_list[0].strip() if len(func_num_list) >= 1 else None
t['like'] = func_num_list[1].strip() if len(func_num_list) >= 2 else None
t['mark'] = func_num_list[2].strip() if len(func_num_list) >= 3 else None
self.red.sadd("tianya:ty:detail:url", resp.url)
yield t
pipelines.py⬇️
class TianyaPipeline:
def process_item(self, item):
# 判断，是否在redis中存储了，若已存储就不进入数据库
print(item)
return item
执行 scrapy crawl ty⬇️
(.venv) PS G:\PyCharm-2024-workplace\PythonLearn\第六章_Scrapy框架\01_基础入门\tianya> scrapy crawl ty
{'author': '木子居士',
'like': '0',
'mark': '0',
'reply': '3',
'title': '兄弟们，今年计划亏多少'}
{'author': '马克思',
'like': '0',
'mark': '0',
'reply': '3',
'title': '天涯社区网速测试如何？'}
{'author': None,
'like': '0',
'mark': '0',
'reply': '0',
'title': '出售维修二手租赁回收安捷伦 T1141A RFID HF测试仪'}
{'author': '马克思', 'like': '0', 'mark': '0', 'reply': '2', 'title': '新人求助'}
{'author': 'Snow',
'like': '23',
'mark': '14',
'reply': '378',
'title': '用我30年的经历，为你揭开周易的神秘-四书五经丶(6)'}
{'author': '点逐清风',
'like': '111',
'mark': '40',
'reply': '1387',
'title': '(长篇)女性秘史◆那些风华绝代、风情万种的女人，为你打开女人的所有秘密'}
{'author': '小笨笨',
'like': '28',
'mark': '15',
'reply': '447',
'title': '一个潜水多年的体制内的生意人来实际谈谈老百姓该怎么办？'}
{'author': '龙井茶',
'like': '26',
'mark': '17',
'reply': '475',
'title': '奶奶遗留一本怪书，里面都是一些吓人古术，给我带来了可怕的经历'}
{'author': '兰风筝',
'like': '7',
'mark': '0',
'reply': '283',
'title': '有这样的女友，你是什么感受'}
{'author': '马克思',
'like': '7',
'mark': '0',
'reply': '198',
'title': '这样的能结婚吗??'}
{'author': 'ighaonr',
'like': '1',
'mark': '0',
'reply': '99',
'title': '迷糊不迷糊你看了在说'}
{'author': '鑫伍',
'like': '16',
'mark': '7',
'reply': '238',
'title': '【经济专栏】赚未来十年的钱【已出版】'}
{'author': 'luyi',
'like': '0',
'mark': '0',
'reply': '37',
'title': '想她了可经打给她，建行卡号没变哦！！！'}
{'author': '马克思',
'like': '13',
'mark': '12',
'reply': '365',
'title': '【国庆礼包】天涯神帖1-211部大放送(2025/10/4修复)'}
{'author': '啦啦啦',
'like': '230',
'mark': '111',
'reply': '2822',
'title': '（全网最全最详细）天涯神贴合集1000篇 | 超全合集，无需解压'}
{'author': 'chenpingan',
'like': '2',
'mark': '0',
'reply': '187',
'title': '下班了，来波福利，'}
{'author': 'oylhoylh',
'like': '91',
'mark': '28',
'reply': '896',
'title': '【寒门再难出贵子】神帖限时开放'}
{'author': 'yugaochao',
'like': '0',
'mark': '0',
'reply': '59',
'title': '异域风情！！不一样的美！！'}
{'author': '247574',
'like': '4',
'mark': '0',
'reply': '235',
'title': '短裙过膝袜，绝对YYDS'}
{'author': None, 'like': '0', 'mark': '0', 'reply': '1', 'title': '为什么需要礼拜五'}
redis 数据库⬇️
13.3. 分布式
目录结构⬇️
items.py 和 pipelines.py 保持不变
ty.py 改动如下⬇️
import scrapy
from scrapy_redis.spiders import RedisSpider
from tianya2.items import TianyaItem
class TySpider(RedisSpider):
name = "ty"
allowed_domains = ["tianya.org"]
# start_urls = ["https://www.tianya.org/"]
redis_key = "ty_start_url"
def parse(self, resp, **kwargs):
# 自定义请求头
custom_headers = {
'Cookie': 'bbs_token=dnsSNob06Ok7KtBitQaxQjh5DGbAOfa4lAIqzSqHVljjwm9v; Hm_lvt_bdc65c3d1a40d80d03c75152017b11eb=1768646265; bbs_sid=7u2q69307h7c0na0lv64rur2cc'
}
# 进入详情页
li_list = resp.xpath("//ul[@class='list-unstyled threadlist mb-0']/li")
for li in li_list:
href = li.xpath("./@data-href").extract_first()
# 交给 Scrapy-Redis 做去重和调度
yield scrapy.Request(
url=resp.urljoin(href),
callback=self.parse_detail,
headers=custom_headers,
**kwargs
)
def parse_detail(self, resp):
t = TianyaItem()
title = resp.xpath("//div[@class='card card-thread']/div/div/div/h4/text()").extract_first()
author = resp.xpath("//span[@class='username']/a/text()").extract_first()
func_num_list = resp.xpath("//div[contains(@class, 'thread_left_func_num')]/text()").extract()
t['title'] = title.strip() if title else None
t['author'] = author.strip() if author else None
t['reply'] = func_num_list[0].strip() if len(func_num_list) >= 1 else None
t['like'] = func_num_list[1].strip() if len(func_num_list) >= 2 else None
t['mark'] = func_num_list[2].strip() if len(func_num_list) >= 3 else None
yield t
使用 redis 的布隆过滤器去重，所以需安装依赖：scrapy-redis-bloomfilter
settings.py⬇️
BOT_NAME = "tianya2"
SPIDER_MODULES = ["tianya2.spiders"]
NEWSPIDER_MODULE = "tianya2.spiders"
# debug < info < warning < error < critical
LOG_LEVEL = "INFO"
ADDONS = {}
ROBOTSTXT_OBEY = False
CONCURRENT_REQUESTS_PER_DOMAIN = 1
DOWNLOAD_DELAY = 1
# 配置全局请求头
DEFAULT_REQUEST_HEADERS = {
'UserAgent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0'
}
# ========== redis 相关配置 ==========
REDIS_PARAMS = {
'host': '127.0.0.1',
'port': 6379,
'db': 0,
}
# ========== scrapy_redis 相关配置 ==========
# 分布式调度器
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
# 分布式去重 (这里使用布隆过滤器)
DUPEFILTER_CLASS = "scrapy_redis_bloomfilter.dupefilter.RFPDupeFilter"
# 哈希函数个数，默认为 6
BLOOMFILTER_HASH_NUMBER = 6
# BloomFilter 的 bit 参数，默认为 30，占用 128MB 空间，去重量级为 1 亿
BLOOMFILTER_BIT = 30
# 断点续爬 (可选)
SCHEDULER_PERSIST = True
ITEM_PIPELINES = {
"tianya2.pipelines.TianyaPipeline": 300,
"scrapy_redis.pipelines.RedisPipeline": 301,
}
FEED_EXPORT_ENCODING = "utf-8"
在控制台 tianya2 文件夹目录下执行命令 scrapy crawl ty
进入 redis-cli 执行 lpush ty_start_url https://www.tianya.org/命令
控制台输出如下⬇️
redis 数据库⬇️
使用 MySQL 存储数据⬇️
创建 spider_test 数据库：
创建 new 数据表：
-- 天涯论坛爬虫数据存储表：存储抓取的帖子基础信息
CREATE TABLE `news` (
`id` bigint(11) NOT NULL AUTO_INCREMENT COMMENT '主键ID，自增唯一标识',
`title` varchar(255) NOT NULL COMMENT '帖子标题',
`author` varchar(255) NOT NULL COMMENT '帖子发布作者',
`reply` int(11) NOT NULL COMMENT '帖子回复数',
`like` int(11) NOT NULL COMMENT '帖子点赞数（`like`为MySQL关键字，需反引号包裹）',
`mark` int(11) NOT NULL COMMENT '帖子收藏数',
`url` varchar(255) NOT NULL COMMENT '帖子唯一链接（作为唯一键，避免重复抓取）',
PRIMARY KEY (`id`) COMMENT '主键约束：基于自增ID',
UNIQUE KEY `uk_url` (`url`) COMMENT '唯一键约束：保证帖子URL不重复，避免重复插入数据'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='天涯论坛帖子爬虫数据表';
安装依赖：pymysql
settings.py 添加配置：
ITEM_PIPELINES = {
"tianya2.pipelines.TianyaPipeline": 300,
# "scrapy_redis.pipelines.RedisPipeline": 301,
"tianya2.pipelines.MysqlPipeline": 302,
}
MYSQL_CONFIG = {
'host': '127.0.0.1',
'port': 3306,
'user': 'root',
'password': '你的MySQL密码',
'db': 'spider_test',
'charset': 'utf8mb4'
}
pipelines.py 代码：
import pymysql
from twisted.enterprise import adbapi
class TianyaPipeline:
def process_item(self, item, spider):
spider.logger.info(f"成功抓取到了 {item}")
return item
class MysqlPipeline:
def __init__(self, db_pool):
self.db_pool = db_pool
@classmethod
def from_crawler(cls, crawler):
db_params = crawler.settings.get('MYSQL_CONFIG')
db_pool = adbapi.ConnectionPool(
'pymysql',
**db_params,
autocommit=True,
cursorclass=pymysql.cursors.DictCursor
)
return cls(db_pool)
def process_item(self, item, spider):
# 异步插入数据
query = self.db_pool.runInteraction(self.insert_item, item)
query.addErrback(self.handle_error, item, spider)
return item
def insert_item(self, cursor, item):
insert_sql = """
INSERT INTO news(title, author, reply, `like`, mark, url)
VALUES (%s, %s, %s, %s, %s, %s)
"""
cursor.execute(insert_sql, (
item.get('title') or ' ',
item.get('author') or ' ',
item.get('reply', 0),
item.get('like', 0),
item.get('mark', 0),
item.get('url') or ' ',
))
def handle_error(self, failure, item, spider):
spider.logger.error(f"MySQL插入失败：{failure}，Item：{item}")
def close_spider(self, spider):
self.db_pool.close()
按照之前的步骤启动爬虫程序，可观察到数据已成功入库：
北京人形旗下天工造物具身智能开源社区，聚焦具身天工与慧思开物两大平台
更多推荐
29
0- 0
已为社区贡献1条内容
所有评论(0)