#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Product Hunt网站数据抓取脚本 使用Selenium连接到现有的Chrome实例，抓取Product Hunt产品信息 """ from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.common.exceptions import TimeoutException, NoSuchElementException from webdriver_manager.chrome import ChromeDriverManager from selenium.webdriver.chrome.service import Service import time import json import re from datetime import datetime from loguru import logger # 配置日志 logger.add("producthunt_scraper.log", rotation="10 MB", level="INFO") class ProductHuntScraper: """Product Hunt网站数据抓取器""" def __init__(self, debug_address="127.0.0.1:5003"): """ 初始化抓取器，连接到现有的Chrome实例 Args: debug_address (str): Chrome调试地址，默认为"127.0.0.1:5003" """ self.debug_address = debug_address self.driver = None self.wait = None def connect_to_chrome(self): """ 连接到现有的Chrome实例 Returns: bool: 连接是否成功 """ try: logger.info(f"尝试连接到Chrome调试实例: {self.debug_address}") options = webdriver.ChromeOptions() options.add_experimental_option("debuggerAddress", self.debug_address) # 使用webdriver-manager自动管理Chrome驱动程序 service = Service(ChromeDriverManager().install()) self.driver = webdriver.Chrome(service=service, options=options) self.wait = WebDriverWait(self.driver, 10) logger.info(f"成功连接到Chrome实例，当前页面标题: {self.driver.title}") return True except Exception as e: logger.error(f"连接Chrome实例失败: {e}") return False def navigate_to_product(self, product_url): """ 导航到指定的Product Hunt产品页面 Args: product_url (str): 产品页面URL Returns: bool: 导航是否成功 """ try: logger.info(f"正在导航到产品页面: {product_url}") self.driver.get(product_url) # 等待页面加载完成 self.wait.until(EC.presence_of_element_located((By.TAG_NAME, "body"))) time.sleep(3) # 额外等待，确保动态内容加载 logger.info(f"成功导航到产品页面，当前标题: {self.driver.title}") return True except TimeoutException: logger.error("页面加载超时") return False except Exception as e: logger.error(f"导航到产品页面失败: {e}") return False def extract_product_info(self): """ 从当前页面提取产品信息 Returns: dict: 包含产品信息的字典 """ product_info = {} try: # 提取产品名称 - h1标签下的字符串 try: h1_element = self.driver.find_element(By.TAG_NAME, "h1") product_name = h1_element.text.strip() logger.info(f"找到产品名称: {product_name}") product_info['name'] = product_name or "未找到产品名称" except NoSuchElementException: logger.error("未找到h1标签") product_info['name'] = "未找到产品名称" except Exception as e: logger.error(f"提取产品名称失败: {e}") product_info['name'] = "提取失败" # 提取产品简介 - div的class="relative text-16 font-normal text-gray-700"的字符串 try: description_element = self.driver.find_element(By.CSS_SELECTOR, "div.relative.text-16.font-normal.text-gray-700") product_description = description_element.text.strip() logger.info(f"找到产品简介: {product_description[:50]}...") product_info['description'] = product_description or "未找到产品简介" except NoSuchElementException: logger.error("未找到产品简介div") product_info['description'] = "未找到产品简介" except Exception as e: logger.error(f"提取产品简介失败: {e}") product_info['description'] = "提取失败" # 提取第一个评论 - div的class是flex flex-1 flex-col gap-2，提取字符串的所有内容 try: comment_elements = self.driver.find_elements(By.CSS_SELECTOR, "div.flex.flex-1.flex-col.gap-2") if comment_elements: first_comment = comment_elements[0].text.strip() logger.info(f"找到第一个评论: {first_comment[:50]}...") product_info['first_comment'] = first_comment else: logger.warning("未找到任何评论") product_info['first_comment'] = "未找到评论" except Exception as e: logger.error(f"提取第一个评论失败: {e}") product_info['first_comment'] = "提取失败" # 添加当前URL和抓取时间 product_info['url'] = self.driver.current_url product_info['scraped_at'] = datetime.now().isoformat() return product_info except Exception as e: logger.error(f"提取产品信息时出错: {e}") return { 'name': "提取失败", 'description': "提取失败", 'first_comment': "提取失败", 'url': self.driver.current_url if self.driver else "未知", 'scraped_at': datetime.now().isoformat(), 'error': str(e) } def scrape_product(self, product_url): """ 抓取指定URL的产品信息 Args: product_url (str): 产品页面URL Returns: dict: 产品信息字典 """ if not self.driver: if not self.connect_to_chrome(): logger.error("无法连接到Chrome实例") return None if not self.navigate_to_product(product_url): logger.error("无法导航到产品页面") return None return self.extract_product_info() def save_to_file(self, product_info, filename=None): """ 将产品信息保存到JSON文件 Args: product_info (dict): 产品信息 filename (str, optional): 文件名。如果未提供，将自动生成 Returns: str: 保存的文件名 """ if not filename: now = datetime.now() filename = f"product_{now.strftime('%Y%m%d_%H%M%S')}.json" try: with open(filename, 'w', encoding='utf-8') as f: json.dump(product_info, f, ensure_ascii=False, indent=2) logger.info(f"产品信息已保存到 {filename}") return filename except Exception as e: logger.error(f"保存文件失败: {e}") raise def close(self): """ 关闭连接 """ if self.driver: self.driver.quit() logger.info("已关闭Chrome连接") if __name__ == "__main__": # 示例用法 scraper = ProductHuntScraper() try: # 要抓取的产品URL product_url = "https://www.producthunt.com/products/elsie-ai-beta" # 抓取产品信息 product_info = scraper.scrape_product(product_url) if product_info: # 打印产品信息 logger.info("抓取到的产品信息:") logger.info(json.dumps(product_info, ensure_ascii=False, indent=2)) # 保存到文件 filename = scraper.save_to_file(product_info) logger.info(f"产品信息已保存到: {filename}") else: logger.error("未能获取产品信息") except Exception as e: logger.error(f"程序执行出错: {e}") finally: # 注意：这里不关闭连接，以便可以继续使用Chrome实例 # 如果需要关闭连接，取消下面的注释 # scraper.close() pass