feapder框架

import feapder


class MySpider(feapder.AirSpider):
    
    # 发送请求方法
    def start_requests(self):
        yield feapder.Request("https://spidertools.cn")

    # 解析响应方法
    def parse(self, request, response):
        # 提取网站title
        print(response.xpath("//title/text()").extract_first())
        # 提取网站描述
        print(response.xpath("//meta[@name='description']/@content").extract_first())
        print("网站地址: ", response.url)


if __name__ == "__main__":
    # 构造对象后，调用start方法启动爬虫
    MySpider().start()

def get_response(self, save_cached=False):
    """
    获取带有selector功能的response
    @param save_cached: 保存缓存，方便调试时不用每次都重新下载
    @return:
    """
    pass

def get_response_from_cached(self, save_cached=True):
    """
    用于从缓存中取response
    当缓存不存在时，会先下载，然后将响应存入缓存，之后再返回响应
    @param save_cached: 保存缓存，方便调试时不用每次都重新下载
    @return:
    """
    pass

import feapder


class AirSpiderTest(feapder.AirSpider):
    
    # 爬虫自定义配置，仅对当前爬虫有效，优先级大于配置文件
    __custom_setting__ = dict(
        PROXY_EXTRACT_API="代理提取地址",
    )
    
    # 分发请求任务函数
    def start_requests(self):
        yield feapder.Request("https://www.baidu.com")
        # 设置自定义解析函数
        yield feapder.Request("url2", callback=self.parser_detail)
        # 设置自定义下载中间件
        yield feapder.Request("url3", download_midware=self.my_midware)

    # 默认响应解析函数
    def parse(self, request, response):
        # 抛出异常即可自动重试
        if response.status_code != 200:
        	raise Exception("非法页面")
        # 在parse方法中使用yield返回Request，实现递归爬取
        print(response)
        
    # 自定义解析函数
    def parse_detail(self, request, response):
        pass
    
    # 默认下载中间件，在parse函数之前调用
    def download_midware(self, request):
        request.headers = {'User-Agent':"lalala"}
        return request
    
    # 自定义下载中间件
    def my_midware(self, request):
        return request
    
    # 校验函数, 可用于校验response是否正确
    # 若函数内抛出异常，则重试请求
    # 若返回True 或 None，则进入解析函数
    # 若返回False，则抛弃当前请求
    # 可通过request.callback_name 区分不同的回调函数，编写不同的校验逻辑
    def validate(self, request, response):
        pass


if __name__ == "__main__":
    # 使用多线程
    AirSpiderTest(thread_count=10).start()

from feapder.db.mysqldb import MysqlDB

class AirSpiderTest(feapder.AirSpider):
    __custom_setting__ = dict(
        MYSQL_IP="localhost",
        MYSQL_PORT = 3306,
        MYSQL_DB = "feapder",
        MYSQL_USER_NAME = "feapder",
        MYSQL_USER_PASS = "feapder123"
    )
def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.db = MysqlDB()
    # MysqlDB操作方法详见官方文档

from feapder import Item


class SpiderDataItem(Item):
    """
    This class was generated by feapder.
    command: feapder create -i spider_data.
    """

    def __init__(self, *args, **kwargs):
        # self.id = None  # type : int(10) unsigned | allow_null : NO | key : PRI | default_value : None | extra : auto_increment | column_comment : 
        self.title = None  # type : varchar(255) | allow_null : YES | key :  | default_value : None | extra : | column_comment :

import feapder
from items import *


class TestSpider(feapder.Spider):
    def start_requests(self):
        yield feapder.Request("https://www.baidu.com")
        
    def parse(self, request, response):
        title = response.xpath("//title/text()").extract_first()
        item = spider_data_item.SpiderDataItem()
        item.title = title
        # 在parse中使用yield返回item，实现批量入库
        yield item

import feapder
from feapder import ArgumentParser


class TaskSpiderTest(feapder.TaskSpider):
    # 自定义数据库，若项目中有setting.py文件，此自定义可删除
    # redis必选，mysql可选
    __custom_setting__ = dict(
        REDISDB_IP_PORTS="localhost:6379",
        REDISDB_USER_PASS="",
        REDISDB_DB=0,
        MYSQL_IP="localhost",
        MYSQL_PORT=3306,
        MYSQL_DB="feapder",
        MYSQL_USER_NAME="feapder",
        MYSQL_USER_PASS="feapder123",
    )

    def add_task(self):
        # 添加种子任务：在调用start_monitor_task()时会调用这个函数用于初始化任务，即数据库中添加任务的url
        self._redisdb.zadd(self._task_table, {"id": 1, "url": "https://www.baidu.com"})

    def start_requests(self, task):
        task_id, url = task
        yield feapder.Request(url, task_id=task_id)

    def parse(self, request, response):
        print(response.xpath("//title/text()").extract_first())
        print(response.xpath("//meta[@name='description']/@content").extract_first())
        print("网站地址: ", response.url)

        # mysql 需要更新任务状态为做完 即 state=1
        # yield self.update_task_batch(request.task_id)

