zhangxiaomeng1991

 

最近想要抓取拉勾网的数据,最开始是使用Scrapy的,但是遇到了下面两个问题:

  1. 前端页面是用JS模板引擎生成的
  2. 接口主要是用POST提交参数的

目前不会处理使用JS模板引擎生成的HTML页面,用POST的提交参数的话,接口统一,也没有必要使用Scrapy,所以就萌生了自己写一个简单的Python爬虫的想法。

本文中的部分链接可能需要FQ。

参考资料:

  1. http://www.ibm.com/developerworks/aix/library/au-threadingpython/
  2. http://stackoverflow.com/questions/10525185/python-threading-how-do-i-lock-a-thread

一个爬虫的简单框架

一个简单的爬虫框架,主要就是处理网络请求,Scrapy使用的是Twisted(一个事件驱动网络框架,以非阻塞的方式对网络I/O进行异步处理),这里不使用异步处理,等以后再研究这个框架。如果使用的是Python3.4及其以上版本,到可以使用asyncio这个标准库。

这个简单的爬虫使用多线程来处理网络请求,使用线程来处理URL队列中的url,然后将url返回的结果保存在另一个队列中,其它线程在读取这个队列中的数据,然后写到文件中去。

该爬虫主要用下面几个部分组成。

1 URL队列和结果队列

将将要爬去的url放在一个队列中,这里使用标准库Queue。访问url后的结果保存在结果队列中

初始化一个URL队列

from Queue import Queue
urls_queue = Queue()
out_queue = Queue()

2 请求线程

使用多个线程,不停的取URL队列中的url,并进行处理:

import threading

class ThreadCrawl(threading.Thread):
    def __init__(self, queue, out_queue):
        threading.Thread.__init__(self)
        self.queue = queue
        self.out_queue = out_queue

    def run(self):
        while True:
            item = self.queue.get()
            self.queue.task_down()

下面是部分标准库Queue的使用方法:

Queue.get([block[, timeout]])
Remove and return an item from the queue. If optional args block is true and timeout is None (the default), block if necessary until an item is available.
Queue.task_done()
Indicate that a formerly enqueued task is complete. Used by queue consumer threads. For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete.

如果队列为空,线程就会被阻塞,直到队列不为空。处理队列中的一条数据后,就需要通知队列已经处理完该条数据。

处理线程

处理结果队列中的数据,并保存到文件中。如果使用多个线程的话,必须要给文件加上锁。

lock = threading.Lock()
f = codecs.open(\'out.txt\', \'w\', \'utf8\')

当线程需要写入文件的时候,可以这样处理:

with lock:
    f.write(something)

程序的执行结果

运行状态:
运行状态

抓取结果:
抓取结果

源码

代码还不完善,将会持续修改中。

# coding: utf-8
\'\'\'
Author mr_zys

Email myzysv5@sina.com 
\'\'\'

from Queue import Queue
import threading
import urllib2
import time
import json
import codecs
from bs4 import BeautifulSoup

urls_queue = Queue()
data_queue = Queue()
lock = threading.Lock()
f = codecs.open(\'out.txt\', \'w\', \'utf8\')


class ThreadUrl(threading.Thread):

    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue

    def run(self):
        pass


class ThreadCrawl(threading.Thread):

    def __init__(self, url, queue, out_queue):
        threading.Thread.__init__(self)
        self.url = url
        self.queue = queue
        self.out_queue = out_queue

    def run(self):
        while True:
            item = self.queue.get()
            data = self._data_post(item)
            try:
                req = urllib2.Request(url=self.url, data=data)
                res = urllib2.urlopen(req)
            except urllib2.HTTPError, e:
                raise e.reason
            py_data = json.loads(res.read())
            res.close()
            item[\'first\'] = \'false\'
            item[\'pn\'] = item[\'pn\'] + 1
            success = py_data[\'success\']
            if success:
                print \'Get success...\'
            else:
                print \'Get fail....\'
            print \'pn is : %s\' % item[\'pn\']
            result = py_data[\'content\'][\'result\']
            if len(result) != 0:
                self.queue.put(item)
            print \'now queue size is: %d\' % self.queue.qsize()
            self.out_queue.put(py_data[\'content\'][\'result\'])
            self.queue.task_done()

    def _data_post(self, item):
        pn = item[\'pn\']
        first = \'false\'
        if pn == 1:
            first = \'true\'
        return \'first=\' + first + \'&pn=\' + str(pn) + \'&kd=\' + item[\'kd\']

    def _item_queue(self):
        pass


class ThreadWrite(threading.Thread):

    def __init__(self, queue, lock, f):
        threading.Thread.__init__(self)
        self.queue = queue
        self.lock = lock
        self.f = f

    def run(self):
        while True:
            item = self.queue.get()
            self._parse_data(item)
            self.queue.task_done()

    def _parse_data(self, item):
        for i in item:
            l = self._item_to_str(i)
            with self.lock:
                print \'write %s\' % l
                self.f.write(l)

    def _item_to_str(self, item):
        positionName = item[\'positionName\']
        positionType = item[\'positionType\']
        workYear = item[\'workYear\']
        education = item[\'education\']
        jobNature = item[\'jobNature\']
        companyName = item[\'companyName\']
        companyLogo = item[\'companyLogo\']
        industryField = item[\'industryField\']
        financeStage = item[\'financeStage\']
        companyShortName = item[\'companyShortName\']
        city = item[\'city\']
        salary = item[\'salary\']
        positionFirstType = item[\'positionFirstType\']
        createTime = item[\'createTime\']
        positionId = item[\'positionId\']
        return positionName + \' \' + positionType + \' \' + workYear + \' \' + education + \' \' + \
            jobNature + \' \' + companyLogo + \' \' + industryField + \' \' + financeStage + \' \' + \
            companyShortName + \' \' + city + \' \' + salary + \' \' + positionFirstType + \' \' + \
            createTime + \' \' + str(positionId) + \'\n\'


def main():
    for i in range(4):
        t = ThreadCrawl(
            \'http://www.lagou.com/jobs/positionAjax.json\', urls_queue, data_queue)
        t.setDaemon(True)
        t.start()
    datas = [
        {\'first\': \'true\', \'pn\': 1, \'kd\': \'Java\'}
        #{\'first\': \'true\', \'pn\': 1, \'kd\': \'Python\'}
    ]
    for d in datas:
        urls_queue.put(d)
    for i in range(4):
        t = ThreadWrite(data_queue, lock, f)
        t.setDaemon(True)
        t.start()

    urls_queue.join()
    data_queue.join()

    with lock:
        f.close()
    print \'data_queue siez: %d\' % data_queue.qsize()
main()

总结

主要是熟悉使用Python的多线程编程,以及一些标准库的使用Queuethreading

分类:

技术点:

相关文章: