CoolBotDataSys/crawlers/base.py

"""爬虫基类 - 提供通用爬虫功能"""
import requests
from typing import Optional, Dict, List, Any, Callable
from datetime import datetime
from abc import ABC, abstractmethod
import time
import logging
import random
from urllib.parse import urljoin
import json

logger = logging.getLogger(__name__)

class BaseSpider(ABC):
    """爬虫基类"""
    
    def __init__(self, name: str, source: str):
        self.name = name
        self.source = source
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
            "Accept-Encoding": "gzip, deflate, br",
            "Connection": "keep-alive",
        })
        
        # 限流配置
        self.min_delay = 2.0  # 最小请求间隔（秒）
        self.max_delay = 5.0  # 最大请求间隔（秒）
        self.last_request_time = 0
        
        # 重试配置
        self.max_retries = 3
        self.retry_delay = 5
        
        # 代理配置（可选）
        self.proxies: Optional[Dict] = None
        
        self.logger = logging.getLogger(f"{__name__}.{name}")
    
    def _random_delay(self):
        """随机延时（模拟人类行为）"""
        delay = random.uniform(self.min_delay, self.max_delay)
        elapsed = time.time() - self.last_request_time
        if elapsed < delay:
            time.sleep(delay - elapsed)
        self.last_request_time = time.time()
    
    def _request(self, method: str, url: str, **kwargs) -> Optional[requests.Response]:
        """发送请求（带重试和延时）"""
        for attempt in range(self.max_retries):
            try:
                self._random_delay()
                
                response = self.session.request(
                    method=method,
                    url=url,
                    proxies=self.proxies,
                    timeout=30,
                    **kwargs
                )
                
                if response.status_code == 200:
                    return response
                elif response.status_code == 403:
                    self.logger.warning(f"403 Forbidden，可能需要登录: {url}")
                    return response
                elif response.status_code == 404:
                    self.logger.warning(f"404 Not Found: {url}")
                    return None
                elif response.status_code >= 500:
                    self.logger.warning(f"服务器错误 {response.status_code}，重试 {attempt + 1}/{self.max_retries}")
                    time.sleep(self.retry_delay)
                    continue
                else:
                    response.raise_for_status()
                    
            except requests.exceptions.Timeout:
                self.logger.warning(f"请求超时，重试 {attempt + 1}/{self.max_retries}")
            except requests.exceptions.RequestException as e:
                self.logger.warning(f"请求异常: {e}，重试 {attempt + 1}/{self.max_retries}")
                time.sleep(self.retry_delay)
        
        return None
    
    def get(self, url: str, **kwargs) -> Optional[requests.Response]:
        """GET请求"""
        return self._request("GET", url, **kwargs)
    
    def post(self, url: str, **kwargs) -> Optional[requests.Response]:
        """POST请求"""
        return self._request("POST", url, **kwargs)
    
    def save_cookies(self, filepath: str):
        """保存Cookies"""
        with open(filepath, "w") as f:
            json.dump(self.session.cookies.get_dict(), f)
        self.logger.info(f"Cookies已保存到 {filepath}")
    
    def load_cookies(self, filepath: str):
        """加载Cookies"""
        try:
            with open(filepath, "r") as f:
                cookies = json.load(f)
                self.session.cookies.update(cookies)
            self.logger.info(f"Cookies已从 {filepath} 加载")
        except FileNotFoundError:
            self.logger.warning(f"Cookies文件不存在: {filepath}")
    
    def parse_user(self, html: str, url: str) -> Optional[Dict]:
        """解析用户信息（子类实现）"""
        pass
    
    def parse_posts(self, html: str, url: str) -> List[Dict]:
        """解析帖子列表（子类实现）"""
        pass
    
    def parse_post_detail(self, html: str, url: str) -> Optional[Dict]:
        """解析帖子详情（子类实现）"""
        pass


class PaginationSpider(BaseSpider):
    """分页爬虫基类"""
    
    def __init__(self, name: str, source: str):
        super().__init__(name, source)
        self.max_pages = 10  # 默认最大页数
    
    def crawl_paginated(self, base_url: str, page_parser: Callable, max_pages: Optional[int] = None) -> List[Dict]:
        """爬取分页数据"""
        if max_pages:
            self.max_pages = max_pages
        
        all_items = []
        for page in range(1, self.max_pages + 1):
            page_url = self._get_page_url(base_url, page)
            self.logger.info(f"爬取第 {page} 页: {page_url}")
            
            response = self.get(page_url)
            if not response:
                break
            
            items = page_parser(response.text, response.url)
            if not items:
                self.logger.info(f"第 {page} 页无数据，停止")
                break
            
            all_items.extend(items)
            self.logger.info(f"第 {page} 页获取 {len(items)} 条数据")
        
        return all_items
    
    def _get_page_url(self, base_url: str, page: int) -> str:
        """生成页码URL（子类可重写）"""
        if "?" in base_url:
            return f"{base_url}&page={page}"
        return f"{base_url}?page={page}"