AutoScraperX/spiders/base_spider.py

import asyncio
import uuid
from typing import List, Dict, Optional, Any

import aiohttp

from core.models.video_item import VideoItem
from core.utils.helpers import generate_titles
from core.utils.request_preparer import RequestPreparer
from core.utils.spider_config import SpiderConfig
from core.utils.extractors import safe_extract, extract_fields
from core.utils.log.logger_manager import LoggerManager
from services.async_mysql_service import AsyncMysqlService
from services.pipeline import PiaoQuanPipeline
from core.base.async_request_client import AsyncRequestClient
from services.async_mq_producer import AsyncMQProducer


class BaseSpider:
    """
    通用爬虫基类，支持：
    - 依赖请求参数动态替换（cursor 或其它参数）
    - 支持单请求和依赖请求的分页抓取
    - 统一日志、MQ推送、异常捕获、异步请求
    子类只需根据业务重写少量方法，如 process_video/process_item。
    """

    def __init__(self, rule_dict: Dict, user_list: List, env: str = "prod"):
        self.rule_dict = rule_dict
        self.user_list = user_list
        self.env = env
        self.class_name = self.__class__.__name__.lower()

        # --- 1. 初始化核心组件 ---
        self._setup_configuration()
        self._setup_logging()
        self._setup_services()
        self._setup_state()

    #  初始化辅助方法
    def _setup_configuration(self):
        """加载并设置爬虫的核心配置。"""
        self.platform_config = SpiderConfig.get_platform_config(classname=self.class_name)
        if not self.platform_config:
            raise ValueError(f"找不到爬虫配置: {self.class_name}")

        self.platform = self.platform_config.platform
        self.mode = self.platform_config.mode
        self.url = self.platform_config.url
        self.method = self.platform_config.method.upper()
        self.headers = self.platform_config.headers or {}

        # 请求和解析相关的配置
        self.request_body_template = self.platform_config.request_body or {}
        self.response_parse_config = self.platform_config.response_parse or {}
        self.data_path = self.response_parse_config.get("data_path")
        # self.next_cursor_path = self.response_parse_config.get("next_cursor")
        self.field_map = self.response_parse_config.get("fields", {})

        # 爬取行为相关的配置
        self.loop_times = self.platform_config.loop_times or 100
        self.loop_interval = self.platform_config.loop_interval or 5
        self.timeout = self.platform_config.request_timeout or 30
        self.max_retries = self.platform_config.max_retries or 3
        self.feishu_sheetid = self.platform_config.feishu_sheetid

    def _setup_logging(self):
        """初始化日志记录器。"""
        self.logger = LoggerManager.get_logger(platform=self.platform, mode=self.mode)
        self.aliyun_log = LoggerManager.get_aliyun_logger(platform=self.platform, mode=self.mode)
        self.logger.info(f"爬虫 '{self.platform}/{self.mode}' 初始化...")
        self.logger.info(f"最大循环次数: {self.loop_times}, 循环间隔: {self.loop_interval}s")

    def _setup_services(self):
        """初始化外部服务客户端。"""
        self.request_client = AsyncRequestClient(logger=self.logger, aliyun_log=self.aliyun_log)
        self.db_service = AsyncMysqlService(platform=self.platform, mode=self.mode)
        self.mq_producer = AsyncMQProducer(topic_name="topic_crawler_etl_prod_v2", platform=self.platform,
                                           mode=self.mode)

    def _setup_state(self):
        """初始化爬虫的内部状态。"""
        self.last_response_data = {}
        self.request_preparer = RequestPreparer(
            response_parse_config=self.response_parse_config,
            logger=self.logger,
            aliyun_log=self.aliyun_log
        )



    async def run(self):
        """ 爬虫主流程 """
        await self.before_run()
        total_success, total_fail = 0, 0
        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=self.timeout)) as session:
            for loop_index in range(self.loop_times):
                if not await self.is_video_count_sufficient():
                    self.logger.info(f"视频抓取数量已达上限，停止爬取")
                    return
                succ, fail = await self.run_single_loop(session)
                total_success += succ
                total_fail += fail
                await self._wait_for_next_loop(loop_index + 1)

            self.logger.info(f"爬虫完成 成功:{total_success} 失败:{total_fail}")

    async def run_single_loop(self, session) -> (int, int):
        """
        执行单轮的请求、解析和处理。
        返回: (本轮成功处理的数量, 本轮失败处理的数量)
        """
        success_count, fail_count = 0, 0
        try:
            # 爬取数据
            videos = await self.crawl_data(session)
            if not videos:
                self.logger.info(f"无数据返回，停止本轮")
                return success_count, fail_count

            for video in videos:
                # 依赖接口请求
                video_obj = await self.fetch_dependent_data(video)
                res = await self.process_and_push_video(video_obj)
                if res:
                    success_count += 1
                else:
                    fail_count += 1
            self.logger.info(f"接口返回<{len(videos)}>条视频，处理成功<{success_count}>条，处理失败：<{fail_count}>")
            await self.after_run()

        except Exception as e:
            self.logger.exception(f"运行异常: {e}")

        return success_count, fail_count

    async def fetch_dependent_data(self, video: Dict) -> Dict:
        """
        可在子类重写以实现依赖请求，用返回结果补充原有 video。
        默认不做处理。
        """
        return video

    async def crawl_data(self, session) -> Optional[List[Dict]]:
        """
        请求接口，自动渲染动态参数，自动更新游标
        支持单请求和多请求（分页）逻辑。
        """
        request_body = self.request_preparer.prepare(self.request_body_template,
                                                     self.last_response_data)
        # 发送请求
        response = await self.request_client.request(
            session=session,
            method=self.method,
            url=self.url,
            headers=self.headers,
            json = request_body
        )

        if not response:
            self.logger.error(f"响应为空")
            return

        self.last_response_data = response
        # 解析数据列表
        data_list = safe_extract(response, self.data_path)
        if not data_list:
            self.logger.info(f"接口返回视频列表为空{response}")
            self.aliyun_log.logging(
                code="9021",
                message="接口返回视频列表为空",
                data= response
            )
            return

        return data_list

    async def process_and_push_video(self, video: Dict[str, Any]) -> bool:
        """
        数据处理完整流程（字段映射 -> 校验 -> 推送）
        子类可重写 process_video 或 filter_data 来定制处理和校验逻辑
        """
        try:
            # 字段映射
            video_obj = await self.process_video(video)
            if not video_obj:
                return False

            if not await self.filter_data(video_obj):
                return False

            await self.integrated_video_handling(video_obj)
            pushed = await self.push_to_etl(video_obj)
            return pushed
        except Exception as e:
            self.logger.exception(f"视频处理异常: {e}")
            return False

    async def process_video(self, video: Dict) -> Optional[Dict]:
        """
        统一字段抽取及 VideoItem 初始化
        子类可重写或扩展以定制字段映射、过滤等
        """
        self.logger.debug(f"处理视频数据: {video.get('title', '无标题')}")
        if self.user_list:
            import random
            publish_user = random.choice(self.user_list)
        else:
            self.logger.error(f"未获取到用户列表数据{self.user_list}")
            return

        item_kwargs = extract_fields(video, self.field_map, logger=self.logger,aliyun_log=self.aliyun_log)
        item_kwargs.update({
            "user_id": publish_user.get("uid"),
            "user_name": publish_user.get("nick_name"),
            "platform": self.platform,
            "strategy": self.mode,
        })

        try:
            item = VideoItem(**item_kwargs)
            video_dict = await item.produce_item()
            if not video_dict:
                self.logger.warning(f"VideoItem 校验失败")
                return None
            return video_dict
        except Exception as e:
            self.logger.error(f"VideoItem 初始化失败: {e}")
            return None

    async def filter_data(self, video: Dict) -> bool:
        """
        数据校验过滤，默认使用 PiaoQuanPipeline
        子类可重写此方法实现自定义过滤
        """
        pipeline = PiaoQuanPipeline(
            platform=self.platform,
            mode=self.mode,
            rule_dict=self.rule_dict,
            env=self.env,
            item=video,
            trace_id=self.platform + str(uuid.uuid1())
        )
        return await pipeline.process_item()

    async def integrated_video_handling(self, video: Dict) -> None:
        """
        钩子函数：可在此实现自动生成标题或其他业务逻辑
        """
        await generate_titles(self.feishu_sheetid, video)

    async def push_to_etl(self, video: Dict) -> bool:
        """
        推送消息到消息队列ETL
        """
        try:
            await self.mq_producer.send_msg(video)
            self.logger.info(f"成功推送视频至ETL")
            return True
        except Exception as e:
            self.logger.exception(f"推送ETL失败: {e}")
            return False

    async def is_video_count_sufficient(self) -> bool:
        """
        判断当天抓取的视频是否已达到上限，达到则停止继续抓取
        """
        max_count = self.rule_dict.get("videos_cnt", {}).get("min", 0)
        if max_count <= 0:
            return True
        async with AsyncMysqlService(self.platform, self.mode) as mysql:
            current_count = await mysql.get_today_videos()
            if current_count >= max_count:
                self.aliyun_log.logging(code="1011", message="视频数量达到当日最大值", data=f"<今日视频数量>{current_count}")
                return False
            self.logger.info(f"{self.platform} 今日入库视频数: {current_count}/{max_count}")
            self.aliyun_log.logging(code="1012",
                                    message=f"目前入库量{current_count}",
                                    data=f"{current_count}/{max_count}"
                                    )
            return True

    async def _wait_for_next_loop(self, current_loop: int) -> None:
        """等待下次循环"""
        if current_loop < self.loop_times and self.loop_interval > 0:
            self.logger.info(f"等待 {self.loop_interval} 秒后进行下一次请求")
            await asyncio.sleep(self.loop_interval)

    async def before_run(self):
        """运行前预处理钩子，子类可覆盖"""
        pass

    async def after_run(self):
        """运行后处理钩子，子类可覆盖"""
        pass