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()