import logging import time from abc import ABC, abstractmethod from typing import Dict, List, Optional, Any from datetime import datetime import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry from config.settings import CRAWLER_CONFIG from utils.helpers import retry, random_delay, clean_price class BaseCrawler(ABC): """ 爬虫基类 所有平台爬虫都应该继承这个类 """ platform: str = '' def __init__(self): self.logger = logging.getLogger(__name__) self.config = CRAWLER_CONFIG.get(self.platform, {}) self.headers = self.config.get('headers', {}) self.timeout = self.config.get('timeout', 30) self.retry_times = self.config.get('retry_times', 3) self.retry_delay = self.config.get('retry_delay', 2) self.delay_range = self.config.get('delay_range', (1, 3)) self.session = self._create_session() def _create_session(self) -> requests.Session: """ 创建带有重试机制的会话 """ session = requests.Session() session.headers.update(self.headers) retry_strategy = Retry( total=self.retry_times, backoff_factor=self.retry_delay, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET", "POST"] ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("http://", adapter) session.mount("https://", adapter) return session def _make_request(self, method: str, url: str, params: Optional[Dict] = None, data: Optional[Dict] = None, json: Optional[Dict] = None, headers: Optional[Dict] = None) -> Optional[requests.Response]: """ 发送HTTP请求 :param method: HTTP方法 GET/POST :param url: 请求URL :param params: URL参数 :param data: 表单数据 :param json: JSON数据 :param headers: 额外的请求头 :return: Response对象或None """ try: request_headers = self.headers.copy() if headers: request_headers.update(headers) self.logger.debug(f"发送 {method} 请求: {url}") if method.upper() == 'GET': response = self.session.get( url, params=params, headers=request_headers, timeout=self.timeout ) else: response = self.session.post( url, params=params, data=data, json=json, headers=request_headers, timeout=self.timeout ) response.raise_for_status() return response except requests.exceptions.RequestException as e: self.logger.error(f"请求失败: {url} - {e}") return None def _delay(self, min_seconds: float = None, max_seconds: float = None) -> float: """ 随机延迟 :param min_seconds: 最小延迟时间，默认使用配置中的值 :param max_seconds: 最大延迟时间，默认使用配置中的值 :return: 实际延迟时间 """ min_sec = min_seconds if min_seconds is not None else self.delay_range[0] max_sec = max_seconds if max_seconds is not None else self.delay_range[1] return random_delay(min_sec, max_sec) @abstractmethod def search(self, keyword: str, **kwargs) -> List[Dict[str, Any]]: """ 搜索商品（抽象方法，子类必须实现） :param keyword: 搜索关键词 :param kwargs: 其他参数 :return: 商品列表 """ pass @abstractmethod def get_product_detail(self, product_id: str, **kwargs) -> Optional[Dict[str, Any]]: """ 获取商品详情（抽象方法，子类必须实现） :param product_id: 商品ID :param kwargs: 其他参数 :return: 商品详情 """ pass @abstractmethod def get_price(self, product_id: str, **kwargs) -> Optional[Dict[str, Any]]: """ 获取商品价格（抽象方法，子类必须实现） :param product_id: 商品ID :param kwargs: 其他参数 :return: 价格信息 """ pass def crawl_products(self, keywords: List[str], save_to_db: bool = True, **kwargs) -> List[Dict[str, Any]]: """ 爬取多个关键词的商品 :param keywords: 关键词列表 :param save_to_db: 是否保存到数据库 :param kwargs: 其他参数 :return: 所有爬取的商品列表 """ all_products = [] for keyword in keywords: self.logger.info(f"开始搜索关键词: {keyword}") products = self.search(keyword, **kwargs) self.logger.info(f"搜索 '{keyword}' 找到 {len(products)} 个商品") for product in products: product['keyword'] = keyword product['crawl_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') all_products.append(product) self._delay() if save_to_db: self._save_products_to_db(all_products) return all_products def _save_products_to_db(self, products: List[Dict[str, Any]]) -> int: """ 保存商品到数据库 :param products: 商品列表 :return: 成功保存的数量 """ try: from utils.db_utils import DBUtils db = DBUtils() success_count = 0 for product in products: product_data = self._prepare_product_data(product) db_product = db.add_product(product_data) if db_product: price_data = self._prepare_price_data(product) if price_data: db.add_price_history(price_data) success_count += 1 self.logger.info(f"成功保存 {success_count}/{len(products)} 个商品到数据库") return success_count except Exception as e: self.logger.error(f"保存商品到数据库失败: {e}") return 0 def _prepare_product_data(self, product: Dict[str, Any]) -> Dict[str, Any]: """ 准备商品数据，转换为数据库格式 :param product: 原始商品数据 :return: 格式化后的商品数据 """ return { 'product_id': str(product.get('product_id', '')), 'name': product.get('name', ''), 'url': product.get('url', ''), 'image_url': product.get('image_url', ''), 'shop_name': product.get('shop_name', ''), 'platform': self.platform, 'category': product.get('category', ''), 'brand': product.get('brand', ''), 'is_wholesale': 1 if product.get('is_wholesale', False) else 0, 'min_order_quantity': product.get('min_order_quantity'), 'unit': product.get('unit', ''), 'current_price': clean_price(str(product.get('price', 0))), 'original_price': clean_price(str(product.get('original_price', 0))) if product.get('original_price') else None, 'currency': product.get('currency', 'CNY'), 'sales_volume': product.get('sales_volume'), 'rating': product.get('rating'), 'review_count': product.get('review_count'), 'stock': product.get('stock'), 'description': product.get('description', ''), 'crawl_time': datetime.now() } def _prepare_price_data(self, product: Dict[str, Any]) -> Optional[Dict[str, Any]]: """ 准备价格历史数据 :param product: 商品数据 :return: 价格数据字典或None """ price = clean_price(str(product.get('price', 0))) if price <= 0: return None return { 'product_id': str(product.get('product_id', '')), 'price': price, 'original_price': clean_price(str(product.get('original_price', 0))) if product.get('original_price') else None, 'currency': product.get('currency', 'CNY'), 'platform': self.platform, 'source_url': product.get('url', ''), 'crawl_time': datetime.now(), 'price_type': 'wholesale' if product.get('is_wholesale', False) else 'retail', 'min_quantity': product.get('min_order_quantity'), 'max_quantity': product.get('max_order_quantity') } def close(self): """ 关闭会话 """ if hasattr(self, 'session'): self.session.close() def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.close()