automatic_crawler/spider/crawler_online/tiantianjufuqi.py

import os
import random
import sys
import time
import uuid
import json

from datetime import datetime

import requests

from application.common import Feishu
from application.common.feishu import FsData
from application.common.feishu.feishu_utils import FeishuUtils
from application.common.ffmpeg.ffmpeg_utils import Ffmpeg
from application.common.gpt import GPT4oMini

sys.path.append(os.getcwd())

from application.items import VideoItem
from application.pipeline import PiaoQuanPipeline
from application.common.messageQueue import MQ
from application.common.proxies import tunnel_proxies
from application.common.log import AliyunLogger
from application.common.mysql import MysqlHelper



class TTJFFQRecommend(object):
    """
    福气矩阵-天天聚福气
    """

    def __init__(self, platform, mode, rule_dict, user_list, env="prod"):
        self.limit_flag = False
        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.mysql = MysqlHelper(mode=self.mode, platform=self)

    def get_recommend_list(self):
        if self.expire_flag:
            self.aliyun_log.logging(
                code="2000",
                message="本轮已经抓取到足够的数据，自动退出\t{}".format(self.download_cnt),
            )
            return
        """
        获取推荐页视频
        """
        headers = {
            'Host': 'api.xinghetime.com',
            'xweb_xhr': '1',
            'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36 MicroMessenger/6.8.0(0x16080000) NetType/WIFI MiniProgramEnv/Mac MacWechat/WMPF MacWechat/3.8.6(0x13080610) XWEB/1156',
            'content-type': 'application/json',
            'accept': '*/*',
            'sec-fetch-site': 'cross-site',
            'sec-fetch-mode': 'cors',
            'sec-fetch-dest': 'empty',
            'referer': 'https://servicewechat.com/wxa12a841184757478/7/page-frame.html',
            'accept-language': 'zh-CN,zh;q=0.9'
        }
        data_rule = FsData()
        title_rule = data_rule.get_title_rule()
        while True:
            time.sleep(random.randint(1, 10))
            url = "https://api.xinghetime.com/luckvideo/video/getRecommendVideos"
            payload = json.dumps({
                "baseParam": {
                    "mid": "openid_oeBeO4lCgcJCey9JEJm9ZHPnOln8",
                    "pageSource": "video-home",
                    "appType": 2
                },
                "bizParam": {
                    "pageSize": 10
                }
            })

            response = requests.request("POST", url, headers=headers, data=payload)
            for index, video_obj in enumerate(response.json()['data'], 1):
                try:
                    self.aliyun_log.logging(
                        code="1001", message="扫描到一条视频", data=video_obj
                    )
                    self.process_video_obj(video_obj,title_rule)
                except Exception as e:
                    self.aliyun_log.logging(
                        code="3000",
                        message="抓取单条视频失败,第{}条报错原因是{}".format(
                             index, e
                        ),
                    )
            if self.limit_flag:
                return
            time.sleep(random.randint(5, 10))

    def process_video_obj(self, video_obj, title_rule):
        """
        处理视频
        :param video_obj:
        """
        time.sleep(random.randint(3, 8))
        trace_id = self.platform + str(uuid.uuid1())
        our_user = random.choice(self.user_list)
        play_str = video_obj["playCountFormat"]
        play_cnt = int(play_str.replace("+", "").replace("次播放", ""))
        item = VideoItem()
        item.add_video_info("video_id", video_obj["videoId"])
        item.add_video_info("video_title", video_obj["title"])
        item.add_video_info("play_cnt", play_cnt)
        item.add_video_info("publish_time_stamp", int(time.time()))
        item.add_video_info("out_user_id", video_obj["videoId"])
        item.add_video_info("cover_url", video_obj["coverImagePath"])
        item.add_video_info("like_cnt", 0)
        item.add_video_info("video_url", video_obj["videoPath"])
        item.add_video_info("out_video_id", video_obj["videoId"])
        item.add_video_info("platform", self.platform)
        item.add_video_info("strategy", self.mode)
        item.add_video_info("session", "{}-{}".format(self.platform, int(time.time())))
        item.add_video_info("user_id", our_user["uid"])
        item.add_video_info("user_name", our_user["nick_name"])
        # 获取当前时间
        current_time = datetime.now()
        formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
        values = [[
            video_obj["videoId"],
            formatted_time,
            video_obj["title"],
            video_obj["coverImagePath"],
            video_obj["videoPath"],
            play_cnt
        ]]
        Feishu.insert_columns(self.platform, 'fuxiaoshun', "MZrtFR", "ROWS", 1, 2)
        time.sleep(0.5)
        Feishu.update_values(self.platform, 'fuxiaoshun', "MZrtFR", "A2:Z2", values)
        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():
            video_url = video_obj["videoPath"]
            ffmpeg = Ffmpeg()
            new_video_url = ffmpeg.merge_m3u8(video_url)
            if not new_video_url:
                return
            item.add_video_info("video_url", new_video_url)
            title_list = 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["videoPath"],
                            video_obj["coverImagePath"],
                            title,
                            new_title,
                            formatted_time,
                        ]
                    ]
                    FeishuUtils.insert_columns("U5dXsSlPOhiNNCtEfgqcm1iYnpf", "ftQdRy", "ROWS", 1, 2)
                    time.sleep(0.5)
                    FeishuUtils.update_values("U5dXsSlPOhiNNCtEfgqcm1iYnpf", "ftQdRy", "A2:Z2", values)
            self.download_cnt += 1
            self.mq.send_msg(mq_obj)
            self.aliyun_log.logging(code="1002", message="成功发送至 ETL", data=mq_obj)
            if self.download_cnt >= int(
                    self.rule_dict.get("videos_cnt", {}).get("min", 200)
            ):
                self.limit_flag = True




    def run(self):
        self.get_recommend_list()


if __name__ == '__main__':
    J = TTJFFQRecommend(
        platform="tiantianjufuqi",
        mode="recommend",
        rule_dict={},
        user_list=[{'uid': "123456", 'nick_name': "xiaoxiao"}],

    )
    J.get_recommend_list()