site.py.example

"""
抓取服务示例文件
参考:
xicidaili.py
kuaidaili.py
"""

# 引入必要库
from src.app.ip_get import IPGet, SiteResponse
from src.lib.structs import SiteData, SiteResponseData

# 定义抓取唯一 KEY
key = 'example'


@IPGet.config(key)
def config():
    """
    服务配置，需指定 name 和 pages
    :return:
    """
    site = SiteData()
    site.name = '服务名称'
    site.pages = ['https://www.example.com']  # 要抓取的页面 list 类型
    return site


@IPGet.parse(key)
def parse(resp: SiteResponse):
    """
    解析每个 page 的内容，获取到 ip 和 port 后创建一个 SiteResponseData 对象，并通过 yield 方式返回
    :param resp:
    :return:
    """
    items = resp.xpath('//tr')[1:]
    for item in items:
        try:
            res = SiteResponseData()
            res.ip = item.xpath('.//td[1]//text()')[0]
            res.port = item.xpath('.//td[2]//text()')[0]
            yield res
        except Exception:
            continue


if __name__ == '__main__':
    """
    通过 IPGet 中的 test_crawl 方法，可以在当前文件测试当前的抓取服务
    """
    from src.lib.func import run_until_complete

    runner = IPGet.test_crawl(key)
    run_until_complete(runner)