51Testing软件测试论坛

标题: pyspider使用实例 [打印本页]

---

作者: 测试积点老人    时间: 2018-11-29 14:26
标题: pyspider使用实例


创建项目：

[attach]119687[/attach]
创建后项目内容：

[attach]119688[/attach]

实例
操作步骤：
[attach]119689[/attach]

[attach]119690[/attach]

数据库内容如下：

        使用mongodb数据库:

[attach]119691[/attach]

代码如下：

1. #!/usr/bin/env python
   
2. # -*- encoding: utf-8 -*-
   
3. # Created on 2018-06-14 17:11:18
   
4. # Project: qiushi
   
5. 
   
6. from pyspider.libs.base_handler import *
   
7. import pymongo, requests
   
8. from fake_useragent import UserAgent
   
9. import time
   
10. 
    
11. def get_proxy():
    
12.     return requests.get('http://localhost:5010/get/').text
    
13.     #由于代理池获取的免费代理，不稳定！因此后面的就不在使用代理，仅仅在这里把获取过程的封装函数放在这里！！
    
14.    
    
15. class Handler(BaseHandler):
    
16.    
    
17.     ua = UserAgent()
    
18.     headers = {
    
19.         'User-Agent': ua.random,
    
20.         # 'Host':
    
21.         # 'Referer':
    
22.     }
    
23.    
    
24.     # 创建client和db
    
25.     client = pymongo.MongoClient('localhost')
    
26.     db = client['qidian']
    
27.    
    
28.     # 对于整个爬虫项目的全局配置：所有的self.crawl()在请求的时候都会加载这个配置。
    
29.     crawl_config = {
    
30.         'headers': headers,
    
31.         # 'proxy': get_proxy(),
    
32.         'itag': 'v0'
    
33.     }
    
34.    
    
35.     # 增量爬虫1：每天重新启动爬虫的时候，只爬取页面上更新的数据。(采用去重策略)
    
36.     # 增量爬虫2：url没有变化，数据更新了。(不能采用去重，每天都要重新爬取)
    
37.      # 项目启动首先进入的函数
    
38.     # @every: 用于设置定时爬取任务：可以是minutes, 也可以设置为seconds。
    
39.     @every(seconds=2 * 60)
    
40.     def on_start(self):
    
41.         # 初始爬取的url
    
42.         self.crawl('https://www.qidian.com/all', callback=self.index_page,validate_cert=False)
    
43.    
    
44. 
    
45.     # age: 主要是对任务url进行去重/过滤(根据taskid)，每一个url有唯一的一个标识taskid，age是一个以秒为单位的时间点，
    
46.             #如果在这个时间范围内，遇到了相同的taskid，这个任务就会被丢弃。
    
47.    
    
48. 
    
49.     @config(age=60)
    
50.     def index_page(self, response):
    
51.         # response.doc返回一个pyquery对象
    
52.         for each in response.doc('h4 > a').items():
    
53.             #fetch_type='js', js_script="":如果是javascript代码，需要在self.crawl()加入该参数
    
54.              self.crawl(each.attr.href, callback=self.detail_page, validate_cert=False)
    
55.             
    
56.         # 找到下一页
    
57.         # next_page = response.doc('.lbf-pagination-item-list > li:nth-of-type(9) > a')
    
58.         # self.crawl(next_page.attr.href, callback=self.index_page, validate_cert=False)
    
59.    
    
60.    
    
61.     # age的默认值是-1，永远不过期。
    
62.     @config(priority=2,age=60)
    
63.     def detail_page(self, response):
    
64.         # 从response中解析详情页的数据
    
65.         name = response.doc('h1 > em').text()
    
66.         author = response.doc('h1 a').text()
    
67.         tag = response.doc('.tag').text()
    
68.         info = response.doc('.intro').text()
    
69.         
    
70.         print(time.time())
    
71.         
    
72.         print(name)
    
73.         print(author)
    
74.         print(tag)
    
75.         print(info)
    
76.         return {
    
77.             "time": str(time.time()),
    
78.             "url": response.url,
    
79.             "title": response.doc('title').text(),
    
80.             "name":name,
    
81.             "author":author,
    
82.             "tag":tag,
    
83.             "info":info,
    
84.         }
    
85.     # on_result是固定的函数，只要一个函数中有return，就会自动调用这个函数。
    
86.     def on_result(self, data):
    
87.         # 将detail_page函数返回的结果，保存至mongodb中
    
88.         #print('接收到数据了.....',response['url'])
    
89.         if data == None:
    
90.             print('空')
    
91.         else:
    
92.             if self.db['q'].update_one({'name': data['name']}, {'$set': data}, True):
    
93.                 print('数据保存成功')
    
94.             else:
    
95.                 print('数据保存失败')

复制代码

---

欢迎光临 51Testing软件测试论坛 (http://bbs.51testing.com/)

Powered by Discuz! X3.2