Skip to content

Commit 8a01432

Browse files
author
Boris
committed
支持pipline
1 parent 52191e9 commit 8a01432

28 files changed

Lines changed: 541 additions & 222 deletions

File tree

‎docs/_sidebar.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
* [Spider进阶](source_code/Spider进阶.md)
2323
* [BatchSpider进阶](source_code/BatchSpider进阶.md)
2424
* [配置文件](source_code/配置文件.md)
25+
* [数据管道-pipline](source_code/pipline.md)
2526
* [Item](source_code/Item.md)
2627
* [UpdateItem](source_code/UpdateItem.md)
2728
* [MysqlDB](source_code/MysqlDB.md)

‎docs/source_code/Response.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -200,7 +200,7 @@ response.code="utf-8"
200200
'�?�好'
201201
>>>str(content, errors='strict')
202202
Traceback (most recent call last):
203-
File "/Users/liubo/workspace/feapder/venv2/lib/python3.6/site-packages/IPython/core/interactiveshell.py", line 3343, in run_code
203+
File "/Users/Boris/workspace/feapder/venv2/lib/python3.6/site-packages/IPython/core/interactiveshell.py", line 3343, in run_code
204204
exec(code_obj, self.user_global_ns, self.user_ns)
205205
File "<ipython-input-11-a129a2aa6283>", line 1, in <module>
206206
str(content, errors='strict')

‎docs/source_code/pipline.md‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
# Pipline
2+
3+
Pipline是数据入库时流经的管道，默认为使用mysql入库，用户可自定义。
4+
5+
注：AirSpider不支持
6+
7+
## 使用方式
8+
9+
### 1. 编写pipline
10+
11+
```python
12+
from feapder.piplines import BasePipline
13+
from typing import Dict, List, Tuple
14+
15+
16+
class Pipline(BasePipline):
17+
"""
18+
pipline 是单线程的，批量保存数据的操作，不建议在这里写网络请求代码，如下载图片等
19+
"""
20+
21+
def save_items(self, table, items: List[Dict]) -> bool:
22+
"""
23+
保存数据
24+
Args:
25+
table: 表名
26+
items: 数据，[{},{},...]
27+
28+
Returns: 是否保存成功 True / False
29+
若False，不会将本批数据入到去重库，以便再次入库
30+
31+
"""
32+
33+
print("自定义pipline， 保存数据 >>>>", table, items)
34+
35+
return True
36+
37+
def update_items(self, table, items: List[Dict], update_keys=Tuple) -> bool:
38+
"""
39+
更新数据
40+
Args:
41+
table: 表名
42+
items: 数据，[{},{},...]
43+
update_keys: 更新的字段字段, 如 ("title", "publish_time")
44+
45+
Returns: 是否更新成功 True / False
46+
若False，不会将本批数据入到去重库，以便再次入库
47+
48+
"""
49+
50+
print("自定义pipline， 更新数据 >>>>", table, items, update_keys)
51+
52+
return True
53+
```
54+
55+
`Pipline`需继承`BasePipline`，类名和存放位置随意，需要实现`save_items`、`update_items`两个接口。一定要有返回值，返回`False`表示数据没保存成功，数据不入去重库，以便再次入库
56+
57+
### 2. 编写配置文件
58+
59+
```python
60+
# 数据入库的pipline，可自定义，默认MysqlPipline
61+
ITEM_PIPLINES = [
62+
"pipline.Pipline"
63+
]
64+
```
65+
66+
将编写好的pipline配置进来，值为类的模块路径，需要指定到具体的类名
67+
68+
## 示例
69+
70+
地址：

‎feapder/buffer/item_buffer.py‎

