如何优化bs4爬虫代码,提升代码整洁度?
问题描述
我正在使用bs4开发一个爬虫项目,完全找不到让代码不再辣眼睛的方法。代码示例如下:
try: max_tweet_date = max( [tweet_date_to_datetime(timeline_item.find('span', {'class': 'tweet-date'}).find('a')['title']) for timeline_item in tweets_or_replies_soup.find_all('div', {'class': 'timeline-item'})] ) except AttributeError: print('No tweets or replies') print() neo4j_database.set_user_updated_now(current_user) except IndexError: print('Not enough tweets or no tweets') print() neo4j_database.set_user_updated_now(current_user) except ValueError: print("Cannot recognise dates, skipping") print() neo4j_database.set_user_updated_now(current_user) if is_tweet_too_old(max_tweet_date): print('User is inactive, deleting') print() neo4j_database.delete_user(current_user)
这类代码总有大量需要处理的异常,且属性处理逻辑迫切需要封装到独立函数中。请问有实用的设计模式或其他解析库能帮我整理这些混乱的代码吗?
解决方案
一、封装解析逻辑,减少嵌套与重复异常处理
把DOM解析和日期转换的逻辑拆成独立函数,用防御式编程避免链式调用的崩溃,同时统一异常处理逻辑:
1. 封装单条推文的日期解析函数
def extract_tweet_date(timeline_item): """提取单条推文的日期并转换为datetime对象""" date_span = timeline_item.find('span', {'class': 'tweet-date'}) if not date_span: return None date_link = date_span.find('a') if not date_link or 'title' not in date_link.attrs: return None try: return tweet_date_to_datetime(date_link['title']) except ValueError: return None
2. 批量处理并统一异常
把列表推导式改成更可控的循环,过滤无效数据后再求最大值,同时合并重复的数据库操作:
# 批量提取有效日期 tweet_dates = [] for item in tweets_or_replies_soup.find_all('div', {'class': 'timeline-item'}): dt = extract_tweet_date(item) if dt: tweet_dates.append(dt) # 统一处理无有效数据的情况 if not tweet_dates: print('No valid tweets or date parsing failed') neo4j_database.set_user_updated_now(current_user) else: max_tweet_date = max(tweet_dates) if is_tweet_too_old(max_tweet_date): print('User is inactive, deleting') neo4j_database.delete_user(current_user) neo4j_database.set_user_updated_now(current_user)
这样既避免了链式调用的AttributeError,也把重复的数据库操作统一处理,减少代码冗余。
二、使用设计模式优化
1. 责任链模式(Chain of Responsibility)
如果后续需要扩展更多解析规则(比如不同版本的页面结构),可以把每个解析步骤做成独立的处理器,依次尝试解析,直到成功或全部失败:
class DateParserHandler: def __init__(self, next_handler=None): self.next_handler = next_handler def parse(self, timeline_item): raise NotImplementedError class DefaultDateParser(DateParserHandler): def parse(self, timeline_item): date_span = timeline_item.find('span', {'class': 'tweet-date'}) if date_span and date_span.find('a') and 'title' in date_span.find('a').attrs: try: return tweet_date_to_datetime(date_span.find('a')['title']) except ValueError: pass return self.next_handler.parse(timeline_item) if self.next_handler else None class FallbackDateParser(DateParserHandler): def parse(self, timeline_item): # 处理另一种页面结构的日期解析 date_tag = timeline_item.find('time', {'datetime': True}) if date_tag: try: return datetime.fromisoformat(date_tag['datetime']) except ValueError: pass return None # 使用时 parser_chain = DefaultDateParser(FallbackDateParser()) tweet_dates = [parser_chain.parse(item) for item in timeline_items if parser_chain.parse(item)]
这种模式适合页面结构多变的场景,方便扩展解析规则。
2. 策略模式(Strategy Pattern)
如果针对不同类型的页面需要不同的解析策略,可以定义多个解析策略类,根据页面类型切换使用,核心是把解析逻辑和业务逻辑解耦,后续新增页面类型只需添加新策略类即可。
三、替代解析库推荐
1. Parsel
基于bs4开发,支持XPath和CSS选择器,语法更简洁,同时可以通过get()/getall()方法避免键不存在的异常:
from parsel import Selector selector = Selector(text=html_content) # 使用getall()获取所有匹配的属性值,不存在则返回空列表 date_titles = selector.css('div.timeline-item span.tweet-date a::attr(title)').getall() tweet_dates = [] for title in date_titles: try: tweet_dates.append(tweet_date_to_datetime(title)) except ValueError: pass
getall()直接返回所有匹配结果,避免了链式find()的繁琐。
2. Scrapy的Item Loader
如果爬虫规模较大,可以用Scrapy的Item Loader来封装数据提取逻辑,它内置了输入处理器和输出处理器,能自动处理空值、转换数据格式,减少重复代码:
from scrapy.loader import ItemLoader from scrapy.loader.processors import TakeFirst, MapCompose class TweetLoader(ItemLoader): default_output_processor = TakeFirst() date_in = MapCompose(lambda x: tweet_date_to_datetime(x) if x else None) loader = TweetLoader(selector=timeline_item) loader.add_css('date', 'span.tweet-date a::attr(title)') tweet_date = loader.get_output_value('date')
它会自动过滤空值,处理转换异常(需要自定义处理器时可以扩展),适合结构化的数据提取。
内容的提问来源于stack exchange,提问作者Андрей Прохоров
相关产品推荐
相关产品推荐

