Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Commit e017bdd

Browse filesBrowse files
author
Boris
committed
完善文档, update_items改为非必须实现接口
1 parent 601f3fc commit e017bdd
Copy full SHA for e017bdd

7 files changed

+22-11Lines changed: 22 additions & 11 deletions

File tree

Expand file treeCollapse file tree
Open diff view settings
Filter options
Expand file treeCollapse file tree
Open diff view settings
Collapse file

‎docs/_sidebar.md‎

Copy file name to clipboardExpand all lines: docs/_sidebar.md
+1-1Lines changed: 1 addition & 1 deletion
  • Display the source diff
  • Display the rich diff
Original file line numberDiff line numberDiff line change
@@ -22,9 +22,9 @@
2222
* [Spider进阶](source_code/Spider进阶.md)
2323
* [BatchSpider进阶](source_code/BatchSpider进阶.md)
2424
* [配置文件](source_code/配置文件.md)
25-
* [数据管道-pipeline](source_code/pipeline.md)
2625
* [Item](source_code/Item.md)
2726
* [UpdateItem](source_code/UpdateItem.md)
27+
* [数据管道-pipeline](source_code/pipeline.md)
2828
* [MysqlDB](source_code/MysqlDB.md)
2929
* [RedisDB](source_code/RedisDB.md)
3030
* [工具库-tools](source_code/tools.md)
Collapse file

‎docs/source_code/pipline.md‎ ‎docs/source_code/pipeline.md‎docs/source_code/pipline.md renamed to docs/source_code/pipeline.md

Copy file name to clipboardExpand all lines: docs/source_code/pipeline.md
+6-4Lines changed: 6 additions & 4 deletions
  • Display the source diff
  • Display the rich diff
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
# Pipeline
22

3-
Pipeline是数据入库时流经的管道，默认为使用mysql入库，用户可自定义。
3+
Pipeline是数据入库时流经的管道，默认为使用mysql入库，用户可自定义，以便对接其他数据库。
44

55
注：AirSpider不支持
66

@@ -36,7 +36,7 @@ class Pipeline(BasePipeline):
3636

