import os import sys import json import random import uuid import time import traceback from datetime import datetime,timedelta import requests 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 ZhongQingKanDianRecommend: 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.session = requests.session() def __del__(self): if self.session: self.LocalLog.info("session 被正确关闭") self.session.close() def send_request(self, path, data): """发送带重试机制的API请求""" for attempt in range(self.MAX_RETRIES): try: response = self.session.post( f"{self.API_BASE_URL}{path}", data=data, timeout=self.TIMEOUT, headers=self.COMMON_HEADERS ) resp_data = response.json() # 检查响应格式 if 'code' not in resp_data: self.LocalLog.warning(f"{path}响应缺少code字段，尝试重试") raise ValueError("Missing 'code' in response") code = resp_data['code'] # 成功情况 (code=0) if code == 0: self.LocalLog.info(f"{path}请求成功：{resp_data}") return resp_data # 特定错误码不重试 if code == 29036: self.LocalLog.warning(f"{path}返回code:29036，消息：{resp_data}") return None # 其他错误码重试 self.LocalLog.warning(f"{path}返回错误码{code}，尝试重试,响应内容：{resp_data}") except Exception as e: tb_info = traceback.format_exc() self.LocalLog.error(f"{path}请求异常: {str(e)} \n {tb_info}") if attempt < self.MAX_RETRIES - 1: time.sleep(random.randint(5, 10)) # 所有重试失败，记录错误并返回None self.LocalLog.error(f"{path}达到最大重试次数") self.aliyun_log.logging( code="3000", message=f"请求 {path} 失败，达到最大重试次数", data=data ) return None def req_recommend_list(self): """ 同步请求推荐视频列表。 :return: 推荐视频列表的有效响应数据，如果请求失败则返回 None """ try: url = '/crawler/zhong_qing_kan_dian/recommend' body = json.dumps({"cursor": ""}) self.LocalLog.info(f"开始请求推荐{body}") resp = self.send_request(url, body) return resp except Exception as e: tb_info = traceback.format_exc() self.aliyun_log.logging( code="1003", message=f"请求推荐视频列表时发生异常，错误信息: {str(e)}\n{tb_info}", data={"url": url} ) self.LocalLog.info(f"请求推荐视频列表 {url} 时发生异常:{str(e)} \n{tb_info}") return None def req_detail(self, content_link, **kwargs): """ 同步请求视频详情。 :param content_link: 视频内容链接 :param label: 视频标签（如 "recommend" 或 "related"） :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 = self.send_request(url, body) if not resp: 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) self.process_video_obj(data) 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}") def control_request_recommend(self): """ 控制推荐视频列表的请求和处理流程。 :return: 无返回值，根据下载数量限制控制流程 """ while self.limit_flag: try: self.download_cnt = self.db_ops.get_today_videos() if self.download_cnt >= self.rule_dict.get("videos_cnt", {}).get("min", 100): self.aliyun_log.logging( code="2010", message=f"今日已经达到最大量", data=self.download_cnt ) self.LocalLog.info(f"当日视频已达到最大爬取量{self.download_cnt}") return if is_less_than_30_minutes(): self.LocalLog.info("时间已不足，停止执行") return self.LocalLog.info(f"开始推荐视频列表的请求和处理流程,今日已爬推荐 {self.download_cnt} 个视频") recommend_resp = self.req_recommend_list() if not recommend_resp: continue recommend_list = recommend_resp.get("data", {}).get("data", []) self.LocalLog.info(f"获取的推荐列表长度：{len(recommend_list)}") for video_obj in recommend_list: # if not self.limit_flag: # self.LocalLog.info(f"今日视频数量已达最大量{self.download_cnt}") # return content_link = video_obj.get("share_url") content_id = video_obj.get("id") self.LocalLog.info(f"content_link == {content_link} \n content_id == {content_id}") if not (content_link and content_id): continue # 当前内容id保存到redis self.redis_ops.save_recommend_video(content_id) self.req_detail(content_link, **video_obj) time.sleep(random.randint(5, 30)) except Exception as e: tb_info = traceback.format_exc() self.aliyun_log.logging( code="3008", message=f"控制推荐视频请求和处理时发生异常，错误信息: {str(e)}", data={} ) self.LocalLog.info(f"控制推荐视频请求和处理时发生异常:\n{tb_info}") self.LocalLog.info(f"循环结束，当前 limit_flag 值为: {self.limit_flag}") 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"] # 插入用户信息，使用异步方法并等待结果 self.db_ops.insert_user(account_id, account_name, account_avatar) 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", self.convert_number(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", int(video_obj['collect_num'])) item.add_video_info("share_cnt", int(video_obj["share_num"])) item.add_video_info("comment_cnt", int(video_obj["cmt_num"])) 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.mq.send_msg(mq_obj) self.download_cnt += 1 self.aliyun_log.logging( code="2009", message=f"成功发送视频到etl", data={"video_obj": video_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", 100): self.aliyun_log.logging( code="2010", message=f"今日已经达到最大量", data=self.download_cnt ) self.LocalLog.info("当日视频已达到最大爬取量") self.limit_flag = False 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}") def convert_number(self, s): if not isinstance(s, str): return s try: return float(s.strip('万')) * 10000 if '万' in s else int(s) except ValueError: self.LocalLog.info(f"无法将 '{s}' 转换为有效的数字。") def run(self): """ 运行主流程，执行推荐视频和相关推荐视频的请求，直到达到下载数量限制。 :return: 无返回值，程序运行的主逻辑 """ self.LocalLog.info("开始执行中青看点推荐抓取...") self.control_request_recommend() def is_less_than_30_minutes(): now = datetime.now() tomorrow = now.date() + timedelta(days=1) midnight = datetime.combine(tomorrow, datetime.min.time()) time_left = midnight - now return time_left.total_seconds() < 30 * 60 if __name__ == '__main__': ZhongQingKanDianRecommend( platform="zhongqingkandian", mode="recommend", rule_dict={'videos_cnt': {'min': 2, 'max': 0}, 'duration': {'min': 30, 'max': 1200}}, user_list=[{"uid": 81522822, "link": "中青看点推荐", "nick_name": "免不了俗"}] ).run()