def start(args):
    """
    用mysql做种子表
    """
    spider = TaskSpiderTest(
        task_table="spider_task", # 任务表名
        task_keys=["id", "url"], # 表里查询的字段
        redis_key="test:task_spider", # redis里做任务队列的key
        keep_alive=True, # 是否常驻，适用于master/work模式，任务完成后爬虫不会退出，继续等待任务
    )
    if args == 1:
        spider.start_monitor_task()
    else:
        spider.start()


def start2(args):
    """
    用redis做种子表
    """
    spider = TaskSpiderTest(
        task_table="spider_task2", # 任务表名
        task_table_type="redis", # 任务表类型为redis
        redis_key="test:task_spider", # redis里做任务队列的key
        keep_alive=True, # 是否常驻
        use_mysql=False, # 若用不到mysql，可以不使用
    )
    # TaskSpider的爬取分为两步，分别是master和work
    # master负责初始化任务，监控批次进度，下发批次等，通过start_monitor_task()启动
    # work负责实际地爬取，通过start()启动
    if args == 1:
        spider.start_monitor_task()
    else:
        spider.start()


if __name__ == "__main__":
    parser = ArgumentParser(description="测试TaskSpider")
    parser.add_argument("--start", type=int, help="用mysql做种子表 (1|2）", function=start)
    parser.add_argument("--start2", type=int, help="用redis做种子表 (1|2）", function=start2)
    parser.start()

    # 下发任务  python3 task_spider_test.py --start 1
    # 采集  python3 task_spider_test.py --start 2

import feapder
from feapder import ArgumentParser
from items import *

class TestSpider(feapder.BatchSpider):
    
    # 该方法用于重置任务，每个批次开始时，默认重置非状态-1的任务为状态0
    # 将该方法重写置空，可以实现增量爬取
    def init_task(self):
        pass

    def start_requests(self, task):
        # task为任务表中取出的一条任务
        id, url = task  # id、url为所取的字段，在main函数中指定
        # id, url = task.id, task.url
        yield feapder.Request(url, task_id=id, render=True)  # task_id为任务id，用于更新任务状态

    def parse(self, request, response):
        title = response.xpath('//title/text()').extract_first()  # 取标题
        item = spider_data_item.SpiderDataItem()  # 声明一个item
        item.title = title  # 给item属性赋值
        yield item  # 返回item， item会自动批量入库
        yield self.update_task_batch(request.task_id, 1) # 更新任务状态为1

    def exception_request(self, request, response):
        """
        @summary: 请求或者parser里解析出异常的request
        @result: request / callback / None (返回值必须可迭代)
        """
        yield request

    def failed_request(self, request, response):
        """
        @summary: 超过最大重试次数的request
        @result: request / item / callback / None (返回值必须可迭代)
        """
        yield request
        yield self.update_task_batch(request.task_id, -1)  # 更新任务状态为-1

def crawl_test(args):
    spider = test_spider.TestSpider(
        redis_key="feapder:test_batch_spider",
        task_table="batch_spider_task",  # mysql中的任务表
        task_keys=["id", "url"],  # 需要获取任务表里的字段名，可添加多个
        task_state="state",  # mysql中任务状态字段
        batch_record_table="batch_spider_record",  # mysql中的批次记录表，feapder自动创建
        batch_name="批次爬虫测试",  # 批次名字
        batch_interval=7,  # 批次周期，以天为单位，若以小时为单位，可写1 / 24
    )

    if args == 1:
        spider.start_monitor_task()  # 下发及监控任务
    else:
        spider.start()  # 采集

if __name__ == "__main__":
    parser = ArgumentParser(description="批次爬虫测试")
    parser.add_argument("--start", type=int, function=crawl_test)
    parser.start()

import feapder
from items import *

# 将feapder.Spider改为feapder.BaseParser
class TestParser(feapder.BaseParser):
    def start_requests(self):
        yield feapder.Request("https://www.baidu.com")
        
    def parse(self, request, response):
        pass

spider = feapder.Spider(redis_key="feapder:test_spider_integration")
spider.add_parser(TestParser)  # 注意此处传入类名
spider.start()

import feapder

# 将feapder.BatchSpider改为feapder.BatchParser
class TencentNewsParser(feapder.BatchParser):

    def start_requests(self, task):
        task_id = task[0]
        url = task[1]
        yield feapder.Request(url, task_id=task_id)

    def parse(self, request, response):
        title = response.xpath("//title/text()").extract_first()
        print(self.name, title)
        yield self.update_task_batch(request.task_id, 1)
        
spider = feapder.BatchSpider(
    task_table="batch_spider_integration_task",  # mysql中的任务表
    batch_record_table="batch_spider_record",  # mysql中的批次记录表
    batch_name="批次爬虫集成测试",  # 批次名字
    batch_interval=7,  # 批次时间，以天为单位，若以小时为单位，可写1 / 24
    task_keys=["id", "url", "parser_name"],  # 集成BatchSpider，需要将BatchSpider的名字取出
    redis_key="feapder:test_batch_spider_integration",
    task_state="state",  # mysql中任务状态字段
)
spider.add_parser(SinaNewsParser)
spider.add_parser(TencentNewsParser)
# spider.start_monitor_task()
# spider.start()