3737
def update_items(self, table, items: List[Dict], update_keys=Tuple) -> bool:
3838
"""
39-
更新数据
39+
更新数据, 与UpdateItem配合使用，若爬虫中没使用UpdateItem，则可不实现此接口
4040
Args:
4141
table: 表名
4242
items: 数据，[{},{},...]
@@ -52,12 +52,14 @@ class Pipeline(BasePipeline):
5252
return True
5353
```
5454

55-
`Pipeline`需继承`BasePipeline`，类名和存放位置随意，需要实现`save_items`、`update_items`两个接口。一定要有返回值，返回`False`表示数据没保存成功，数据不入去重库，以便再次入库
55+
`Pipeline`需继承`BasePipeline`，类名和存放位置随意，需要实现`save_items`接口。一定要有返回值，返回`False`表示数据没保存成功，数据不入去重库，以便再次入库
56+
57+
`update_items`接口与`UpdateItem`配合使用，更新数据时使用，若爬虫中没使用UpdateItem，则可不实现此接口
5658

5759
### 2. 编写配置文件
5860

5961
```python
60-
# 数据入库的pipeline，可自定义，默认MysqlPipeline
62+
# 数据入库的pipeline，支持多个
6163
ITEM_PIPELINES = [
6264
"pipeline.Pipeline"
6365
]
Collapse file

‎feapder/buffer/item_buffer.py‎

Copy file name to clipboardExpand all lines: feapder/buffer/item_buffer.py
+9-2Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
from feapder.dedup import Dedup
1919
from feapder.network.item import Item, UpdateItem
2020
from feapder.pipelines import BasePipeline
21+
from feapder.pipelines.mysql_pipeline import MysqlPipeline
2122
from feapder.utils.log import log
2223

2324
MAX_ITEM_COUNT = 5000 # 缓存中最大item数
@@ -37,13 +38,14 @@ def __new__(cls, *args, **kwargs):
3738
class ItemBuffer(threading.Thread, Singleton):
3839
dedup = None
3940

40-
def __init__(self, redis_key):
41+
def __init__(self, redis_key, task_table=None):
4142
if not hasattr(self, "_table_item"):
4243
super(ItemBuffer, self).__init__()
4344

4445
self._thread_stop = False
4546
self._is_adding_to_db = False
4647
self._redis_key = redis_key
48+
self._task_table = task_table
4749

4850
self._items_queue = Queue(maxsize=MAX_ITEM_COUNT)
4951
self._db = RedisDB()
@@ -246,6 +248,11 @@ def __export_to_db(self, tab_item, datas, is_update=False, update_keys=()):
246248

247249
for pipeline in self._pipelines:
248250
if is_update:
251+
if to_table == self._task_table and not isinstance(
252+
pipeline, MysqlPipeline
253+
):
254+
continue
255+
249256
if not pipeline.update_items(to_table, datas, update_keys=update_keys):
250257
log.error(
251258
f"{pipeline.__class__.__name__} 更新数据失败. table: {to_table} items: {datas}"
@@ -260,7 +267,7 @@ def __export_to_db(self, tab_item, datas, is_update=False, update_keys=()):
260267
return False
261268

262269
# 若是任务表, 且上面的pipeline里没mysql，则需调用mysql更新任务
263-
if not self._have_mysql_pipeline and is_update and to_table.endswith("_task"):
270+
if not self._have_mysql_pipeline and is_update and to_table == self._task_table:
264271
self.mysql_pipeline.update_items(to_table, datas, update_keys=update_keys)
265272

266273
def __add_item_to_db(
Collapse file

‎feapder/core/scheduler.py‎

Copy file name to clipboardExpand all lines: feapder/core/scheduler.py
+3-1Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@ def __init__(
4545
send_run_time=True,
4646
batch_interval=0,
4747
wait_lock=True,
48+
task_table=None
4849
):
4950
"""
5051
@summary: 调度器
@@ -59,6 +60,7 @@ def __init__(
5960
@param send_run_time: 发送运行时间
6061
@param batch_interval: 抓取时间间隔 默认为0 天为单位 多次启动时，只有当前时间与第一次抓取结束的时间间隔大于指定的时间间隔时，爬虫才启动
6162
@param wait_lock: 下发任务时否等待锁，若不等待锁，可能会存在多进程同时在下发一样的任务，因此分布式环境下请将该值设置True
63+
@param task_table: 任务表， 批次爬虫传递
6264
---------
6365
@result:
6466
"""
@@ -80,7 +82,7 @@ def __init__(
8082
)
8183

8284
self._request_buffer = RequestBuffer(redis_key)
83-
self._item_buffer = ItemBuffer(redis_key)
85+
self._item_buffer = ItemBuffer(redis_key, task_table)
8486

8587
self._collector = Collector(redis_key)
8688
self._parsers = []
Collapse file

‎feapder/core/spiders/batch_spider.py‎

Copy file name to clipboardExpand all lines: feapder/core/spiders/batch_spider.py
+1Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,7 @@ def __init__(
111111
auto_start_requests=False,
112112
send_run_time=send_run_time,
113113
batch_interval=batch_interval,
114+
task_table=task_table
114115
)
115116

116117
self._redisdb = RedisDB()
Collapse file

‎feapder/pipelines/__init__.py‎

Copy file name to clipboardExpand all lines: feapder/pipelines/__init__.py
+1-2Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,10 +32,9 @@ def save_items(self, table, items: List[Dict]) -> bool:
3232

3333
return True
3434

35-
@abc.abstractmethod
3635
def update_items(self, table, items: List[Dict], update_keys=Tuple) -> bool:
3736
"""
38-
更新数据
37+
更新数据, 与UpdateItem配合使用，若爬虫中没使用UpdateItem，则可不实现此接口
3938
Args:
4039
table: 表名
4140
items: 数据，[{},{},...]
Collapse file

‎tests/test-pipeline/pipeline.py‎

Copy file name to clipboardExpand all lines: tests/test-pipeline/pipeline.py
+1-1Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ def save_items(self, table, items: List[Dict]) -> bool:
3535

3636
def update_items(self, table, items: List[Dict], update_keys=Tuple) -> bool:
3737
"""
38-
更新数据
38+
更新数据, 与UpdateItem配合使用，若爬虫中没使用UpdateItem，则可不实现此接口
3939
Args:
4040
table: 表名
4141
items: 数据，[{},{},...]

0 commit comments

Comments
0 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.