Skip to content

Commit 89dffb7

Browse files
author
Boris
committed
修复表名提取错误的bug
1 parent 7baadb0 commit 89dffb7

6 files changed

Lines changed: 284 additions & 66 deletions

File tree

‎README.md‎

Lines changed: 111 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -72,57 +72,134 @@ From Git:
7272

7373
## 为什么不使用scrapy
7474

75-
**scrapy**给我的印象：
75+
scrapy给我的印象：
7676

77-
1. **重**：框架中的许多东西都用不到，如CrawlSpider、XMLFeedSpider
77+
1. 重，框架中的许多东西都用不到，如CrawlSpider、XMLFeedSpider
7878
2. 中间件不灵活
79-
3. 从数据库中取任务作为种子抓取不支持，需要自己写代码取任务，维护任务状态
79+
3. 不支持从数据库中取任务作为种子抓取
8080
4. 数据入库不支持批量，需要自己写批量逻辑
8181
5. 启动方式需要用scrapy命令行，打断点调试不方便
82-
6. 英文文档，阅读理解起来费精力
8382

84-
**feapder** 正是迎着以上痛点而生的，以数据库中取任务作为种子为例，写法如下：
83+
## 举例说明
8584

85+
本文以某东的商品爬虫为例，假如我们有1亿个商品，需要每7天全量更新一次，如何做呢？
86+
87+
### 1. 准备种子任务
88+
89+
首先需要个种子任务表来存储这些商品id，设计表如下：
90+
91+
![-w1028](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/03/09/16152931277517.jpg?x-oss-process=style/markdown-media)
92+
93+
```sql
94+
CREATE TABLE `jd_item_task` (
95+
`id` int(11) NOT NULL AUTO_INCREMENT,
96+
`item_id` varchar(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL COMMENT '商品id',
97+
`state` int(11) DEFAULT '0' COMMENT '任务状态 0 待抓取 1 抓取成功 2 抓取中 -1 抓取失败',
98+
PRIMARY KEY (`id`)
99+
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
86100
```
87-
import feapder
88-
from items import *
89101

102+
然后将这1亿个商品id录入进来，作为种子任务
103+
104+
![-w357](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/03/09/16152932156268.jpg?x-oss-process=style/markdown-media)
105+
106+
### 2. 准备数据表
107+
108+
![-w808](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/03/09/16152934374807.jpg?x-oss-process=style/markdown-media)
90109

91-
class TestSpider(feapder.BatchSpider):
110+
```sql
111+
CREATE TABLE `jd_item` (
112+
`id` int(11) NOT NULL AUTO_INCREMENT,
113+
`title` varchar(255) DEFAULT NULL,
114+
`batch_date` date DEFAULT NULL COMMENT '批次时间',
115+
`crawl_time` datetime DEFAULT NULL COMMENT '采集时间',
116+
PRIMARY KEY (`id`)
117+
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
118+
```
119+
120+
需求是每7天全量更新一次，即数据要以7天为维度划分，因此设置个`batch_date`字段，表示每条数据所属的批次。
121+
122+
这里只是演示，因此只采集标题字段
123+
124+
### 3. 采集
125+
126+
若使用`scrapy`，需要手动将这些种子任务分批取出来发给爬虫，还需要维护种子任务的状态，以及上面提及的批次信息`batch_date`。并且为了保证数据的时效性，需要对采集进度进行监控，写个爬虫十分繁琐。
127+
128+
而`feapder`内置了批次爬虫，可以很方便的应对这个需求。完整的爬虫写法如下：
129+
130+
```python
131+
import feapder
132+
from feapder import Item
133+
from feapder.utils import tools
134+
135+
136+
class JdSpider(feapder.BatchSpider):
137+
# 自定义数据库，若项目中有setting.py文件，此自定义可删除
138+
__custom_setting__ = dict(
139+
REDISDB_IP_PORTS="localhost:6379",
140+
REDISDB_DB=0,
141+
MYSQL_IP="localhost",
142+
MYSQL_PORT=3306,
143+
MYSQL_DB="feapder",
144+
MYSQL_USER_NAME="feapder",
145+
MYSQL_USER_PASS="feapder123",
146+
)
92147

93148
def start_requests(self, task):
94-
# task 为在任务表中取出的每一条任务
95-
id, url = task # id， url为所取的字段，main函数中指定的
96-
yield feapder.Request(url, task_id=id)
149+
task_id, item_id = task
150+
url = "https://item.jd.com/{}.html".format(item_id)
151+
yield feapder.Request(url, task_id=task_id) # 携带task_id字段
97152

98153
def parse(self, request, response):
99-
title = response.xpath('//title/text()').extract_first() # 取标题
100-
item = spider_data_item.SpiderDataItem() # 声明一个item
101-
item.title = title # 给item属性赋值
102-
yield item # 返回item， item会自动批量入库
103-
yield self.update_task_batch(request.task_id, 1) # 更新任务状态为1
104-
154+
title = response.xpath("string(//div[@class='sku-name'])").extract_first(default="").strip()
155+
156+
item = Item()
157+
item.table_name = "jd_item" # 指定入库的表名
158+
item.title = title
159+
item.batch_date = self.batch_date # 获取批次信息，批次信息框架自己维护
160+
item.crawl_time = tools.get_current_date() # 获取当前时间
161+
yield item # 自动批量入库
162+
yield self.update_task_batch(request.task_id, 1) # 更新任务状态
163+
164+
105165
if __name__ == "__main__":
106-
spider = TestSpider(
107-
redis_key="feapder:test_batch_spider", # redis中存放任务等信息的根key
108-
task_table="batch_spider_task", # mysql中的任务表
109-
task_keys=["id", "url"], # 需要获取任务表里的字段名，可添加多个
166+
spider = JdSpider(
167+
redis_key="feapder:jd_item", # redis中存放任务等信息key前缀
168+
task_table="jd_item_task", # mysql中的任务表
169+
task_keys=["id", "item_id"], # 需要获取任务表里的字段名，可添加多个
110170
task_state="state", # mysql中任务状态字段
111-
batch_record_table="batch_spider_batch_record", # mysql中的批次记录表
112-
batch_name="批次爬虫测试(周全)", # 批次名字
113-
batch_interval=7, # 批次周期 天为单位 若为小时 可写 1 / 24。 这里为每一天一个批次
171+
batch_record_table="jd_item_batch_record", # mysql中的批次记录表，自动生成
172+
batch_name="京东商品爬虫(周度全量)", # 批次名字
173+
batch_interval=7, # 批次周期 天为单位 若为小时 可写 1 / 24
114174
)
115175

176+
# 下面两个启动函数 相当于 master、worker。需要分开运行
177+
spider.start_monitor_task() # maser: 下发及监控任务
178+
# spider.start() # worker: 采集
116179

117-
spider.start_monitor_task() # 下发及监控任务
118-
# spider.start() # 采集
119180
```
120181

121-
任务表：`batch_spider_task`
122-
![-w398](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/02/22/16139773315622.jpg?x-oss-process=style/markdown-media)
182+
我们分别运行`spider.start_monitor_task()`与`spider.start()`，待爬虫结束后，观察数据库
183+
184+
**任务表**：`jd_item_task`
185+
186+
![-w282](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/03/09/16152953028811.jpg?x-oss-process=style/markdown-media)
187+
188+
任务均已完成了，框架有任务丢失重发机制，直到所有任务均已做完
189+
190+
**数据表**：`jd_item`:
123191

