import random from datetime import datetime import threading import time import math import requests from global_function_construct import definition_function from config import global_Configuration from logger_config import CreateLoggerHandler log_path = f"{global_Configuration['root_path']}/error_file/qie_video.log" logger_handler = CreateLoggerHandler().setup_logger(log_path=log_path) class PaserGetQieVideo(definition_function): """ 初始化方法，用于设置直播间的相关参数和请求头信息 参数: room_id: 直播间ID，用于构建请求URL proxy: 可选的代理服务器地址，用于网络请求 """ def __init__(self, procedures_id=2): self.projectVideo_id = procedures_id self.server_name = 'http://8.129.239.64' self.room_form = f"{self.projectVideo_id}" + "_video:{}" super().__init__() self.redis_server = self.create_connect_redis(True, redis_db=self.redis_database_config['redis_db'], host=self.redis_database_config['redis_host']) self.qie_config = self.global_Configuration.get('qie_config') self.domainName = self.qie_config.get('domainName') # todo 请求获取视频链接 def requests_get_flv_url(self, room_id, ua=None, ua_version=None, request_session=requests.Session()): try: t = math.ceil(time.time() * 1000 / 60000 - 1) # 时间 result = self.md5_encrypt(f"aWR6ZWly{room_id}RgP7DW01naDYQSV0{t}") headers = { "accept": "application/json, text/plain, */*", "accept-language": "zh-CN,zh;q=0.9,ja;q=0.8", "cache-control": "no-cache", "pragma": "no-cache", "priority": "u=1, i", "referer": f"{self.domainName}/{room_id}", "sec-ch-ua": f'"Google Chrome";v="{ua_version or self.ua_version}", "Chromium";v="{ua_version or self.ua_version}", "Not.A/Brand";v="24"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": "\"Linux\"", "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "token;": "", "user-agent": ua or self.ua} url = f"{self.domainName}/swf_api/web_room/{room_id}" params = {"cdn": "ws", "nofan": "yes", "_t": t, "sign": result} response = self.send_requests(request_session, url, headers=headers, params=params, data_type=1, impersonate=False) data = response.get('data') rtmp_live = data.get('rtmp_live') rtmp_url = data.get('rtmp_url') flv_url = f"{rtmp_url}/{rtmp_live}" if rtmp_live: return flv_url except Exception as err: logger_handler.error(f"requests_get_flv_url >>>>>>获取flv_url失败：{err}") print(f"requests_get_flv_url >>>>>>获取flv_url失败：{err}") def get_flv_url_configuration(self, room_id): ua, ua_version = self.update_ua() flv_headers = { "accept": "*/*", "accept-language": "zh-CN,zh;q=0.9,ja;q=0.8", "cache-control": "no-cache", "origin": self.domainName, "pragma": "no-cache", "priority": "u=1, i", "referer": f"{self.domainName}/{room_id}", "sec-ch-ua": f'"Google Chrome";v="{ua_version or self.ua_version}", "Chromium";v="{ua_version or self.ua_version}", "Not.A/Brand";v="24"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": "\"Linux\"", "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "cross-site", "user-agent": ua or self.ua} request_session = requests.Session() proxy = self.getProxy(tunnel_mode=self.global_Configuration.get('tunnel_mode'), type_=False, proxy_list=self.proxies_list, logger=logger_handler) if proxy: request_session.proxies = {'http': proxy, 'https': proxy} flv_url = self.requests_get_flv_url(room_id, ua=ua, ua_version=ua_version, request_session=request_session) request_session.close() return flv_url, flv_headers # todo 获取视频链接进行推流 def flv_pushTheFlow_rtmp(self, room_key, quality=2000): video_key = room_key.replace("room", "video") try: video_data = self.redis_server.hgetall(room_key) or {} room_id = video_data.get('videoId') server_ip = video_data.get("origin") or self.server_name rtmp_url = f"rtmp://{self.server_name.split('//')[-1]}:1935/live/{room_id}" for _ in range(5): flv_url, flv_headers = self.get_flv_url_configuration(room_id) """主函数""" if flv_url: cmd_list = self.build_ffmpeg_cmd(rtmp_url, flv_url, flv_headers) # 配置参数 video_dict = {"rtcUrl": f"{server_ip}:1985/rtc/v1/whep/?app=live&stream={room_id}", 'flvUrl': f"{server_ip}/live/{room_id}.flv", 'm3u8Url': f"{server_ip}/live/{room_id}.m3u8", "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "pixels_list": ''} if self.execution_cmd(self.redis_server, video_key, cmd_list, video_dict, logger_handler=logger_handler): self.redis_server.delete(video_key) break else: self.video_state.update({video_key: "获取 flv_url 失败"}) self.send_email(f"{self.projectVideo_id}--flv_pushTheFlow_rtmp>>>>>>>>>>{str(room_id)}", logger_handler=logger_handler) if self.redis_server.exists(room_key): self.redis_server.delete(room_key) except Exception as err: logger_handler.exception(f"flv_pushTheFlow_rtmp>>>{video_key}>>>>>>>>>{err}") self.send_email(f"{self.projectVideo_id}--flv_pushTheFlow_rtmp>>>>{video_key}>>>>>>{str(err)}", logger_handler=logger_handler) self.video_state.update({video_key: f"flv_pushTheFlow_rtmp>>>{video_key}>>>>>>>>>{str(err)}"}) self.redis_server.delete(video_key) if self.redis_server.exists(room_key): self.redis_server.delete(room_key) if self.thread_statistics.get(room_key): del self.thread_statistics[room_key] if self.video_state.get(video_key): del self.video_state[video_key] # 获取房间id进行推流 def allocation_pushTheFlow(self, quality=500): while True: try: if not self.check_redis_connection(self.redis_server): # 判断redis是否断开进行重连接 self.redis_server = self.create_connect_redis(True, redis_db=self.redis_database_config['redis_db'], host=self.redis_database_config['redis_host']) self.global_Configuration = self.redis_server.get_json("global_Configuration") self.qie_config = self.global_Configuration['qie_config'] self.server_name = self.qie_config['server_name'] redis_server_keys = self.redis_server.keys(f'{self.projectVideo_id}_room:*') for room_key in redis_server_keys: if not self.redis_server.exists(room_key.replace("room", "video")): target = threading.Thread(target=self.flv_pushTheFlow_rtmp, args=(room_key, quality), daemon=True) target.start() self.thread_statistics.update({room_key: target}) # target.join() self.redis_server.set_json(f"{self.projectVideo_id}:PaserGetQieVideo", {"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "thread_num": len(self.thread_statistics), "thread": list(self.thread_statistics.keys())}) self.redis_server.set_json(f"{self.projectVideo_id}:GetHuYaVideo_state", {"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "data_num": len(self.video_state), 'data': self.video_state}) if random.randint(1, 20) in [2, 4, 6, 8]: print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "_______________", redis_server_keys) except Exception as err: logger_handler.exception(f"allocation_pushTheFlow>>>>>>>>>>{str(err)}") self.send_email(f"{self.projectVideo_id}--allocation_pushTheFlow>>>>>>>>>>{str(err)}", logger_handler=logger_handler) time.sleep(2) # todo 获取视频列表 def run_main(self): self.print_state = True self.allocation_pushTheFlow() # self.flv_pushTheFlow_rtmp(10207092, "0") if __name__ == '__main__': PaserGetQieVideo().run_main()