Python Scarpy 整合 FastAPI

前言

刑啊，我们开始Python爬虫网页分析啦。由于某虫的叫法较为敏感（求生欲拉满），以下均将其称为网页分析吧！

说在前面
网页分析技术，是我们日常学习与工作中常见的需求之一，因此，多少了解甚至会一点网页分析技术，是当今程序猿的基本要求。基本功了属于是。本文基于我实际的工作记录，分享一下Scrapy整合FastAPI个人解决方案

注意，使用网页分析技术，请严格遵守相关法律法规，本文仅供技术学习与交流，代码例子均为虚拟的脱敏示例。

学习目标

• 会用 Python 进行简单的网页分析实战，将分析到的数据存储到mysql数据表

需求描述
基于某合法业务，分析某网页，并将数据插入到表里。

既然我们要网页分析，首先要选择框架；其次是由于需要将数据保存到表里，为了避免异构造成耦合的麻烦，需要在python代码里直接连接数据库并将数据插入到表里；最后，由于最终是以api的形式给其它服务调用，则需要选择web框架，将分析逻辑做成接口供其它服务调用。

准备工作
基于上面需求描述，本次我们选用

• fastapi[all] ：无脑选择 fastapi 所有依赖，构建fastapi项目，供封装接口使用
• mysqlclient 与 SQLAlchemy：数据库相关依赖
• 网页分析依赖：Scrapy

示例：

pip install fastapi[all]
pip install mysqlclient==2.1.1
pip install SQLAlchemy==2.0.23
pip install scrapy

对应 requirements.txt：

fastapi[all]
mysqlclient==2.1.1
SQLAlchemy==2.0.23
scrapy~=2.11.1

1. 网页分析入门

1.1 基本原理

网页分析的原理是模拟浏览器对网页进行访问，并获取访问的信息。这就意味着一般情况下，网页分析只能获取我们平常能在浏览器获取的信息。与浏览网页 + f12查看部分信息并无多大差别。

一般网页分析原理
以 Java 的 Jsoup为例，一般传统的网页分析流程是：发送Http请求获取网页信息 -> 接收网页信息，分析里面的字段 -> 再根据需求与获取的信息，若需要二次请求则进行二次请求，再重复上述步骤，不需要则存储数据。

与攻击的区别
与网络攻击不同的是，攻击是试图破坏系统或获取后台权限以达到不法目的。而网页分析则是在遵守相应法律法规的前提下，合法获取目标服务的信息，以支持实际的运营。

注意
使用网络分析技术，务必要遵循对应的法律法规！

1.2 Scrapy 原理

Scrapy 是一个支持分布式的Python爬虫框架。它在基于传统的网页分析技术原理上进行了优化，如下图所示：

图片来源于网络，侵删

Scrapy的核心是Scrapy引擎，通过调度器的队列，调度下载器及爬虫文件的中间结果给引擎，再通过引擎将分析结果给管道做结果的处理。

Scrapy的基本原理如上，而实际操作中，若我们不需要对底层进行细致的研究，整体爬取的流程及代码相对简单。下面我们开始学习如何使用Scrapy并整合FastAPI，让我们的网页分析程序可通过API访问。

2. 创建项目

此小节介绍创建 FastAPI + Scrapy 项目的流程。

2.1 创建Scrapy项目

2.2.1 创建Scrapy项目

进入我们目标项目目录，执行

scrapy startproject <prject_name>

执行上述命令后，生成基础项目，里面包含了Scrapy项目的基本文件。

接下来，我们用pycharm等工具打开，然后需要注意一下是否设置了 python解释器。若没有设置，需要参考如下截图设置并生成.venv文件：

然后，在虚拟环境下再执行一遍：

pip install scrapy

这是为了在虚拟环境中也有scrapy相关依赖

2.2.2 创建Spider

terminal 执行 命令，参考如下：

scrapy genspider itcast "itcast.cn"

稍微解释一下，该命令的结构为:

scrapy genspider <spider_name> <domain>

domain为网站域名，意思为要分析的网站范围。后续的url都基于这个域名。要在实际业务使用的同学请将spider_name 和 domain 改成实际的。

那么此处我们就参照参考教程，爬取某客首页吧！

2.2.3 执行Demo

在上一小节执行了genspider命令后，项目里又增加了一些py文件。有点像脚手架。

我们稍作修改，将response保存为文件：

 def parse(self, response):filename = "test.html"open(filename, 'w').write(response.body)pass

执行一下:

scrapy crawl itcast

若成功在项目内生成test.html文件，至此scrapy项目创建与spider创建成功！

2.2 引入FastAPI

引入FastAPI很简单，只需要新建一个main.py，然后参考如下代码：

