spider_demo.py

import json
import sys
import traceback
import time
import asyncio
import aiohttp
import lzma
import hashlib
import func_spider as fn
if sys.platform not in ('win32', 'cygwin', 'cli'):
    import uvloop
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

get_md5 = lambda x: hashlib.md5(x.encode(encoding='utf-8')).hexdigest()


class NewsCrawlerAsync:
    def __init__(self, log_name):
        self._workers = 0
        self._workers_max = 30
        self.logger = fn.init_logger(log_name + '.log')
        self.loop = asyncio.get_event_loop()
        self.session = aiohttp.ClientSession(loop=self.loop)

    async def save_to_db(self, save_data: dict):
        _id = save_data['_id']
        html = save_data.get('html')
        if isinstance(html, str):
            html = html.encode('utf8')
        html_lzma = lzma.compress(html)  # 压缩html
        save_data['html'] = html_lzma
        good = False
        try:
            # 入库逻辑
            save_data
            good = True
        except Exception as e:
            if e.args[0] == 1062:
                # Duplicate entry
                good = True
            else:
                traceback.print_exc()
                raise e
        return good

    async def process(self, task: dict):
        # 提取任务数据逻辑
        url = task.get('url')

        http_return_code, html, redirected_url = await fn.fetch(self.session, url)
        # 反扒判断逻辑
        if http_return_code != 200:
            return
        url_hash = get_md5(url)
        save_data = {}
        save_data['_id'] = url_hash
        save_data['http_return_code'] = http_return_code
        save_data['html'] = html
        await self.save_to_db(save_data)
        self._workers -= 1

    async def loop_crawl(self):
        last_rating_time = time.time()
        counter = 0
        sleep_time = 3
        while 1:
            task_str = ''  # get one redis pool task
            if not task_str or ('{' not in task_str and '}' not in task_str):
                print(f'no task_str  to crawl or task_no_format, sleep {sleep_time}')
                await asyncio.sleep(sleep_time)
                continue
            task = task_str
            if not isinstance(task_str, dict):
                task = json.loads(task)
            self._workers += 1
            counter += 1
            print(f'crawl task:{task}')
            asyncio.ensure_future(self.process(task))

            gap = time.time() - last_rating_time
            if gap > 5:
                rate = counter / gap
                print('\tloop_crawl() rate:%s, counter: %s, workers: %s' % (round(rate, 2), counter, self._workers))
                last_rating_time = time.time()
                counter = 0
            if self._workers > self._workers_max:
                print('====== got workers_max, sleep 3 sec to next worker =====')
                await asyncio.sleep(sleep_time)

    def run(self):
        try:
            self.loop.run_until_complete(self.loop_crawl())
        except KeyboardInterrupt:
            print('stopped by yourself!')
            pass


if __name__ == '__main__':
    nc = NewsCrawlerAsync('yrx-async')
    nc.run()

func_spider.py

def init_logger(log_name: str):
    pass


async def fetch(session, url: str, headers=None, timeout=9):
    _headers = {
        'User-Agent': ('Mozilla/5.0 (compatible; MSIE 9.0; '
                       'Windows NT 6.1; Win64; x64; Trident/5.0)'),
    }
    if headers:
        _headers = headers
    try:
        async with session.get(url, headers=_headers, timeout=timeout) as response:
            http_return_code = response.status
            html = await response.read()
            encoding = response.get_encoding()
            if encoding == 'gb2312':
                encoding = 'gbk'
            html = html.decode(encoding, errors='ignore')
            redirected_url = str(response.url)
    except Exception as e:
        msg = 'Failed download: {} | exception: {}, {}'.format(url, str(type(e)), str(e))
        print(msg)
        html = ''
        http_return_code = 0
        redirected_url = url
    return http_return_code, html, redirected_url