Lines changed: 43 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
@email: boris@bzkj.tech
99
"""
1010

11+
import importlib
1112
import threading
1213
from queue import Queue
1314

@@ -16,12 +17,14 @@
1617
from feapder.db.redisdb import RedisDB
1718
from feapder.dedup import Dedup
1819
from feapder.network.item import Item, UpdateItem
19-
from feapder.utils.export_data import ExportData
20+
from feapder.piplines import BasePipline
2021
from feapder.utils.log import log
2122

2223
MAX_ITEM_COUNT = 5000 # 缓存中最大item数
2324
UPLOAD_BATCH_MAX_SIZE = 1000
2425

26+
MYSQL_PIPLINE_PATH = "feapder.piplines.mysql_pipline.MysqlPipline"
27+
2528

2629
class Singleton(object):
2730
def __new__(cls, *args, **kwargs):
@@ -56,23 +59,34 @@ def __init__(self, redis_key):
5659
# 'xxx:xxx_item': ['id', 'name'...] # 记录redis中item名与需要更新的key对应关系
5760
}
5861

59-
self._export_data = ExportData() if setting.ADD_ITEM_TO_MYSQL else None
62+
self._piplines = self.load_piplines()
6063

61-
self.db_tip()
64+
self._have_mysql_pipline = MYSQL_PIPLINE_PATH in setting.ITEM_PIPLINES
65+
self._mysql_pipline = None
6266

6367
if setting.ITEM_FILTER_ENABLE and not self.__class__.dedup:
6468
self.__class__.dedup = Dedup(to_md5=False)
6569

66-
def db_tip(self):
67-
msg = ""
68-
if setting.ADD_ITEM_TO_MYSQL:
69-
msg += "item 自动入mysql "
70-
if setting.ADD_ITEM_TO_REDIS:
71-
msg += "item 自动入redis "
72-
if not msg:
73-
log.warning("*** 请注意检查item是否入库 !!!")
74-
else:
75-
log.info(msg)
70+
def load_piplines(self):
71+
piplines = []
72+
for pipline_path in setting.ITEM_PIPLINES:
73+
module, class_name = pipline_path.rsplit(".", 1)
74+
pipline_cls = importlib.import_module(module).__getattribute__(class_name)
75+
pipline = pipline_cls()
76+
if not isinstance(pipline, BasePipline):
77+
raise ValueError(f"{pipline_path} 需继承 feapder.piplines.BasePipline")
78+
piplines.append(pipline)
79+
80+
return piplines
81+
82+
@property
83+
def mysql_pipline(self):
84+
if not self._mysql_pipline:
85+
module, class_name = MYSQL_PIPLINE_PATH.rsplit(".", 1)
86+
pipline_cls = importlib.import_module(module).__getattribute__(class_name)
87+
self._mysql_pipline = pipline_cls()
88+
89+
return self._mysql_pipline
7690

7791
def run(self):
7892
while not self._thread_stop:
@@ -225,48 +239,29 @@ def __pick_items(self, items, is_update_item=False):
225239
return datas_dict
226240

227241
def __export_to_db(self, tab_item, datas, is_update=False, update_keys=()):
228-
export_success = False
229-
# 打点 校验
230242
to_table = tools.get_info(tab_item, ":s_(.*?)_item$", fetch_one=True)
231-
item_name = to_table + "_item"
232-
self.check_datas(table=to_table, datas=datas)
233243

234-
if setting.ADD_ITEM_TO_MYSQL: # 任务表需要入mysql
235-
if isinstance(setting.ADD_ITEM_TO_MYSQL, (list, tuple)):
236-
for item in setting.ADD_ITEM_TO_MYSQL:
237-
if item in item_name:
238-
export_success = (
239-
self._export_data.export_items(to_table, datas)
240-
if not is_update
241-
else self._export_data.update_items(
242-
to_table, datas, update_keys=update_keys
243-
)
244-
)
244+
# 打点 校验
245+
self.check_datas(table=to_table, datas=datas)
245246

246-
else:
247-
export_success = (
248-
self._export_data.export_items(to_table, datas)
249-
if not is_update
250-
else self._export_data.update_items(
251-
to_table, datas, update_keys=update_keys
247+
for pipline in self._piplines:
248+
if is_update:
249+
if not pipline.update_items(to_table, datas, update_keys=update_keys):
250+
log.error(
251+
f"{pipline.__class__.__name__} 更新数据失败. table: {to_table} items: {datas}"
252252
)
253-
)
254-
255-
if setting.ADD_ITEM_TO_REDIS:
256-
if isinstance(setting.ADD_ITEM_TO_REDIS, (list, tuple)):
257-
for item in setting.ADD_ITEM_TO_REDIS:
258-
if item in item_name:
259-
self._db.sadd(tab_item, datas)
260-
export_success = True
261-
log.info("共导出 %s 条数据 到redis %s" % (len(datas), tab_item))
262-
break
253+
return False
263254

264255
else:
265-
self._db.sadd(tab_item, datas)
266-
export_success = True
267-
log.info("共导出 %s 条数据 到redis %s" % (len(datas), tab_item))
256+
if not pipline.save_items(to_table, datas):
257+
log.error(
258+
f"{pipline.__class__.__name__} 保存数据失败. table: {to_table} items: {datas}"
259+
)
260+
return False
268261

269-
return export_success
262+
# 若是任务表, 且上面的pipline里没mysql，则需调用mysql更新任务
263+
if not self._have_mysql_pipline and is_update and to_table.endswith("_task"):
264+
self.mysql_pipline.update_items(to_table, datas, update_keys=update_keys)
270265

271266
def __add_item_to_db(
272267
self, items, update_items, requests, callbacks, items_fingerprints

‎feapder/piplines/__init__.py‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
# -*- coding: utf-8 -*-
2+
"""
3+
Created on 2021/3/17 10:57 下午
4+
---------
5+
@summary:
6+
---------
7+
@author: Boris
8+
@email: boris_liu@foxmail.com
9+
"""
10+
11+
import abc
12+
from typing import Dict, List, Tuple
13+
14+
15+
class BasePipline(metaclass=abc.ABCMeta):
16+
"""
17+
pipline 是单线程的，批量保存数据的操作，不建议在这里写网络请求代码，如下载图片等
18+
"""
19+
20+
@abc.abstractmethod
21+
def save_items(self, table, items: List[Dict]) -> bool:
22+
"""
23+
保存数据
24+
Args:
25+
table: 表名
26+
items: 数据，[{},{},...]
27+
28+
Returns: 是否保存成功 True / False
29+
若False，不会将本批数据入到去重库，以便再次入库
30+
31+
"""
32+
33+
return True
34+
35+
@abc.abstractmethod
36+
def update_items(self, table, items: List[Dict], update_keys=Tuple) -> bool:
37+
"""
38+
更新数据
39+
Args:
40+
table: 表名
41+
items: 数据，[{},{},...]
42+
update_keys: 更新的字段字段, 如 ("title", "publish_time")
43+
44+
Returns: 是否更新成功 True / False
45+
若False，不会将本批数据入到去重库，以便再次入库
46+
47+
"""
48+
49+
return True

‎feapder/piplines/mysql_pipline.py‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
# -*- coding: utf-8 -*-
2+
"""
3+
Created on 2018-07-29 22:48:30
4+
---------
5+
@summary: 导出数据
6+
---------
7+
@author: Boris
8+
@email: boris_liu@foxmail.com
9+
"""
10+
from typing import Dict, List, Tuple
11+
12+
import feapder.utils.tools as tools
13+
from feapder.db.mysqldb import MysqlDB
14+
from feapder.piplines import BasePipline
15+
from feapder.utils.log import log
16+
17+
18+
class MysqlPipline(BasePipline):
19+
def __init__(self):
20+
self._to_db = None
21+
22+
@property
23+
def to_db(self):
24+
if not self._to_db:
25+
self._to_db = MysqlDB()
26+
27+
return self._to_db
28+
29+
def save_items(self, table, items: List[Dict]) -> bool:
30+
"""
31+
保存数据
32+
Args:
33+
table: 表名
34+
items: 数据，[{},{},...]
35+
36+
Returns: 是否保存成功 True / False
37+
若False，不会将本批数据入到去重库，以便再次入库
38+
39+
"""
40+
41+
sql, datas = tools.make_batch_sql(table, items)
42+
add_count = self.to_db.add_batch(sql, datas)
43+
datas_size = len(datas)
44+
if add_count is None:
45+
log.error("导出数据到表 %s 失败" % (table))
46+
else:
47+
log.info(
48+
"共导出 %s 条数据 到 %s, 重复 %s 条" % (datas_size, table, datas_size - add_count)
49+
)
50+
51+
return add_count != None
52+
53+
def update_items(self, table, items: List[Dict], update_keys=Tuple) -> bool:
54+
"""
55+
更新数据
56+
Args:
57+
table: 表名
58+
items: 数据，[{},{},...]
59+
update_keys: 更新的字段字段, 如 ("title", "publish_time")
60+
61+
Returns: 是否更新成功 True / False
62+
若False，不会将本批数据入到去重库，以便再次入库
63+
64+
"""
65+
66+
sql, datas = tools.make_batch_sql(
67+
table, items, update_columns=update_keys or list(items[0].keys())
68+
)
69+
update_count = self.to_db.add_batch(sql, datas)
70+
if update_count is None:
71+
log.error("更新表 %s 数据失败" % (table))
72+
else:
73+
msg = "共更新 %s 条数据 到 %s" % (update_count // 2, table)
74+
if update_keys:
75+
msg += " 更新字段为 {}".format(update_keys)
76+
log.info(msg)
77+
78+
return update_count != None

‎feapder/setting.py‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,11 @@
3030
# 适用于redis哨兵模式
3131
REDISDB_SERVICE_NAME = os.getenv("REDISDB_SERVICE_NAME")
3232

33+
# 数据入库的pipline，可自定义，默认MysqlPipline
34+
ITEM_PIPLINES = [
35+
"feapder.piplines.mysql_pipline.MysqlPipline"
36+
]
37+
3338
# 爬虫相关
3439
# COLLECTOR
3540
COLLECTOR_SLEEP_TIME = 1 # 从任务队列中获取任务到内存队列的间隔

‎feapder/templates/project_template/setting.py‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,11 @@
1717
# 默认 0 到 15 共16个数据库
1818
REDISDB_DB = 0
1919

20+
# 数据入库的pipline，可自定义，默认MysqlPipline
21+
# ITEM_PIPLINES = [
22+
# "feapder.piplines.mysql_pipline.MysqlPipline"
23+
# ]
24+
2025
# # 爬虫相关
2126
# # COLLECTOR
2227
# COLLECTOR_SLEEP_TIME = 1 # 从任务队列中获取任务到内存队列的间隔

0 commit comments

Comments
 (0)