import uvicornfrom fastapi import FastAPI, applications
from fastapi.openapi.docs import get_swagger_ui_html# 解决国内Swagger无法正常显示问题
def swagger_monkey_patch(*args, **kwargs):"""fastapi的swagger ui默认使用国外cdn, 所以导致文档打不开, 需要对相应方法做替换在应用生效前, 对swagger ui html做替换:param args::param kwargs::return:"""return get_swagger_ui_html(*args, **kwargs,swagger_js_url='https://cdn.staticfile.org/swagger-ui/4.15.5/swagger-ui-bundle.min.js',  # 改用国内cdnswagger_css_url='https://cdn.staticfile.org/swagger-ui/4.15.5/swagger-ui.min.css')applications.get_swagger_ui_html = swagger_monkey_patchapp = FastAPI()
# 指定Swagger 版本为3.0.0
app.openapi_version = "3.0.0"@app.get("/")
async def root():return {"message": "Hello, this is an fastapi project to analyse the website!"}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=8080)

接下来，就可以根据上述代码示例的main方法启动项目，并在swagger查看api接口。FastAPI 默认是swagger路径在 ip:端口号/docs

2. 获取Cookie

网页分析第一步，就是要确保我们所访问的各页面携带登录认证信息。那就是Cookie。

本次我采取的模拟登录策略是采用urllib的方式post请求并获取Cookie，然后将这个模拟登录的方法单独写成一个py文件，供scrapy爬虫调用，示例代码如下：

def get_cookie():cookie_list = []try:url = "http://demo/xxx"data = {"param1": "1","param2": "sfavxv","username": "username","password": "password"}data = urllib.parse.urlencode(data).encode('utf-8')req = urllib.request.Request(url, data=data, method='POST')with urllib.request.urlopen(req) as response:# 获取响应头中的 Cookiecookies = response.headers.get('Set-Cookie')if cookies:cookie_list = cookies.split(', ')else:print("No cookies found in response headers")except Exception as e:print(e)result = get_result(cookie_list)cookies_dict = cookie_str_to_dict(result)print(cookies_dict)return cookies_dictdef get_result(cookie_list):result = ""if cookie_list:# 拼接结果result = "; ".join(cookie.split(';')[0] for cookie in cookie_list)return resultdef cookie_str_to_dict(cookie_str):cookie_dict = {}cookies = cookie_str.split('; ')for cookie in cookies:key, value = cookie.split('=')# path 不需要if key != "Path":cookie_dict[key] = value# 现阶段手动补充cookie, 后续找到方法再重写cookie_dict['cd_1'] = ''cookie_dict['iswbzh'] = 1return cookie_dict

3. 数据建模

数据建模是开始爬取前的固定步骤之一。数据建模实际上是在Scrapy项目里的item.py ，创建一个继承 scrapy.Item 的 类。我们需要在爬取逻辑中返回这个类的对象，并可将该对象当作字典使用。具体接收这一步，框架内部已经帮我们封装好了。一般开发时我们不必过多关注此步。

3.1 Scrapy 数据建模

数据建模demo代码如下:

class DemoSpiderItem(scrapy.Item):# 代码code = scrapy.Field()# 使用类型use_type = scrapy.Field()# 使用编号use_num = scrapy.Field()

注意，这个类首先要继承scrapy.Item类，然后每个字段都要调用scrapy.Field()。语义上，这表示这个类的字段值都来自爬取的字段。

3.2 SQLAlchemy 创建实体类

在数据获取完毕后，我们需要对数据进行存储。我选择的是使用SQLAlchemy进行存储。因此，在数据建模的同时，我们需要创建SQLAlchemy实体类。这是为了在下一步获取到数据后，可以通过ORM直接执行持久化操作。示例如下：

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base# 创建引擎
engine = create_engine('mysql://root:root@localhost:3306/kcl2024?charset=utf8', echo=True)
Base = declarative_base()# 定义映射关系类class Subject(Base):__tablename__ = 'subject'id = Column(Integer, primary_key=True)name = Column(String(255), unique=True, nullable=True)score = Column(Integer, unique=False)

3. 分析网页

常见的网页分析获取数据方法有xpath和css，基本上这两种就够用了。前提其它方式，请读者自行了解。

3.1 xpath 分析

调用 response 的 xpath，以下是parse方法里的示例：

	def parse(self, response, *args, **kwargs):# 二级页面表格result_table = response.xpath('//*[@id="dataForm"]/div/div/table')# xpath示例company_name = result_table.xpath('//*[@id="SYDWNAME"]/@value').get()# 获取其它信息列表other_list = response.xpath('//*[@id="tr001"]')for result in other_list:item['device_code'] = result.xpath('//*[@id="tr001"]/td[4]/text()').get().replace(" ", '')item['company_name'] = company_nameyield item

3.2 css 分析