124-
批次记录表记录着每个批次的抓取状态，自动生成
125-
![-w899](http://markdown-media.oss-cn-beijing.aliyuncs.com/2020/12/20/16084680404224.jpg?x-oss-process=style/markdown-media)
192+
![-w569](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/03/09/16152952623851.jpg?x-oss-process=style/markdown-media)
193+
194+
数据里携带了批次时间信息，我们可以根据这个时间来对数据进行划分。当前批次为3月9号，若7天一批次，则下一批次为3月18号。
195+
196+
**批次表**：`jd_item_batch_record`
197+
198+
![-w901](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/03/09/16152953428596.jpg?x-oss-process=style/markdown-media)
199+
200+
启动参数中指定，自动生成。批次表里详细记录了每个批次的抓取状态，如任务总量、已做量、失败量、是否已完成等信息
201+
202+
### 4. 监控
126203

127204
feapder会自动维护任务状态，每个批次（采集周期）的进度，并且内置丰富的报警，保证我们的数据时效性，如：
128205

@@ -139,10 +216,13 @@ feapder会自动维护任务状态，每个批次（采集周期）的进度，
139216

140217
![-w416](http://markdown-media.oss-cn-beijing.aliyuncs.com/2020/12/29/16092335882158.jpg?x-oss-process=style/markdown-media)
141218

219+
1. 下载情况监控
142220

143-
## 学习交流
221+
![-w1299](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/02/09/16128568548280.jpg?x-oss-process=style/markdown-media)
144222

145223

224+
## 学习交流
225+
146226
知识星球：
147227

148228
![知识星球](http://markdown-media.oss-cn-beijing.aliyuncs.com/2020/02/16/zhi-shi-xing-qiu.jpeg)

‎docs/README.md‎

Lines changed: 111 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -72,57 +72,134 @@ From Git:
7272

7373
## 为什么不使用scrapy
7474

75-
**scrapy**给我的印象：
75+
scrapy给我的印象：
7676

77-
1. **重**：框架中的许多东西都用不到，如CrawlSpider、XMLFeedSpider
77+
1. 重，框架中的许多东西都用不到，如CrawlSpider、XMLFeedSpider
7878
2. 中间件不灵活
79-
3. 从数据库中取任务作为种子抓取不支持，需要自己写代码取任务，维护任务状态
79+
3. 不支持从数据库中取任务作为种子抓取
8080
4. 数据入库不支持批量，需要自己写批量逻辑
8181
5. 启动方式需要用scrapy命令行，打断点调试不方便
82-
6. 英文文档，阅读理解起来费精力
8382

84-
**feapder** 正是迎着以上痛点而生的，以数据库中取任务作为种子为例，写法如下：
83+
## 举例说明
8584

85+
本文以某东的商品爬虫为例，假如我们有1亿个商品，需要每7天全量更新一次，如何做呢？
86+
87+
### 1. 准备种子任务
88+
89+
首先需要个种子任务表来存储这些商品id，设计表如下：
90+
91+
![-w1028](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/03/09/16152931277517.jpg?x-oss-process=style/markdown-media)
92+
93+
```sql
94+
CREATE TABLE `jd_item_task` (
95+
`id` int(11) NOT NULL AUTO_INCREMENT,
96+
`item_id` varchar(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL COMMENT '商品id',
97+
`state` int(11) DEFAULT '0' COMMENT '任务状态 0 待抓取 1 抓取成功 2 抓取中 -1 抓取失败',
98+
PRIMARY KEY (`id`)
99+
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
86100
```
87-
import feapder
88-
from items import *
89101

102+
然后将这1亿个商品id录入进来，作为种子任务
103+
104+
![-w357](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/03/09/16152932156268.jpg?x-oss-process=style/markdown-media)
105+
106+
### 2. 准备数据表
107+
108+
![-w808](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/03/09/16152934374807.jpg?x-oss-process=style/markdown-media)
90109

91-
class TestSpider(feapder.BatchSpider):
110+
```sql
111+
CREATE TABLE `jd_item` (
112+
`id` int(11) NOT NULL AUTO_INCREMENT,
113+
`title` varchar(255) DEFAULT NULL,
114+
`batch_date` date DEFAULT NULL COMMENT '批次时间',
115+
`crawl_time` datetime DEFAULT NULL COMMENT '采集时间',
116+
PRIMARY KEY (`id`)
117+
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
118+
```
119+
120+
需求是每7天全量更新一次，即数据要以7天为维度划分，因此设置个`batch_date`字段，表示每条数据所属的批次。
121+
122+
这里只是演示，因此只采集标题字段
123+
124+
### 3. 采集
125+
126+
若使用`scrapy`，需要手动将这些种子任务分批取出来发给爬虫，还需要维护种子任务的状态，以及上面提及的批次信息`batch_date`。并且为了保证数据的时效性，需要对采集进度进行监控，写个爬虫十分繁琐。
127+
128+
而`feapder`内置了批次爬虫，可以很方便的应对这个需求。完整的爬虫写法如下：
129+
130+
```python
131+
import feapder
132+
from feapder import Item
133+
from feapder.utils import tools
134+
135+
136+
class JdSpider(feapder.BatchSpider):
137+
# 自定义数据库，若项目中有setting.py文件，此自定义可删除
138+
__custom_setting__ = dict(
139+
REDISDB_IP_PORTS="localhost:6379",
140+
REDISDB_DB=0,
141+
MYSQL_IP="localhost",
142+
MYSQL_PORT=3306,
143+
MYSQL_DB="feapder",
144+
MYSQL_USER_NAME="feapder",
145+
MYSQL_USER_PASS="feapder123",
146+
)
92147

93148
def start_requests(self, task):
94-
# task 为在任务表中取出的每一条任务
95-
id, url = task # id， url为所取的字段，main函数中指定的
96-
yield feapder.Request(url, task_id=id)
149+
task_id, item_id = task
150+
url = "https://item.jd.com/{}.html".format(item_id)
151+
yield feapder.Request(url, task_id=task_id) # 携带task_id字段
97152

98153
def parse(self, request, response):
99-
title = response.xpath('//title/text()').extract_first() # 取标题
100-
item = spider_data_item.SpiderDataItem() # 声明一个item
101-
item.title = title # 给item属性赋值
102-
yield item # 返回item， item会自动批量入库
103-
yield self.update_task_batch(request.task_id, 1) # 更新任务状态为1
104-
154+
title = response.xpath("string(//div[@class='sku-name'])").extract_first(default="").strip()
155+
156+
item = Item()
157+
item.table_name = "jd_item" # 指定入库的表名
158+
item.title = title
159+
item.batch_date = self.batch_date # 获取批次信息，批次信息框架自己维护
160+
item.crawl_time = tools.get_current_date() # 获取当前时间
161+
yield item # 自动批量入库
162+
yield self.update_task_batch(request.task_id, 1) # 更新任务状态
163+
164+
105165
if __name__ == "__main__":
106-
spider = TestSpider(
107-
redis_key="feapder:test_batch_spider", # redis中存放任务等信息的根key
108-
task_table="batch_spider_task", # mysql中的任务表
109-
task_keys=["id", "url"], # 需要获取任务表里的字段名，可添加多个
166+
spider = JdSpider(
167+
redis_key="feapder:jd_item", # redis中存放任务等信息key前缀
168+
task_table="jd_item_task", # mysql中的任务表
169+
task_keys=["id", "item_id"], # 需要获取任务表里的字段名，可添加多个
110170
task_state="state", # mysql中任务状态字段
111-
batch_record_table="batch_spider_batch_record", # mysql中的批次记录表
112-
batch_name="批次爬虫测试(周全)", # 批次名字
113-
batch_interval=7, # 批次周期 天为单位 若为小时 可写 1 / 24。 这里为每一天一个批次
171+
batch_record_table="jd_item_batch_record", # mysql中的批次记录表，自动生成
172+
batch_name="京东商品爬虫(周度全量)", # 批次名字
173+
batch_interval=7, # 批次周期 天为单位 若为小时 可写 1 / 24
114174
)
115175

176+
# 下面两个启动函数 相当于 master、worker。需要分开运行
177+
spider.start_monitor_task() # maser: 下发及监控任务
178+
# spider.start() # worker: 采集
116179

117-
spider.start_monitor_task() # 下发及监控任务
118-
# spider.start() # 采集
119180
```
120181

121-
任务表：`batch_spider_task`
122-
![-w398](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/02/22/16139773315622.jpg?x-oss-process=style/markdown-media)
182+
我们分别运行`spider.start_monitor_task()`与`spider.start()`，待爬虫结束后，观察数据库
183+
184+
**任务表**：`jd_item_task`
185+
186+
![-w282](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/03/09/16152953028811.jpg?x-oss-process=style/markdown-media)
187+
188+
任务均已完成了，框架有任务丢失重发机制，直到所有任务均已做完
189+
190+
**数据表**：`jd_item`:
123191

124-
批次记录表记录着每个批次的抓取状态，自动生成
125-
![-w899](http://markdown-media.oss-cn-beijing.aliyuncs.com/2020/12/20/16084680404224.jpg?x-oss-process=style/markdown-media)
192+
![-w569](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/03/09/16152952623851.jpg?x-oss-process=style/markdown-media)
193+
194+
数据里携带了批次时间信息，我们可以根据这个时间来对数据进行划分。当前批次为3月9号，若7天一批次，则下一批次为3月18号。
195+
196+
**批次表**：`jd_item_batch_record`
197+
198+
![-w901](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/03/09/16152953428596.jpg?x-oss-process=style/markdown-media)
199+
200+
启动参数中指定，自动生成。批次表里详细记录了每个批次的抓取状态，如任务总量、已做量、失败量、是否已完成等信息
201+
202+
### 4. 监控
126203

127204
feapder会自动维护任务状态，每个批次（采集周期）的进度，并且内置丰富的报警，保证我们的数据时效性，如：
128205

@@ -139,6 +216,10 @@ feapder会自动维护任务状态，每个批次（采集周期）的进度，
139216

140217
![-w416](http://markdown-media.oss-cn-beijing.aliyuncs.com/2020/12/29/16092335882158.jpg?x-oss-process=style/markdown-media)
141218

219+
1. 下载情况监控
220+
221+
![-w1299](http://markdown-media.oss-cn-beijing.aliyuncs.com/2021/02/09/16128568548280.jpg?x-oss-process=style/markdown-media)
222+
142223

143224
## 学习交流
144225

‎feapder/VERSION‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
1.1.8
1+
1.1.9

‎feapder/buffer/item_buffer.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -227,7 +227,7 @@ def __pick_items(self, items, is_update_item=False):
227227
def __export_to_db(self, tab_item, datas, is_update=False, update_keys=()):
228228
export_success = False
229229
# 打点 校验
230-
to_table = tools.get_info(tab_item, ":s_(.*?)_item", fetch_one=True)
230+
to_table = tools.get_info(tab_item, ":s_(.*?)_item$", fetch_one=True)
231231
item_name = to_table + "_item"
232232
self.check_datas(table=to_table, datas=datas)
233233

‎feapder/utils/export_data.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ def every_table_per_export_callback(to_table, auto_update, batch_count):
102102
log.info("无表数据")
103103
for table in tables:
104104
from_table = table
105-
to_table = tools.get_info(str(from_table), ":s_(.*?)_item", fetch_one=True)
105+
to_table = tools.get_info(str(from_table), ":s_(.*?)_item$", fetch_one=True)
106106
if callable(every_table_per_export_callback):
107107
to_table, auto_update, batch_count = every_table_per_export_callback(
108108
to_table, auto_update, batch_count
@@ -125,7 +125,7 @@ def export_items(self, tab_item, items_data):
125125
@result:
126126
"""
127127

128-
to_table = tools.get_info(tab_item, ":s_(.*?)_item", fetch_one=True)
128+
to_table = tools.get_info(tab_item, ":s_(.*?)_item$", fetch_one=True)
129129
sql, datas = tools.make_batch_sql(to_table, items_data)
130130
add_count = self.to_db.add_batch(sql, datas)
131131
datas_size = len(datas)
@@ -149,7 +149,7 @@ def update_items(self, tab_item, items_data, update_keys=()):
149149
---------
150150
@result:
151151
"""
152-
to_table = tools.get_info(tab_item, ":s_(.*?)_item", fetch_one=True)
152+
to_table = tools.get_info(tab_item, ":s_(.*?)_item$", fetch_one=True)
153153
sql, datas = tools.make_batch_sql(
154154
to_table,
155155
items_data,

0 commit comments

Comments
 (0)