python的crawlee库介绍

pip install 'crawlee[all]'

playwright install

import asynciofrom crawlee.crawlers import BeautifulSoupCrawler, BeautifulSoupCrawlingContextasync def main() -> None:    crawler = BeautifulSoupCrawler()    @crawler.router.default_handler    async def request_handler(context: BeautifulSoupCrawlingContext) -> None:        url = context.request.url        # 使用 BeautifulSoup 对象提取标题        title = context.soup.title.string if context.soup.title else ''        context.log.info(f'网页 {url} 的标题是: {title}.')    await crawler.run(['https://crawlee.dev/'])if __name__ == '__main__':    asyncio.run(main())

import asynciofrom datetime import timedeltafrom crawlee.crawlers import PlaywrightCrawler, PlaywrightCrawlingContextasync def main() -> None:    crawler = PlaywrightCrawler(        headless=False,  # 设为 False 可以看到浏览器窗口，便于调试        max_requests_per_crawl=10,    )    @crawler.router.default_handler    async def request_handler(context: PlaywrightCrawlingContext) -> None:        context.log.info(f'正在处理 {context.request.url}')        # 等待页面中的特定元素加载完成        await context.page.wait_for_selector('h1', timeout=timedelta(seconds=5))        # 获取页面标题        title = await context.page.title()        context.log.info(f'页面标题: {title}')        # 获取页面内容        content = await context.page.content()        context.log.info(f'页面内容长度: {len(content)}')        # 查找页面中的所有链接并加入队列，实现深度爬取        await context.enqueue_links()    await crawler.run(['https://example.com'])if __name__ == '__main__':    asyncio.run(main())

import asynciofrom crawlee.crawlers import ParselCrawler, ParselCrawlingContextfrom crawlee.router import Routerrouter = Router[ParselCrawlingContext]()@router.default_handlerasync def default_handler(context: ParselCrawlingContext) -> None:    """处理普通页面的默认逻辑"""    context.log.info(f'默认处理器正在处理: {context.request.url}')    # 为特定模式的 URL 添加标签，使其被其他处理器处理    if '/product/' in context.request.url:        context.request.label = 'PRODUCT'    else:        # 提取所有产品页面的链接并加入队列        await context.enqueue_links(            selector='a.product-link',            label='PRODUCT'        )@router.handler('PRODUCT', label='PRODUCT')async def product_handler(context: ParselCrawlingContext) -> None:    """专门处理产品页面的逻辑"""    context.log.info(f'产品处理器正在处理: {context.request.url}')    # 使用 CSS 选择器提取产品数据    product_name = context.selector.css('h1.product-title::text').get()    product_price = context.selector.css('.price::text').get()    context.log.info(f'产品名称: {product_name}, 价格: {product_price}')    # 将结构化数据保存到数据集中    await context.push_data({        'url': context.request.url,        'name': product_name,        'price': product_price    })async def main() -> None:    crawler = ParselCrawler(        request_handler=router,        max_requests_per_crawl=20,    )    await crawler.run(['https://example-ecommerce-site.com'])if __name__ == '__main__':    asyncio.run(main())

import asynciofrom datetime import timedeltafrom crawlee.crawlers import PlaywrightCrawler, PlaywrightCrawlingContextfrom crawlee.sessions import SessionPoolasync def main() -> None:    # 配置会话池，确保登录状态在整个爬取过程中保持    session_pool = SessionPool(max_pool_size=1, session_rotation_enabled=False)    crawler = PlaywrightCrawler(        headless=False,  # 设为 True 可在后台运行        max_requests_per_crawl=10,        session_pool=session_pool,    )    @crawler.router.default_handler    async def request_handler(context: PlaywrightCrawlingContext) -> None:        # 检查是否已登录        if '/login' in context.request.url:            # 填写并提交登录表单            await context.page.fill('#username', 'your_username')            await context.page.fill('#password', 'your_password')            await context.page.click('button[type="submit"]')            await context.page.wait_for_navigation()            context.log.info('登录完成')            # 登录成功后，可以跳转到需要认证的页面            await context.add_requests(['https://example.com/dashboard'])        else:            # 处理需要登录后的页面逻辑            title = await context.page.title()            context.log.info(f'访问到内容页: {title}')            await context.enqueue_links()    await crawler.run(['https://example.com/login'])if __name__ == '__main__':    asyncio.run(main())

import asynciofrom crawlee.crawlers import PlaywrightCrawler, PlaywrightCrawlingContextasync def main() -> None:    crawler = PlaywrightCrawler(headless=False)    @crawler.router.default_handler    async def request_handler(context: PlaywrightCrawlingContext) -> None:        context.log.info(f'正在处理: {context.request.url}')        # 模拟滚动，加载更多内容        previous_height = await context.page.evaluate('document.body.scrollHeight')        while True:            # 滚动到底部            await context.page.evaluate('window.scrollTo(0, document.body.scrollHeight)')            await context.page.wait_for_timeout(2000)  # 等待新内容加载            # 检查是否有新内容加载            new_height = await context.page.evaluate('document.body.scrollHeight')            if new_height == previous_height:                break  # 没有新内容，停止滚动            previous_height = new_height        # 获取所有商品元素        items = await context.page.query_selector_all('.product-item')        context.log.info(f'共找到 {len(items)} 个商品')        for item in items:            # 提取每个商品的名称和价格            name = await item.query_selector('.product-name')            price = await item.query_selector('.product-price')            if name and price:                context.log.info(f'商品: {await name.inner_text()}, 价格: {await price.inner_text()}')    await crawler.run(['https://example.com/products'])if __name__ == '__main__':    asyncio.run(main())

import asynciofrom crawlee.crawlers import BeautifulSoupCrawler, BeautifulSoupCrawlingContextfrom crawlee.storages import Datasetasync def main() -> None:    # 创建或打开一个数据集    dataset = await Dataset.open()    crawler = BeautifulSoupCrawler()    @crawler.router.default_handler    async def request_handler(context: BeautifulSoupCrawlingContext) -> None:        url = context.request.url        title = context.soup.title.string if context.soup.title else ''        # 将数据推送到数据集        await dataset.push_data({            'url': url,            'title': title,            'scraped_at': context.request.loaded_at.isoformat() if context.request.loaded_at else None        })        context.log.info(f'已保存: {url}')        # 继续爬取新链接        await context.enqueue_links()    await crawler.run(['https://crawlee.dev/'])    # 爬取完成后，可以获取数据集的条目数量    data = await dataset.get_data()    context.log.info(f'共抓取并保存了 {len(data.items)} 条数据')if __name__ == '__main__':    asyncio.run(main())

import asynciofrom datetime import timedeltafrom crawlee.crawlers import AdaptivePlaywrightCrawler, AdaptivePlaywrightCrawlingContextasync def main() -> None:    # 使用 BeautifulSoup 作为静态解析器    crawler = AdaptivePlaywrightCrawler.with_beautifulsoup_static_parser(        max_requests_per_crawl=5,        playwright_crawler_specific_kwargs={'browser_type': 'chromium'},    )    @crawler.router.default_handler    async def request_handler(context: AdaptivePlaywrightCrawlingContext) -> None:        context.log.info(f'正在处理: {context.request.url}')        # 在静态模式下，可以直接使用 parsed_content 属性        if context.is_static:            title = context.parsed_content.title.string if context.parsed_content.title else ''            context.log.info(f'[静态模式] 页面标题: {title}')        else:            # 在浏览器模式下，可以使用页面交互功能            title = await context.page.title()            context.log.info(f'[浏览器模式] 页面标题: {title}')            # 定位特定元素            element = await context.query_selector_one('h2', timeout=timedelta(seconds=5))            if element:                text = await element.text_content()                context.log.info(f'找到元素: {text}')        # 发现并加入新的链接        await context.enqueue_links()    await crawler.run(['https://crawlee.dev/'])if __name__ == '__main__':    asyncio.run(main())