scrapy 对css分析前，需要先创建Select对象，接下来的操作就很好理解了:

    def parse(self, response, *args, **kwargs):selector = Selector(response)# 获取css xpathmaintain_contract_num = selector.xpath('//input[@name="HTBH"]/@value').get()# 获取css xpatheffective_time = selector.xpath('//input[@name="EXPDATE"]/@value').get()# 其它逻辑......

3.3 分页/多级页面策略

当我们要分析的是分页/多级页面时，可以在start_requests方法里先进行页面的预处理，也就是说，先手动分析分页的参数，然后在start_requests方法里事先拼接好二级页面的url参数，然后再通过循环等方式请求，将具体的response交给parse方法解析。

当然，也可以再创建几个中间方法执行循环处理，只需要请求后的response 都 yield parse 方法即可。

如下示例：

    def start_requests(self):cookies_dict = target_website_cookie.get_cookie()# 二级页面初始urlsecondary_start_url = ("http://demo/url")# 定义 二级页面 url 所需参数 字典 数组secondary_params = []# 访问一级页面获取关键信息start_response = scrapy.Request(self.start_urls[0], cookies=cookies_dict)for element in start_response.xpath('//tr[onclick^=ECSideUtil.selectRow]'):tds = element.xpath('td')# 提取do_edit的参数secondary_id = tds[2].xpath('./input[@onclick]/@onclick').get().split("'")[1]# 提取其它信息review_status = tds[9].xpath('text()').get()# 将id和其它信息存入数组secondary_params.append({"secondary_id": secondary_id, "review_status": review_status})for secondary_param in secondary_params:# 拼接二级页面urlsecondary_url = secondary_start_url + secondary_param["secondary_id"]# 携带 cookie 直接访问目标页面yield scrapy.Request(secondary_url, cookies=cookies_dict, callback=self.parse, meta=secondary_param)

4. 管道持久化

分析到的数据总得存起来呗。本文我们采用Mysql数据库存储数据，采用Scrapy管道+SQLAlchemy的策略做数据持久化。

在上面数据建模的小节，我们就已经介绍过SQLAlchemy。这一小节，我们需要在管道里进行持久化操作。

4.1 创建管道

class DemoSpiderPipeline:# 定义用于接收 item 字典列表def __init__(self):self.item_list = []@staticmethoddef open_spider(spider):if spider.name == "TestSpider":def process_item(self, item, spider):if spider.name == "TestSpider":# 将item转换为字典data = dict(item)# 将 data 加入 item_listself.item_list.append(data)return itemdef close_spider(self, spider):# 数据存储逻辑.....

创建管道主要通过spider名来判断具体执行哪个管道。其中，创建的管道主要有三部分：open_spider、process_item以及close_spider。

其中，open_spider 里 执行的是开启管道时的操作，process_item 通常用于对爬取到的数据进行数据处理。注意，process_item 必须 return item。最后是close_spider方法，我们可以在管道结束时做数据进一步处理以及持久化，然后关闭所需资源。

SqlAlchemy存储示例
上面的示例代码存储逻辑用注释表示。若读者也是使用的SqlAlchemy，可以参考如下代码进行数据存储。

            for temp_dict in self.item_list:# 将主键的值设置为空temp_dict['id'] = Nonedemo_data = DemoClass(temp_dict)spider_list.append(demo_data)if spider_list:with Session(db_init.engine) as session:# 1.数据准备：要增加的列表，要删除的列表，要更新的列表、现有数据insert_list = spider_listdelete_list = []# 查询现有的数据try:exist_list = session.query(DemoClass).all()delete_list = exist_listexcept Exception as e:print(e)# 2.执行删除if delete_list:for delete_obj in delete_list:session.delete(delete_obj)session.commit()# 3.执行新增if insert_list:session.add_all(insert_list)session.commit()

5. 补充与总结

5.1 如何在FastAPI触发scrapy？

已知我们通常在本地执行scrapy crawl spider_name来启动本地爬虫。

若没有特殊需求，我们可以用Python来帮我们执行这个命令，示例FastAPI代码如下：

@app.get("/test")
def galaxy_csel():result = ''command = 'scrapy crawl testSpider --nolog'try:result = subprocess.run(command, shell=True, check=True, cwd=".")except subprocess.CalledProcessError as e:print(f"Error executing command: {e}")return {f'execute message:{result}'}

5.2 总结

本文主要基于个人的实际开发，总结了Scrapy的基本用法，并且整合了FastAPI：

• 项目配置、创建
• Spider 网页分析的使用
• 数据建模
• 管道的使用
• Alchemy 数据库存储

结语
感谢大家阅读此文，希望我的内容能够给您带来启发和帮助。如果您喜欢我的文章，欢迎点赞、评论和分享，让更多人看到。期待与您在下一篇文章再相会！

参考资料

• bilibili-Scrapy优秀教程
• Scrapy- 菜鸟教程