web_url_crawl 函数使用说明

概述

web_url_crawl 是一个用于批量爬取网页内容的Python函数，位于 app/core/data_parse/parse_task.py 文件中。该函数基于原有的JavaScript版本（docs/server.js）重写，提供了更强大的功能和更好的错误处理。

功能特性

批量处理: 支持同时处理多个URL
智能重试: 自动重试失败的请求，最多重试3次
内容解析: 使用BeautifulSoup自动解析HTML，提取纯文本内容
反爬虫保护: 模拟真实浏览器请求头，添加随机延迟
详细日志: 提供完整的处理过程日志
错误处理: 完善的异常处理和错误信息记录

函数签名

def web_url_crawl(urls):
    """
    从指定URL数组读取网页内容，格式化后返回
    
    Args:
        urls (list): 字符串数组，每个元素为一个网页URL地址
        
    Returns:
        dict: 包含爬取结果的字典
    """

输入参数

urls (list): 字符串数组，包含要爬取的网页URL地址

返回值

函数返回一个字典，包含以下字段：

{
    'success': True/False,           # 是否成功爬取到内容
    'message': '处理结果描述',        # 处理结果的文字描述
    'data': {
        'total_urls': 0,             # 总URL数量
        'success_count': 0,          # 成功爬取的URL数量
        'failed_count': 0,           # 失败的URL数量
        'contents': [                # 成功爬取的内容列表
            {
                'url': 'URL地址',
                'data': '网页内容',
                'status': 'success',
                'content_length': 内容长度,
                'original_length': 原始内容长度,
                'status_code': HTTP状态码,
                'encoding': 编码格式
            }
        ],
        'failed_items': [            # 失败的URL列表
            {
                'url': 'URL地址',
                'error': '错误信息',
                'status': 'failed'
            }
        ]
    }
}

使用示例

基本用法

from app.core.data_parse.parse_task import web_url_crawl

# 准备URL列表
urls = [
    "https://example.com/page1",
    "https://example.com/page2",
    "https://example.com/page3"
]

# 调用函数
result = web_url_crawl(urls)

# 检查结果
if result['success']:
    print(f"成功爬取 {result['data']['success_count']} 个网页")
    for content in result['data']['contents']:
        print(f"URL: {content['url']}")
        print(f"内容长度: {content['content_length']}")
        print(f"内容预览: {content['data'][:100]}...")
else:
    print(f"爬取失败: {result['message']}")

错误处理

result = web_url_crawl(urls)

# 处理部分成功的情况
if result['data']['success_count'] > 0:
    print(f"部分成功: {result['data']['success_count']} 个成功，{result['data']['failed_count']} 个失败")
    
    # 处理成功的内容
    for content in result['data']['contents']:
        process_content(content['data'])
    
    # 处理失败的项目
    for failed in result['data']['failed_items']:
        print(f"失败URL: {failed['url']}, 错误: {failed['error']}")
else:
    print("所有URL都爬取失败")

配置参数

函数内部包含以下可配置参数：

超时时间: 30秒
最大重试次数: 3次
请求延迟: 0.5-2.0秒随机延迟
User-Agent: 模拟Chrome浏览器
请求头: 包含完整的浏览器标识信息

依赖要求

确保安装以下Python包：

pip install requests beautifulsoup4

或者在 requirements.txt 中添加：

requests>=2.32.3
beautifulsoup4>=4.12.0

注意事项

反爬虫机制: 函数已包含基本的反爬虫保护，但对于某些网站可能需要额外的处理
网络稳定性: 建议在网络稳定的环境下使用
内容大小: 对于大型网页，内容可能很长，注意内存使用
法律合规: 请确保遵守目标网站的robots.txt和使用条款
频率限制: 函数已包含延迟机制，避免过于频繁的请求

测试

可以使用提供的测试脚本验证函数功能：

python test_web_crawl.py

测试脚本会使用一些测试URL来验证函数的各种功能，包括成功爬取、错误处理等。

与JavaScript版本的对比

特性	JavaScript版本	Python版本
并发处理	Promise.all并行处理	顺序处理，带延迟
错误处理	基本的错误捕获	详细的错误分类和重试
内容解析	返回原始HTML	自动解析为纯文本
日志记录	控制台输出	结构化日志记录
重试机制	无	智能重试机制
反爬虫保护	基本请求头	完整的浏览器模拟

扩展建议

并发处理: 可以考虑使用 asyncio 或 concurrent.futures 实现真正的并发爬取
代理支持: 可以添加代理服务器支持，避免IP被封
内容过滤: 可以添加内容过滤规则，只保留特定类型的内容
存储支持: 可以集成数据库存储，保存爬取结果
监控告警: 可以添加爬取状态监控和异常告警功能

web_crawl_usage.md 5.2 KB Histórico Raw