automatic_crawler/spider/crawler_author/zhongqingkandian_author.py

import os
import sys
import asyncio
import json
import random
import uuid
import time
import traceback
from datetime import datetime
import aiohttp

sys.path.append(os.getcwd())
from application.common.feishu import FsData
from application.common.feishu.feishu_utils import FeishuUtils
from application.common.gpt import GPT4oMini
from application.common.messageQueue import MQ
from application.common.log import AliyunLogger
from application.functions.zqkd_db_redis import DatabaseOperations, RedisOperations
from application.items import VideoItem
from application.pipeline import PiaoQuanPipeline
from application.common.log import Local


class ZhongQingKanDianAuthor:
    API_BASE_URL = "http://8.217.192.46:8889"
    COMMON_HEADERS = {
        "Content-Type": "application/json"
    }
    # 最大重试次数
    MAX_RETRIES = 3
    # 最大等待时长
    TIMEOUT = 30

    def __init__(self, platform, mode, rule_dict, user_list, env="prod"):
        """
        初始化
        :param platform: 平台名称 zhongqingkandian
        :param mode: 运行模式  recommend
        :param rule_dict: 规则字典，包含视频数量限制、时长限制等规则 [{"videos_cnt":{"min":100,"max":0}},{"duration":{"min":30,"max":1200}}]
        :param user_list: 用户列表
        :param env: 运行环境，默认为 "prod"
        """
        self.limit_flag = True
        self.platform = platform
        self.mode = mode
        self.rule_dict = rule_dict
        self.user_list = user_list
        self.env = env
        self.download_cnt = 0
        self.mq = MQ(topic_name="topic_crawler_etl_" + self.env)
        self.expire_flag = False
        self.aliyun_log = AliyunLogger(mode=self.mode, platform=self.platform)
        self.db_ops = DatabaseOperations(mode=mode, platform=platform)
        self.redis_ops = RedisOperations(mode=mode, platform=platform)
        data_rule = FsData()
        self.title_rule = data_rule.get_title_rule()
        self.LocalLog = Local.logger(self.platform, self.mode)
        self.curses = 1
        result = self.redis_ops.get_last_scanned_id()
        self.last_scanned_id = 0 if result is None else int(result)
        self.zqkd_user_list = self.db_ops.select_user(self.last_scanned_id)
        self.LocalLog.info(f"获取到的用户列表：{self.zqkd_user_list} \n 昨天最后扫描的用户ID{self.last_scanned_id}")

    async def send_request(self, path, data):
        """
        异步发送 POST 请求到指定路径，带有重试机制。
        :param path: 请求的 API 路径
        :param data: 请求的数据
        :return: 响应的 JSON 数据，如果请求失败则返回 None
        """
        full_url = f"{self.API_BASE_URL}{path}"
        async with aiohttp.ClientSession(headers=self.COMMON_HEADERS) as session:
            for retry in range(self.MAX_RETRIES):
                try:
                    async with session.post(full_url, data=data, timeout=self.TIMEOUT) as response:
                        response.raise_for_status()
                        self.LocalLog.info(f"{path}响应数据:{await response.json()}")
                        return await response.json()
                except (aiohttp.ClientError, json.JSONDecodeError) as e:
                    tb_info = traceback.format_exc()
                    self.LocalLog.info(f"{path}请求失败:{e}  \n{tb_info}")
                    self.aliyun_log.logging(
                        code="3000",
                        message=f"请求 {path} 失败，错误信息: {str(e)}",
                        data={"path": path}
                    )
                    await asyncio.sleep(5)
        return None

    def is_response_valid(self, resp, url):
        """
        检查响应是否有效（状态码为 0 表示有效）。
        :param resp: 响应数据
        :param url: 请求的 URL
        :return: 如果响应有效则返回响应数据，否则返回 None
        """
        try:
            if resp and resp.get('code') != 0:
                self.aliyun_log.logging(
                    code="3000",
                    message=f"抓取{url}失败,请求失败,响应：{resp}"
                )
                self.LocalLog.info(f"{url}请求失败,响应：{resp}")
                return None
            return resp
        except Exception as e:
            tb_info = traceback.format_exc()
            self.aliyun_log.logging(
                code="3000",
                message=f"检查响应有效性时出错，错误信息: {str(e)}",
                data={"url": url, "resp": resp}
            )
            self.LocalLog.info(f"检查 {url} 响应有效性时出错:{e} \n{tb_info}")
            return None

    async def req_user_list(self, account_id):
        """
        异步请求与指定内容 ID 相关的推荐列表。
        :param
        :return: 相关推荐视频列表的有效响应数据，如果请求失败则返回 None
        """
        try:

            url = '/crawler/zhong_qing_kan_dian/blogger'
            body = json.dumps({
                 "account_id": f"{account_id}",
                 "content_type": "全部",
                 "cursor": f"{self.curses}"
            })
            self.LocalLog.info(f"开始请求用户视频列表{body}")
            resp = await self.send_request(url, body)
            return self.is_response_valid(resp, url)
        except Exception as e:
            tb_info = traceback.format_exc()
            self.aliyun_log.logging(
                code="1004",
                message=f"请求相关推荐视频列表时发生异常，错误信息: {str(e)}",
                data={"url": url}
            )
            self.LocalLog.info(f"请求相关推荐视频列表 {url} 时发生异常:{e}   \n{tb_info}")
            return None

    async def req_detail(self, content_link, **kwargs):
        """
        异步请求视频详情。
        :param content_link: 视频内容链接
        :param kwargs: 额外的视频信息
        :return: 无返回值，处理视频详情信息
        """
        try:
            self.LocalLog.info(f"开始请求视频详情，链接: {content_link}")
            url = '/crawler/zhong_qing_kan_dian/detail'
            body = json.dumps({
                "content_link": content_link
            })
            resp = await self.send_request(url, body)
            if not self.is_response_valid(resp, url):
                return
            data = resp.get("data", {}).get("data", {})
            if data.get("content_type") != "video":
                self.aliyun_log.logging(
                    code="3003",
                    message=f"跳过非视频内容",
                    data={"content_link": content_link}
                )
                self.LocalLog.info(f"跳过非视频内容，链接: {content_link}")
                return
            self.LocalLog.info(f"{content_link} 是视频")
            data.update(kwargs)
            await self.process_video_obj(data)
            await asyncio.sleep(10)
        except Exception as e:
            tb_info = traceback.format_exc()
            self.aliyun_log.logging(
                code="1005",
                message=f"请求视频详情时发生异常，错误信息: {str(e)}",
                data={"content_link": content_link}
            )
            self.LocalLog.error(f"请求视频详情，链接 {content_link} 时发生异常:{e}  \n{tb_info}")
    async def control_request_author(self):
        """
        控制相关推荐视频列表的请求和处理流程。
        :return: 无返回值，根据下载数量限制控制流程
        """
        while self.limit_flag:
            try:
                self.LocalLog.info(f"开始用户视频列表的请求和处理流程,今日已爬 {self.download_cnt} 个视频")

                if not self.zqkd_user_list:
                    self.LocalLog.info("没有用户数据")
                    await asyncio.sleep(10)
                    continue
                for user_info in self.zqkd_user_list:
                    current_id, user_id = user_info
                    author_resp = await self.req_user_list(user_id)
                    if current_id > self.last_scanned_id:
                        self.last_scanned_id = current_id
                    if not author_resp:
                        continue
                    author_data = author_resp.get("data", {}).get("data", [])
                    if not author_data["next_cursor"]:
                        continue
                    for author_obj in author_data:
                        author_content_link = author_obj.get("share_url")
                        if author_content_link:
                            await self.req_detail(author_content_link, **author_obj)
            except Exception as e:
                tb_info = traceback.format_exc()
                self.aliyun_log.logging(
                    code="3009",
                    message=f"控制相关推荐视频请求和处理时发生异常，错误信息: {str(e)}",
                    data={}
                )
                self.LocalLog.info(f"控制相关推荐视频请求和处理时发生异常:\n{tb_info}")

    async def process_video_obj(self, video_obj):
        """
        处理视频对象，包括检查视频时长、用户信息、保存数据等操作。
        :param video_obj: 视频对象，包含视频的各种信息
        :return: 无返回值，完成视频对象的处理
        """
        try:
            video_duration = video_obj["video_url_list"][0]['video_duration']
            video_id = video_obj['channel_content_id']
            # 检查视频ID是否存在
            if self.redis_ops.check_video_id_exists(video_id):
                self.aliyun_log.logging(
                    code="3004",
                    message=f"重复视频ID：{video_id}"
                )
                self.LocalLog.info(f"重复视频ID: {video_id}")
                return
            our_user = random.choice(self.user_list)
            trace_id = self.platform + str(uuid.uuid1())
            item = VideoItem()

            account_id = video_obj["channel_account_id"]
            account_name = video_obj["channel_account_name"]
            account_avatar = video_obj["avatar"]
            # # 检查用户ID是否存在
            # """
            # 需要改为判断redis
            # """
            # is_repeat_user = self.db_ops.check_user_id(account_id)
            # if is_repeat_user:
            #     # 更新用户信息，使用异步方法并等待结果
            #     self.LocalLog.info(f"用户{account_id}已经存在数据库中")
            #     self.db_ops.update_user(account_id, account_name, account_avatar)
            # else:
            #     self.LocalLog.info(f"用户{account_id}没在数据库中")
            #     # 插入用户信息，使用异步方法并等待结果
            #     self.db_ops.insert_user(account_id, account_name, account_avatar)
            #     self.redis_ops.add_user_data("task:zqkd_user_id", json.dumps({"uid": account_id}))
            #     self.aliyun_log.logging(code="1007", message=f"用户数据写入成功，用户ID：{account_id}")
            #     self.LocalLog.info(f"用户数据写入成功，用户ID: {account_id}")

            if video_duration > self.rule_dict.get("duration", {}).get("max",
                                                                       1200) or video_duration < self.rule_dict.get(
                    "duration", {}).get("min", 30):
                self.aliyun_log.logging(
                    code="3005",
                    message=f"视频时长不满足条件[>=30s&<=1200s]视频ID：{video_obj['channel_content_id']}，视频时长：{video_duration}"
                )
                self.LocalLog.info(
                    f"视频时长不满足条件，视频ID: {video_obj['channel_content_id']}, 视频时长: {video_duration}")
                return

            item.add_video_info("video_id", video_obj['channel_content_id'])
            item.add_video_info("video_title", video_obj["title"])
            item.add_video_info("play_cnt", video_obj["read_num"])
            item.add_video_info("publish_time_stamp", int(int(video_obj["publish_timestamp"]) / 1000))
            item.add_video_info("out_user_id", video_obj["channel_account_id"])
            item.add_video_info("cover_url", video_obj["image_url_list"][0]['image_url'])
            item.add_video_info("like_cnt", 0)
            item.add_video_info("collection_cnt", 0)
            item.add_video_info("share_cnt", 0)
            item.add_video_info("comment_cnt", 0)
            item.add_video_info("video_url", video_obj["video_url_list"][0]['video_url'])
            item.add_video_info("out_video_id", int(video_obj["channel_content_id"]))
            item.add_video_info("duration", video_obj["video_url_list"][0]['video_duration'])
            item.add_video_info("platform", self.platform)
            item.add_video_info("strategy", self.mode)
            item.add_video_info("session", f"{self.platform}-{int(time.time())}")
            item.add_video_info("user_id", our_user["uid"])
            item.add_video_info("user_name", our_user["nick_name"])

            mq_obj = item.produce_item()
            pipeline = PiaoQuanPipeline(
                platform=self.platform,
                mode=self.mode,
                rule_dict=self.rule_dict,
                env=self.env,
                item=mq_obj,
                trace_id=trace_id
            )
            if pipeline.process_item():
                title_list = self.title_rule.split(",")
                title = video_obj["title"]
                contains_keyword = any(keyword in title for keyword in title_list)
                if contains_keyword:
                    new_title = GPT4oMini.get_ai_mini_title(title)
                    if new_title:
                        item.add_video_info("video_title", new_title)
                        current_time = datetime.now()
                        formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
                        values = [
                            [
                                video_obj["video_url_list"][0]['video_url'],
                                video_obj["image_url_list"][0]['image_url'],
                                title,
                                new_title,
                                formatted_time,
                            ]
                        ]
                        FeishuUtils.insert_columns("U5dXsSlPOhiNNCtEfgqcm1iYnpf", "v8S6nL", "ROWS", 1, 2)
                        time.sleep(0.5)
                        FeishuUtils.update_values("U5dXsSlPOhiNNCtEfgqcm1iYnpf", "v8S6nL", "A2:Z2", values)
            self.download_cnt += 1
            self.mq.send_msg(mq_obj)
            # 保存视频ID
            self.redis_ops.save_video_id(video_obj['channel_content_id'])
            if self.download_cnt >= self.rule_dict.get("videos_cnt", {}).get("min", 300):
                # 记录轮训到的用户id
                self.redis_ops.set_last_scanned_id(self.last_scanned_id)
                self.limit_flag = False
            else:
                self.redis_ops.set_last_scanned_id(0)
                self.curses += 1
        except Exception as e:
            tb_info = traceback.format_exc()
            self.aliyun_log.logging(
                code="1005",
                message=f"处理视频对象时发生异常，错误信息: {str(e)}",
                data={"video_obj": video_obj}
            )
            self.LocalLog.error(f"处理视频对象时发生异常: {e}\n{tb_info}")

    async def run(self):
        """
        运行主流程，异步执行推荐视频和相关推荐视频的请求，直到达到下载数量限制。

        :return: 无返回值，程序运行的主逻辑
        """
        self.LocalLog.info("开始执行中青看点用户视频抓取...")
        await asyncio.gather(
            self.control_request_author()
        )


if __name__ == '__main__':
    asyncio.run(ZhongQingKanDianAuthor(
        platform="zhongqingkandian",
        mode="author",
        rule_dict={"videos_cnt": {"min": 2, "max": 0}},
        user_list=[{"uid": 81525568, "link": "中青看点推荐", "nick_name": "芸芸众生"}]
    ).run())