Python Selenium并发多页爬虫重复代码多难维护,该如何重构?
核心思路:把所有重复的流程代码上提为公共基类,差异化逻辑下沉到各站点的配置类,彻底打破“拷文件改代码”的开发模式
1. 抽离全站点通用的爬虫流程基类
现在三个站点的scraper.py里80%的代码是完全重复的:初始化driver、注册进程退出时的driver关闭钩子、走“打开入口页→收集链接写对应json→读json匹配关键词→打开详情页→提取数值”的固定流程。这些通用逻辑完全可以抽成BaseSiteScraper基类,放在公共目录下,不需要每个站点重复写。
基类只留需要站点自定义的配置项作为可覆写类属性,所有固定流程全部在基类里实现,示例代码:
# core/base_scraper.py import atexit from driver import set_driver from utils.io import read_links_from_json, write_links_to_json class BaseSiteScraper: # 以下为站点专属配置,子类仅需填充这些值,不需要重写流程 site_entry_url: str site_data_file_path: str link_collector_page_cls: type detail_page_cls: type def __init__(self): # 通用初始化逻辑,所有子类完全复用 self.driver = set_driver() self.site_links = {} atexit.register(self._close_driver) def setup(self): # 通用的链接收集流程 self.driver.get(self.site_entry_url) collector = self.link_collector_page_cls(self.driver) collected = collector.get_collected_links() write_links_to_json(collected, self.site_data_file_path) def get_target_value(self, match_keyword: str): # 通用的详情页数值提取流程 self.site_links = read_links_from_json(self.site_data_file_path) target_link = next( (v for k, v in self.site_links.items() if match_keyword in k), None ) if not target_link: return "Not found" self.driver.get(target_link) detail_page = self.detail_page_cls(self.driver) result = detail_page.get_target_number() return result if result else "Not found" def _close_driver(self): # 通用的driver退出逻辑 self.driver.quit()
2. 把Page层的通用能力全部收敛到已有的BasePage基类
现在各站点page.py里的Page类都直接继承公共的BasePage,之前你要加通用能力需要每个站点单独写的问题,本质是通用逻辑没有放在BasePage里。之后不管是加“元素等待重试”“自动关闭弹窗”“页面滚动”这类通用交互能力,只需要在BasePage里实现一次,所有站点的所有Page类自动继承获得对应能力,不需要逐站点修改调试。
比如要加“自动捕获并关闭页面弹窗”的能力,只需要在BasePage的初始化方法里加对应的检测逻辑,所有站点的Page类在初始化时就会自动执行,零额外维护成本。
3. 把main.py的硬编码爬虫列表换成自动发现机制
现在main.py里手动import每个站点的Scraper类,新增站点就要改一次import代码,完全可以实现简单的模块自动扫描逻辑:按照固定命名规则(比如所有以scraper_开头的目录),自动加载目录下继承了BaseSiteScraper的子类,加入调度列表,示例代码:
# main.py from pathlib import Path from importlib import import_module from concurrent.futures import ThreadPoolExecutor from core.base_scraper import BaseSiteScraper def load_all_scrapers(): scrapers = [] # 扫描所有符合命名规则的爬虫目录 for scraper_dir in Path(".").glob("scraper_*"): module_name = scraper_dir.name scraper_module = import_module(f"{module_name}.scraper") # 找模块下所有继承BaseSiteScraper的具体实现类 for _, cls in vars(scraper_module).items(): if issubclass(cls, BaseSiteScraper) and cls is not BaseSiteScraper: scrapers.append(cls()) return scrapers class ScraperScheduler: def __init__(self): self.scrapers = load_all_scrapers() def get_all_sites_value(self, match_keyword: str): futures = [] with ThreadPoolExecutor() as executor: for scraper in self.scrapers: futures.append(executor.submit(scraper.get_target_value, match_keyword)) return "".join(f.result() for f in futures) def refresh_all_sites(self): futures = [] with ThreadPoolExecutor() as executor: for scraper in self.scrapers: futures.append(executor.submit(scraper.setup))
注意:原代码中
executor.submit(scraper.setup())存在调用时机错误,是直接执行完setup()后把返回值提交到线程池,根本不会触发并发执行,重构后的代码中直接传入方法引用,会正确把任务丢给线程池并发执行。
重构后的实际收益
- 新增站点时,只需要写2份内容:1. 站点专属的定位器、两个Page类(链接收集类、详情页类);2. 一个几行代码的
BaseSiteScraper子类,把之前提到的几个配置项填进去即可,不需要重复写driver初始化、流程跳转、文件读写这类通用代码,开发效率提升70%以上。 - 新增全站点通用能力时,只需要修改两个公共文件:
BasePage加通用交互方法,BaseSiteScraper加通用流程逻辑,所有站点自动获得对应能力,不需要逐站点修改、调试,维护成本降低80%以上。 - 调度层完全不需要手动维护爬虫列表,新增站点只要把文件放到对应目录,调度器自动识别加载,彻底消除新增站点时修改main.py的冗余工作。
内容的提问来源于stack exchange,提问作者Dominykas Kišonas

