请求协助完善Pawrade法国斗牛犬爬虫的状态提取功能
修复Pawrade爬虫中Status列的判断逻辑
你的爬虫Status列失效,核心问题在于售罄消息的定位逻辑不够健壮,以及文本匹配未处理空白字符。以下是针对性的修复方案:
问题诊断
- 选择器依赖不稳定:原代码硬编码
h4.mb-0定位售罄消息,若页面结构调整(比如消息移到其他标签),判断会直接失效。 - 文本匹配不严谨:直接用
in匹配原始文本,未去除换行、空格等空白字符,可能导致明明存在消息却匹配失败。 - 重复DOM查询:连续两次调用
puppy_soup.find("h4", class_="mb-0"),既低效又可能出现前后结果不一致的情况。
修复方案
1. 替换Status判断逻辑
使用更灵活的文本匹配方式,同时处理空白字符,确保匹配精准:
# 替换原Status判断代码段 status = "available" sold_message = "This puppy is no longer available." # 优先查找可能包含售罄消息的标签(h4/div/p),并去除文本空白 sold_element = puppy_soup.find( lambda tag: tag.name in ["h4", "div", "p"] and sold_message in tag.get_text(strip=True) ) if sold_element: status = "sold"
如果确认售罄消息固定在某个特定容器(比如带alert类的警告框),可以用更精准的选择器提升效率:
sold_element = puppy_soup.find("div", class_="alert") if sold_element and sold_message in sold_element.get_text(strip=True): status = "sold"
2. 优化DOM查询效率
把重复的find结果存为变量,避免重复遍历DOM:
status = "available" h4_element = puppy_soup.find("h4", class_="mb-0") if h4_element and sold_message in h4_element.get_text(strip=True): status = "sold"
3. 增加异常处理(可选)
处理页面访问失败的情况,避免脚本中断:
try: puppy_response = requests.get(puppy_url, headers={"User-Agent": "Mozilla/5.0"}, timeout=10) puppy_response.raise_for_status() # 触发HTTP错误抛出 puppy_soup = BeautifulSoup(puppy_response.content, "html.parser") except requests.exceptions.RequestException as e: print(f"访问幼犬页面失败 {puppy_url}: {str(e)}") status = "unknown" # 写入空值并跳过后续解析 csv_writer.writerow([adid, name, breed, price, puppy_url, "", "", "", "", "", "", status]) continue
4. 防反爬优化(可选)
添加浏览器请求头,避免被网站拦截:
# 在脚本开头定义请求头 HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36" } # 所有requests.get请求都传入headers response = requests.get(url, headers=HEADERS) puppy_response = requests.get(puppy_url, headers=HEADERS)
修复后的完整核心代码片段
import requests from bs4 import BeautifulSoup import csv HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36" } def scrape_page(url, csv_writer): response = requests.get(url, headers=HEADERS) soup = BeautifulSoup(response.content, "html.parser") spans = soup.find_all("span", class_="fave position-absolute") for span in spans: adid = span.get("data-adid") name = span.get("data-name") breed = span.get("data-breed") price = span.get("data-price") puppy_url = f"https://www.pawrade.com/puppies/french-bulldog/{adid}" status = "available" sold_message = "This puppy is no longer available." dob = weight = registration = color = release_date = microchip = "" try: puppy_response = requests.get(puppy_url, headers=HEADERS, timeout=10) puppy_response.raise_for_status() puppy_soup = BeautifulSoup(puppy_response.content, "html.parser") dob = extract_detail(puppy_soup, "fa-birthday-cake") weight = extract_detail(puppy_soup, "fa-weight") registration = extract_detail(puppy_soup, "fa-trophy") color = extract_detail(puppy_soup, "fa-palette") release_date = extract_detail(puppy_soup, "fa-calendar-alt") microchip = extract_detail(puppy_soup, "fa-microchip") # 修复后的Status判断 sold_element = puppy_soup.find( lambda tag: tag.name in ["h4", "div", "p"] and sold_message in tag.get_text(strip=True) ) if sold_element: status = "sold" except requests.exceptions.RequestException as e: print(f"访问失败 {puppy_url}: {str(e)}") status = "unknown" csv_writer.writerow([adid, name, breed, price, puppy_url, dob, weight, registration, color, release_date, microchip, status]) # 辅助函数和主逻辑保持不变... def extract_detail(soup, class_name): detail_element = soup.find("span", class_=class_name) if detail_element: detail_text = detail_element.find_next_sibling("div").find("small").text.strip() return detail_text return "" base_url = "https://www.pawrade.com/puppies/french-bulldog/" with open("french_bulldogs.csv", mode="w", newline="", encoding="utf-8") as csv_file: csv_writer = csv.writer(csv_file) csv_writer.writerow(["AdID", "Name", "Breed", "Price", "URL", "DOB", "Weight", "Registration", "Color", "Release Date", "Microchip", "Status"]) response = requests.get(base_url, headers=HEADERS) soup = BeautifulSoup(response.content, "html.parser") pagination = soup.find("ul", class_="pagination") if pagination: pages = pagination.find_all("li", class_="page-item") total_pages = int(pages[-2].text.strip()) else: total_pages = 1 for page in range(1, total_pages + 1): url = f"{base_url}?page={page}" scrape_page(url, csv_writer) print("Data has been written to french_bulldogs.csv")
内容的提问来源于stack exchange,提问作者Cameron Ratliff
相关产品推荐
相关产品推荐

