南京电商网站建设去哪儿旅行app下载安装

南京电商网站建设,去哪儿旅行app下载安装,郑州航空港建设局网站,网站栏目设计怎么写前言 在 Python 爬虫开发领域#xff0c;零散的代码片段仅能满足简单的爬取需求#xff0c;面对多站点、大批量、高稳定性的爬取场景时#xff0c;易出现 “代码冗余、维护困难、扩展性差” 等问题。基础爬虫框架的设计是解决上述问题的核心 —— 通过模块化、分层化的架构…前言在 Python 爬虫开发领域零散的代码片段仅能满足简单的爬取需求面对多站点、大批量、高稳定性的爬取场景时易出现 “代码冗余、维护困难、扩展性差” 等问题。基础爬虫框架的设计是解决上述问题的核心 —— 通过模块化、分层化的架构设计将爬虫拆分为可复用的核心组件既能提升开发效率又能保障爬虫的稳定性和可维护性。本文将从框架设计原则、核心组件拆解、实战开发等维度系统讲解基础爬虫框架的设计思路与实现方法帮助开发者构建标准化、可扩展的爬虫体系。摘要本文聚焦基础爬虫框架的结构设计从框架设计核心原则出发拆解请求模块、解析模块、存储模块、调度模块、异常处理模块等核心组件并通过完整的实战案例框架实战链接多站点爬取测试平台、框架性能测试页演示基础爬虫框架的开发与应用。读者可通过本文掌握模块化爬虫框架的设计思路实现从 “零散代码” 到 “标准化框架” 的进阶提升爬虫开发的效率和可维护性。一、爬虫框架设计核心原则1.1 框架设计的核心目标目标维度具体说明模块化将爬虫拆分为独立组件请求、解析、存储等组件间低耦合、高内聚可复用核心组件如请求封装、异常处理可跨项目复用减少重复开发可扩展支持新增爬取站点、解析规则、存储方式时无需修改核心代码可维护代码结构清晰日志完善便于问题定位和功能迭代稳定性具备重试、限流、异常捕获机制保障爬虫持续稳定运行1.2 框架设计的核心原则原则名称具体要求单一职责每个组件仅负责一个核心功能如请求模块仅处理 HTTP 请求解析模块仅处理数据提取开闭原则框架对扩展开放新增功能对修改关闭核心代码无需改动接口隔离不同组件通过清晰的接口交互避免组件间依赖过度异常隔离单个组件的异常不影响整个框架运行异常被限定在组件内部处理配置解耦爬取规则、请求参数、存储路径等配置与代码分离便于灵活调整二、基础爬虫框架核心组件拆解2.1 框架整体架构分层设计plaintext基础爬虫框架架构 ├── 配置层Config管理所有配置项请求头、爬取规则、存储路径等 ├── 核心层Core框架核心组件 │ ├── 请求模块Request封装HTTP请求处理重试、限流、反爬 │ ├── 解析模块Parser封装数据解析逻辑支持多规则解析 │ ├── 存储模块Storage封装数据存储逻辑支持多格式存储 │ ├── 调度模块Scheduler管理爬取任务控制爬取流程 │ └── 异常处理模块Exception统一处理各类异常记录日志 ├── 业务层Business具体爬取业务实现基于核心层扩展 └── 工具层Utils通用工具函数日志、编码处理、数据校验等2.2 核心组件功能详解组件名称核心功能技术选型关键特性配置模块加载 / 管理配置YAML/JSON/ 环境变量PyYAML/python-dotenv配置热更新、默认值兜底请求模块发送 HTTP/HTTPS 请求处理重试 / 超时 / 反爬requests/aiohttp会话复用、请求限流、代理支持解析模块提取目标数据HTML/JSON/XMLBeautifulSoup/jsonpath/lxml多解析规则支持、字段校验存储模块数据持久化文件 / 数据库pandas/sqlalchemy批量存储、去重、事务支持调度模块任务队列管理、爬取流程控制queue/threading单线程 / 多线程调度、优先级控制异常处理模块统一异常捕获、分类处理、日志记录logging异常分级、重试触发、告警提示工具模块通用工具日志、编码、数据校验logging/chardet/pydantic功能复用、标准化输出三、基础爬虫框架实战开发3.1 开发环境准备bash运行# 安装核心依赖 pip install requests PyYAML beautifulsoup4 jsonpath-python pandas sqlalchemy chardet pydantic3.2 框架目录结构标准化plaintextspider_framework/ ├── config/ # 配置目录 │ └── spider_config.yaml # 爬虫配置文件 ├── core/ # 核心组件目录 │ ├── __init__.py │ ├── request_handler.py # 请求模块 │ ├── parse_handler.py # 解析模块 │ ├── storage_handler.py # 存储模块 │ ├── scheduler.py # 调度模块 │ └── exception_handler.py # 异常处理模块 ├── utils/ # 工具目录 │ ├── __init__.py │ ├── log_utils.py # 日志工具 │ └── data_utils.py # 数据工具 ├── business/ # 业务实现目录 │ ├── __init__.py │ └── demo_spider.py # 示例爬虫 └── main.py # 框架入口3.3 核心组件实现3.3.1 工具层日志工具utils/log_utils.pypython运行import logging import os from datetime import datetime def setup_logger(namespider_framework, log_levellogging.DEBUG): 初始化框架日志配置 :param name: 日志器名称 :param log_level: 日志级别 :return: 配置好的logger对象 # 创建日志目录 log_dir logs if not os.path.exists(log_dir): os.makedirs(log_dir) # 日志文件名 log_file f{log_dir}/spider_{datetime.now().strftime(%Y%m%d_%H%M%S)}.log # 日志格式 formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s ) # 初始化logger logger logging.getLogger(name) logger.setLevel(log_level) logger.handlers.clear() # 清空原有处理器 # 控制台处理器 console_handler logging.StreamHandler() console_handler.setFormatter(formatter) logger.addHandler(console_handler) # 文件处理器 file_handler logging.FileHandler(log_file, encodingutf-8) file_handler.setFormatter(formatter) logger.addHandler(file_handler) return logger # 全局日志对象 logger setup_logger()输出结果初始化日志plaintext2025-12-17 15:00:00,000 - spider_framework - INFO - log_utils.py:35 - 日志初始化完成日志文件路径logs/spider_20251217_150000.log原理说明封装日志初始化逻辑支持控制台 文件双输出便于调试和问题追溯按时间戳命名日志文件避免日志覆盖全局 logger 对象供框架所有组件复用保证日志格式统一。3.3.2 配置层配置管理config/spider_config.yaml core/init.py配置文件config/spider_config.yamlyaml# 请求配置 request: headers: User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Referer: https://www.example.com timeout: 10 # 请求超时时间秒 retry_times: 3 # 重试次数 retry_interval: 2 # 重试间隔秒 rate_limit: 1 # 请求频率限制秒/次 # 解析配置 parse: # 示例解析规则JSON接口 json_rules: demo_api: fields: id: $.id title: $.title content: $.body # 示例解析规则HTML页面 html_rules: demo_html: item_selector: div.item fields: name: span.name::text price: span.price::text # 存储配置 storage: default_format: csv # 默认存储格式csv/excel/sqlite save_path: data # 数据保存目录 sqlite_db: data/spider_data.db # SQLite数据库路径配置加载core/init.pypython运行import yaml import os from utils.log_utils import logger # 加载配置文件 def load_config(config_pathconfig/spider_config.yaml): 加载YAML配置文件 :param config_path: 配置文件路径 :return: 配置字典 try: if not os.path.exists(config_path): raise FileNotFoundError(f配置文件不存在{config_path}) with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) logger.info(f✅ 配置文件加载成功{config_path}) return config except Exception as e: logger.error(f❌ 配置文件加载失败{e}, exc_infoTrue) return {} # 全局配置对象 CONFIG load_config()输出结果plaintext2025-12-17 15:01:00,000 - spider_framework - INFO - __init__.py:20 - ✅ 配置文件加载成功config/spider_config.yaml原理说明使用 YAML 文件管理所有配置实现配置与代码解耦封装配置加载逻辑异常捕获并记录日志保证框架鲁棒性全局 CONFIG 对象供所有核心组件调用统一配置入口。3.3.3 核心层请求模块core/request_handler.pypython运行import requests import time from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry from core import CONFIG from utils.log_utils import logger class RequestHandler: 请求处理模块封装HTTP请求处理重试、限流、反爬 def __init__(self): # 从配置加载参数 self.timeout CONFIG.get(request, {}).get(timeout, 10) self.retry_times CONFIG.get(request, {}).get(retry_times, 3) self.retry_interval CONFIG.get(request, {}).get(retry_interval, 2) self.rate_limit CONFIG.get(request, {}).get(rate_limit, 1) self.headers CONFIG.get(request, {}).get(headers, {}) # 初始化session复用连接 self.session self._create_session() # 记录最后一次请求时间用于限流 self.last_request_time 0 def _create_session(self): 创建带重试策略的session session requests.Session() # 配置重试策略 retry_strategy Retry( totalself.retry_times, backoff_factorself.retry_interval, status_forcelist[429, 500, 502, 503, 504] ) # 挂载适配器 adapter HTTPAdapter(max_retriesretry_strategy) session.mount(https://, adapter) session.mount(http://, adapter) # 设置默认请求头 session.headers.update(self.headers) return session def _rate_limit_control(self): 请求频率控制 current_time time.time() interval current_time - self.last_request_time if interval self.rate_limit: sleep_time self.rate_limit - interval logger.debug(f⏳ 请求频率限制等待{sleep_time:.2f}秒) time.sleep(sleep_time) self.last_request_time time.time() def get(self, url, paramsNone, **kwargs): 发送GET请求封装限流、重试、异常处理 try: # 频率控制 self._rate_limit_control() # 发送请求 logger.debug(f 发送GET请求{url}参数{params}) response self.session.get( url, paramsparams, timeoutself.timeout, **kwargs ) # 校验状态码 response.raise_for_status() # 设置编码 response.encoding response.apparent_encoding logger.info(f✅ GET请求成功{url}状态码{response.status_code}) return response except Exception as e: logger.error(f❌ GET请求失败{url}异常{e}, exc_infoTrue) return None def post(self, url, dataNone, jsonNone, **kwargs): 发送POST请求封装限流、重试、异常处理 try: # 频率控制 self._rate_limit_control() # 发送请求 logger.debug(f 发送POST请求{url}数据{data}JSON{json}) response self.session.post( url, datadata, jsonjson, timeoutself.timeout, **kwargs ) # 校验状态码 response.raise_for_status() # 设置编码 response.encoding response.apparent_encoding logger.info(f✅ POST请求成功{url}状态码{response.status_code}) return response except Exception as e: logger.error(f❌ POST请求失败{url}异常{e}, exc_infoTrue) return None def close(self): 关闭session self.session.close() logger.info( 请求session已关闭) # 全局请求处理器对象 request_handler RequestHandler()使用示例 输出结果python运行# 测试GET请求 response request_handler.get(https://jsonplaceholder.typicode.com/posts/1) print(f响应状态码{response.status_code}) print(f响应内容{response.json()})plaintext2025-12-17 15:02:00,000 - spider_framework - DEBUG - request_handler.py:78 - 发送GET请求https://jsonplaceholder.typicode.com/posts/1参数None 2025-12-17 15:02:01,000 - spider_framework - INFO - request_handler.py:92 - ✅ GET请求成功https://jsonplaceholder.typicode.com/posts/1状态码200 响应状态码200 响应内容{userId: 1, id: 1, title: sunt aut facere repellat provident occaecati excepturi optio reprehenderit, body: quia et suscipit\nsuscipit recusandae consequuntur expedita et cum\nreprehenderit molestiae ut ut quas totam\nnostrum rerum est autem sunt rem eveniet architecto}原理说明封装 GET/POST 请求集成重试、限流、编码处理等核心功能使用 Session 复用 TCP 连接提升请求效率基于配置文件动态调整请求参数无需修改代码完善的日志记录和异常处理便于问题定位。3.3.4 核心层解析模块core/parse_handler.pypython运行from bs4 import BeautifulSoup from jsonpath import jsonpath from core import CONFIG from utils.log_utils import logger class ParseHandler: 解析处理模块支持JSON/HTML数据解析 def __init__(self): # 从配置加载解析规则 self.json_rules CONFIG.get(parse, {}).get(json_rules, {}) self.html_rules CONFIG.get(parse, {}).get(html_rules, {}) def parse_json(self, json_data, rule_name): 解析JSON数据基于配置的解析规则 :param json_data: 原始JSON数据dict/list :param rule_name: 解析规则名称对应配置中的json_rules :return: 解析后的数据列表 if not json_data or rule_name not in self.json_rules: logger.warning(f⚠️ JSON解析规则不存在或数据为空{rule_name}) return [] rule self.json_rules[rule_name] fields rule.get(fields, {}) result [] # 处理列表型JSON if isinstance(json_data, list): for item in json_data: parsed_item self._parse_single_json(item, fields) if parsed_item: result.append(parsed_item) # 处理字典型JSON elif isinstance(json_data, dict): parsed_item self._parse_single_json(json_data, fields) if parsed_item: result.append(parsed_item) logger.info(f✅ JSON解析完成{rule_name}解析出{len(result)}条数据) return result def _parse_single_json(self, json_item, fields): 解析单个JSON对象 parsed_item {} for field_name, jsonpath_expr in fields.items(): # 使用jsonpath提取字段支持嵌套 values jsonpath(json_item, jsonpath_expr) if values: # 处理列表结果取第一个值 parsed_item[field_name] values[0] if isinstance(values, list) else values else: parsed_item[field_name] logger.debug(f⚠️ 字段提取失败{field_name}表达式{jsonpath_expr}) return parsed_item def parse_html(self, html_content, rule_name): 解析HTML数据基于配置的解析规则 :param html_content: 原始HTML字符串 :param rule_name: 解析规则名称对应配置中的html_rules :return: 解析后的数据列表 if not html_content or rule_name not in self.html_rules: logger.warning(f⚠️ HTML解析规则不存在或数据为空{rule_name}) return [] rule self.html_rules[rule_name] item_selector rule.get(item_selector) fields rule.get(fields, {}) # 解析HTML soup BeautifulSoup(html_content, lxml) items soup.select(item_selector) result [] for item in items: parsed_item {} for field_name, css_expr in fields.items(): try: # 支持CSS选择器文本提取如span.name::text if ::text in css_expr: css_expr css_expr.replace(::text, ) value item.select_one(css_expr).get_text(stripTrue) if item.select_one(css_expr) else else: value item.select_one(css_expr).get(href, ) if item.select_one(css_expr) else parsed_item[field_name] value except Exception as e: parsed_item[field_name] logger.debug(f⚠️ HTML字段提取失败{field_name}异常{e}) result.append(parsed_item) logger.info(f✅ HTML解析完成{rule_name}解析出{len(result)}条数据) return result # 全局解析处理器对象 parse_handler ParseHandler()使用示例 输出结果python运行# 测试JSON解析 json_data {id: 1, title: 测试标题, body: 测试内容} parsed_json parse_handler.parse_json(json_data, demo_api) print(fJSON解析结果{parsed_json}) # 测试HTML解析 html_content div classitem span classname测试商品/span span classprice99.9元/span /div parsed_html parse_handler.parse_html(html_content, demo_html) print(fHTML解析结果{parsed_html})plaintext2025-12-17 15:03:00,000 - spider_framework - INFO - parse_handler.py:52 - ✅ JSON解析完成demo_api解析出1条数据 JSON解析结果[{id: 1, title: 测试标题, content: 测试内容}] 2025-12-17 15:03:00,000 - spider_framework - INFO - parse_handler.py:95 - ✅ HTML解析完成demo_html解析出1条数据 HTML解析结果[{name: 测试商品, price: 99.9元}]原理说明支持 JSON/HTML 两种主流数据格式解析基于配置规则提取字段JSON 解析使用 jsonpath 支持嵌套字段提取HTML 解析使用 BeautifulSoupCSS 选择器字段提取失败时设置空值并记录日志避免解析流程中断解析规则与代码分离新增解析规则仅需修改配置文件。3.3.5 核心层存储模块core/storage_handler.pypython运行import os import pandas as pd from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData from core import CONFIG from utils.log_utils import logger class StorageHandler: 存储处理模块支持CSV/Excel/SQLite存储 def __init__(self): # 从配置加载存储参数 self.default_format CONFIG.get(storage, {}).get(default_format, csv) self.save_path CONFIG.get(storage, {}).get(save_path, data) self.sqlite_db CONFIG.get(storage, {}).get(sqlite_db, data/spider_data.db) # 创建数据保存目录 if not os.path.exists(self.save_path): os.makedirs(self.save_path) logger.info(f 创建数据保存目录{self.save_path}) # 初始化SQLite引擎按需 self.sqlite_engine None def _init_sqlite(self): 初始化SQLite引擎 if not self.sqlite_engine: self.sqlite_engine create_engine(fsqlite:///{self.sqlite_db}) logger.info(f️ SQLite引擎初始化完成{self.sqlite_db}) def save_to_csv(self, data, filename, headerTrue): 保存数据到CSV文件 :param data: 解析后的数据列表dict/list :param filename: 文件名不含后缀 :param header: 是否写入表头 if not data: logger.warning(⚠️ 无数据可保存到CSV) return False try: file_path os.path.join(self.save_path, f{filename}.csv) df pd.DataFrame(data) # 追加/写入模式 mode a if os.path.exists(file_path) and not header else w df.to_csv( file_path, modemode, headerheader, indexFalse, encodingutf-8-sig ) logger.info(f✅ 数据保存到CSV成功{file_path}数据量{len(data)}) return True except Exception as e: logger.error(f❌ 保存CSV失败{e}, exc_infoTrue) return False def save_to_excel(self, data, filename, headerTrue): 保存数据到Excel文件 if not data: logger.warning(⚠️ 无数据可保存到Excel) return False try: file_path os.path.join(self.save_path, f{filename}.xlsx) df pd.DataFrame(data) with pd.ExcelWriter(file_path, modea if os.path.exists(file_path) and not header else w) as writer: df.to_excel(writer, sheet_namedata, indexFalse, headerheader) logger.info(f✅ 数据保存到Excel成功{file_path}数据量{len(data)}) return True except Exception as e: logger.error(f❌ 保存Excel失败{e}, exc_infoTrue) return False def save_to_sqlite(self, data, table_name): 保存数据到SQLite数据库 if not data: logger.warning(⚠️ 无数据可保存到SQLite) return False try: # 初始化SQLite引擎 self._init_sqlite() # 转换为DataFrame并写入数据库 df pd.DataFrame(data) df.to_sql( table_name, self.sqlite_engine, if_existsappend, indexFalse ) logger.info(f✅ 数据保存到SQLite成功{table_name}数据量{len(data)}) return True except Exception as e: logger.error(f❌ 保存SQLite失败{e}, exc_infoTrue) return False def save(self, data, filename, format_typeNone, table_namespider_data): 统一保存接口自动选择存储方式 :param data: 解析后的数据列表 :param filename: 文件名不含后缀 :param format_type: 存储格式csv/excel/sqlite默认使用配置中的default_format :param table_name: SQLite表名 format_type format_type or self.default_format if format_type csv: return self.save_to_csv(data, filename) elif format_type excel: return self.save_to_excel(data, filename) elif format_type sqlite: return self.save_to_sqlite(data, table_name) else: logger.error(f❌ 不支持的存储格式{format_type}) return False # 全局存储处理器对象 storage_handler StorageHandler()使用示例 输出结果python运行# 测试CSV存储 test_data [{id: 1, title: 测试1}, {id: 2, title: 测试2}] storage_handler.save(test_data, demo_data, format_typecsv)plaintext2025-12-17 15:04:00,000 - spider_framework - INFO - storage_handler.py:35 - 创建数据保存目录data 2025-12-17 15:04:00,000 - spider_framework - INFO - storage_handler.py:60 - ✅ 数据保存到CSV成功data/demo_data.csv数据量2原理说明封装 CSV/Excel/SQLite 三种主流存储方式提供统一保存接口支持追加写入避免重复生成文件CSV 存储使用utf-8-sig编码解决中文乱码问题延迟初始化 SQLite 引擎减少资源占用。3.3.6 核心层调度模块core/scheduler.pypython运行from core import request_handler, parse_handler, storage_handler from utils.log_utils import logger class Scheduler: 调度模块管理爬取任务全流程 def __init__(self): self.task_queue [] # 爬取任务队列 def add_task(self, task): 添加爬取任务 :param task: 任务字典格式 { task_name: 任务名称, url: 目标URL, request_type: get/post, parse_type: json/html, parse_rule: 解析规则名称, storage_format: csv/excel/sqlite, storage_filename: 存储文件名 } required_fields [task_name, url, request_type, parse_type, parse_rule, storage_filename] for field in required_fields: if field not in task: logger.error(f❌ 任务缺少必填字段{field}) return False self.task_queue.append(task) logger.info(f✅ 添加爬取任务成功{task[task_name]}当前任务队列长度{len(self.task_queue)}) return True def run_task(self, task): 执行单个爬取任务 try: logger.info(f 开始执行任务{task[task_name]}) # 1. 发送请求 if task[request_type] get: response request_handler.get(task[url]) elif task[request_type] post: response request_handler.post(task[url], jsontask.get(post_data)) else: logger.error(f❌ 不支持的请求类型{task[request_type]}) return False if not response: return False # 2. 解析数据 if task[parse_type] json: raw_data response.json() parsed_data parse_handler.parse_json(raw_data, task[parse_rule]) elif task[parse_type] html: raw_data response.text parsed_data parse_handler.parse_html(raw_data, task[parse_rule]) else: logger.error(f❌ 不支持的解析类型{task[parse_type]}) return False if not parsed_data: logger.warning(f⚠️ 任务解析无数据{task[task_name]}) return False # 3. 存储数据 storage_result storage_handler.save( parsed_data, task[storage_filename], format_typetask.get(storage_format) ) if storage_result: logger.info(f✅ 任务执行成功{task[task_name]}) return True else: logger.error(f❌ 任务存储失败{task[task_name]}) return False except Exception as e: logger.error(f❌ 任务执行异常{task[task_name]}异常{e}, exc_infoTrue) return False def run_all_tasks(self): 执行所有任务 logger.info(f 开始执行所有爬取任务任务总数{len(self.task_queue)}) success_count 0 fail_count 0 for task in self.task_queue: if self.run_task(task): success_count 1 else: fail_count 1 # 清空任务队列 self.task_queue.clear() logger.info(f 所有任务执行完成成功{success_count}个失败{fail_count}个) return { success: success_count, fail: fail_count } # 全局调度器对象 scheduler Scheduler()原理说明调度模块是框架的核心入口串联 “请求→解析→存储” 全流程标准化任务格式支持批量添加和执行任务任务执行结果统计便于监控爬取效率单个任务异常不影响其他任务执行保证框架稳定性。3.4 业务层示例爬虫business/demo_spider.pypython运行from core.scheduler import scheduler from utils.log_utils import logger def run_demo_spider(): 运行示例爬虫 # 定义爬取任务 demo_task { task_name: JSONPlaceholder帖子爬取, url: https://jsonplaceholder.typicode.com/posts, request_type: get, parse_type: json, parse_rule: demo_api, storage_format: csv, storage_filename: jsonplaceholder_posts } # 添加任务 scheduler.add_task(demo_task) # 执行所有任务 result scheduler.run_all_tasks() logger.info(f 示例爬虫执行结果{result}) return result if __name__ __main__: run_demo_spider()3.5 框架入口main.pypython运行from business.demo_spider import run_demo_spider from utils.log_utils import logger from core.request_handler import request_handler def main(): 框架主入口 logger.info( 基础爬虫框架启动) try: # 运行示例爬虫 run_demo_spider() finally: # 关闭请求session request_handler.close() logger.info( 基础爬虫框架退出) if __name__ __main__: main()完整运行输出结果plaintext2025-12-17 15:05:00,000 - spider_framework - INFO - main.py:8 - 基础爬虫框架启动 2025-12-17 15:05:00,000 - spider_framework - INFO - scheduler.py:45 - ✅ 添加爬取任务成功JSONPlaceholder帖子爬取当前任务队列长度1 2025-12-17 15:05:00,000 - spider_framework - INFO - scheduler.py:89 - 开始执行所有爬取任务任务总数1 2025-12-17 15:05:00,000 - spider_framework - INFO - scheduler.py:59 - 开始执行任务JSONPlaceholder帖子爬取 2025-12-17 15:05:00,000 - spider_framework - DEBUG - request_handler.py:78 - 发送GET请求https://jsonplaceholder.typicode.com/posts参数None 2025-12-17 15:05:01,000 - spider_framework - INFO - request_handler.py:92 - ✅ GET请求成功https://jsonplaceholder.typicode.com/posts状态码200 2025-12-17 15:05:01,000 - spider_framework - INFO - parse_handler.py:52 - ✅ JSON解析完成demo_api解析出100条数据 2025-12-17 15:05:01,000 - spider_framework - INFO - storage_handler.py:60 - ✅ 数据保存到CSV成功data/jsonplaceholder_posts.csv数据量100 2025-12-17 15:05:01,000 - spider_framework - INFO - scheduler.py:80 - ✅ 任务执行成功JSONPlaceholder帖子爬取 2025-12-17 15:05:01,000 - spider_framework - INFO - scheduler.py:97 - 所有任务执行完成成功1个失败0个 2025-12-17 15:05:01,000 - spider_framework - INFO - demo_spider.py:22 - 示例爬虫执行结果{success: 1, fail: 0} 2025-12-17 15:05:01,000 - spider_framework - INFO - request_handler.py:115 - 请求session已关闭 2025-12-17 15:05:01,000 - spider_framework - INFO - main.py:15 - 基础爬虫框架退出四、框架扩展与优化4.1 框架扩展方向扩展维度具体方案技术选型多线程 / 异步爬取扩展调度模块支持多线程 / 异步任务执行threading/aiohttp/asyncio代理池集成扩展请求模块支持自动切换代理 IPrequests-proxies / 自建代理池反爬对抗扩展请求模块集成 Cookie 池、指纹伪装faker/undetected-chromedriver数据去重扩展存储模块支持基于唯一键去重redis/pandas 去重进度监控扩展调度模块添加进度条和实时统计tqdm/prometheus4.2 框架优化建议配置热更新使用 watchdog 监控配置文件变化无需重启框架即可更新配置连接池优化调整 requests.Session 的连接池大小提升大批量请求效率内存优化大批量数据存储时采用分批写入策略避免内存溢出日志轮转使用 logging.handlers.RotatingFileHandler 实现日志文件轮转避免日志文件过大单元测试为核心组件编写单元测试保证框架稳定性pytest。4.3 框架使用规范新增爬取业务仅需在配置文件添加解析规则在业务层定义任务字典无需修改核心代码异常处理规范所有自定义异常需继承框架基础异常类统一异常格式日志规范按级别记录日志DEBUG调试信息INFO正常流程WARNING警告ERROR错误CRITICAL严重错误配置规范新增配置项需在 YAML 文件中添加注释标明用途和默认值。五、总结基础爬虫框架的核心价值在于 “标准化、可复用、可扩展”—— 通过分层化、模块化的架构设计将爬虫开发拆分为配置管理、请求处理、数据解析、数据存储、任务调度等独立组件既降低了开发门槛又提升了代码的可维护性。本文设计的基础爬虫框架覆盖了爬虫开发的核心场景支持 JSON/HTML 解析、多格式存储、重试 / 限流 / 异常处理等核心功能可作为各类爬虫项目的基础模板。在实际应用中需根据业务需求对框架进行扩展简单爬取场景可直接使用核心组件大批量 / 高反爬场景可扩展多线程、代理池、反爬对抗等功能。同时需遵循框架使用规范保证代码的一致性和可维护性。掌握基础爬虫框架的设计思路是从 “初级爬虫开发者” 向 “高级爬虫工程师” 进阶的关键。扩展建议进阶框架学习研究 Scrapy 框架的核心架构吸收其组件设计思想分布式扩展基于 celery/redis 实现分布式爬虫框架提升爬取效率可视化管理结合 FastAPI/Streamlit 开发爬虫管理界面支持任务配置、进度监控、数据查看合规性优化添加 robots.txt 校验、请求频率限制保证爬虫合规性。