From 6aedf44f59446bd9e9f29cbf355952853201f490 Mon Sep 17 00:00:00 2001 From: frxz751113 <156018267+frxz751113@users.noreply.github.com> Date: Thu, 3 Oct 2024 03:07:47 +0800 Subject: [PATCH] Update GAT.py --- py/GAT.py | 186 ++++++++++++++++++++++-------------------------------- 1 file changed, 75 insertions(+), 111 deletions(-) diff --git a/py/GAT.py b/py/GAT.py index d2649c3..0cc8569 100644 --- a/py/GAT.py +++ b/py/GAT.py @@ -207,120 +207,84 @@ with open('gat.txt', 'w', encoding='utf-8') as new_file: line = line.replace(old, new) new_file.write(line) ######################连通性检测 + import requests -import queue -import threading +import time +import cv2 +from urllib.parse import urlparse from tqdm import tqdm -def test_connectivity(url, max_attempts=2): - for _ in range(max_attempts): - try: - response = requests.head(url, timeout=3) - return response.status_code == 200 - except requests.RequestException: - pass - return False -def process_line(line, result_queue): - parts = line.strip().split(",") - if len(parts) == 2 and parts[1]: - channel_name, channel_url = parts - if test_connectivity(channel_url): - result_queue.put((channel_name, channel_url)) + +# 测试HTTP连接并尝试下载数据 +def test_connectivity_and_download(url, initial_timeout=1, retry_timeout=1): + parsed_url = urlparse(url) + if parsed_url.scheme not in ['http', 'https']: + # 非HTTP(s)协议，尝试RTSP检测 + return test_rtsp_connectivity(url, retry_timeout) else: - pass -def main(source_file_path, output_file_path): - with open(source_file_path, "r", encoding="utf-8") as source_file: - lines = source_file.readlines() - result_queue = queue.Queue() - threads = [] - for line in tqdm(lines, desc="检测进行中"): - thread = threading.Thread(target=process_line, args=(line, result_queue)) - thread.start() - threads.append(thread) - for thread in threads: - thread.join() - with open(output_file_path, "w", encoding="utf-8") as output_file: - while not result_queue.empty(): - item = result_queue.get() - if item[0] and item[1]: - output_file.write(f"{item[0]},{item[1]}\n") -if __name__ == "__main__": - try: - source_file_path = "gat.txt" - output_file_path = "gat.txt" - main(source_file_path, output_file_path) - except Exception as e: - print(f"程序发生错误: {e}") - - - - -########################## 函数：获取视频分辨率 -def get_video_resolution(video_path, timeout=0.8): - cap = cv2.VideoCapture(video_path) - if not cap.isOpened(): - return None - width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) - height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) - cap.release() - return (width, height) -# 函数：处理每一行 -def process_line(line, output_file, order_list, valid_count, invalid_count, total_lines): - parts = line.strip().split(',') - if '#genre#' in line: - # 如果行包含 '#genre#'，直接写入新文件 - with threading.Lock(): - output_file.write(line) - print(f"已写入genre行：{line.strip()}") - elif len(parts) == 2: - channel_name, channel_url = parts - resolution = get_video_resolution(channel_url, timeout=8) - if resolution and resolution[1] >= 720: # 检查分辨率是否大于等于720p - with threading.Lock(): - output_file.write(f"{channel_name}[{resolution[1]}p],{channel_url}\n") - order_list.append((channel_name, resolution[1], channel_url)) - valid_count[0] += 1 - print(f"Channel '{channel_name}' accepted with resolution {resolution[1]}p at URL {channel_url}.") - else: - invalid_count[0] += 1 - with threading.Lock(): - print(f"有效: {valid_count[0]}, 无效: {invalid_count[0]}, 总数: {total_lines}, 进度: {(valid_count[0] + invalid_count[0]) / total_lines * 100:.2f}%") -# 函数：多线程工作 -def worker(task_queue, output_file, order_list, valid_count, invalid_count, total_lines): - while True: + # HTTP(s)协议，使用原始方法 try: - line = task_queue.get(timeout=1) - process_line(line, output_file, order_list, valid_count, invalid_count, total_lines) - except Queue.Empty: - break - finally: - task_queue.task_done() + with requests.get(url, stream=True, timeout=initial_timeout) as response: + if response.status_code == 200: + start_time = time.time() + while time.time() - start_time < initial_timeout: + chunk = response.raw.read(512) # 尝试下载1KB数据 + if chunk: + return True # 成功下载数据 + except requests.RequestException as e: + print(f"请求异常: {e}") + pass #这行删掉则会在下载不到数据流的时候进行连通性测试 + + return False # 默认返回False + +print("/" * 80) + +# 测试RTSP连接并尝试读取流 +def test_rtsp_connectivity(url, timeout=3): + cap = cv2.VideoCapture(url) + if not cap.isOpened(): + return False + start_time = time.time() + while time.time() - start_time < timeout: + ret, _ = cap.read() + if ret: + return True # 成功读取帧 + cap.release() + return False # 主函数 -def main(source_file_path, output_file_path): - order_list = [] - valid_count = [0] - invalid_count = [0] - task_queue = Queue() - # 读取源文件 - with open(source_file_path, 'r', encoding='utf-8') as source_file: +def main(输入, 输出): + with open(输入, "r", encoding="utf-8") as source_file: lines = source_file.readlines() - with open(output_file_path + '.txt', 'w', encoding='utf-8') as output_file: - # 创建线程池 - with ThreadPoolExecutor(max_workers=64) as executor: - # 创建并启动工作线程 - for _ in range(128): - executor.submit(worker, task_queue, output_file, order_list, valid_count, invalid_count, len(lines)) - # 将所有行放入队列 - for line in lines: - task_queue.put(line) - # 等待队列中的所有任务完成 - task_queue.join() - print(f"任务完成，有效频道数：{valid_count[0]}, 无效频道数：{invalid_count[0]}, 总频道数：{len(lines)}") + + results = [] + for line_number, line in enumerate(tqdm(lines, desc="检测中")): + parts = line.strip().split(",") + if len(parts) == 2 and parts[1]: # 确保有URL，并且URL不为空 + channel_name, channel_url = parts + try: + is_valid = test_connectivity_and_download(channel_url) + except Exception as e: + print(f"检测URL {channel_url} 时发生错误: {e}") + is_valid = False # 将异常的URL视为无效 + + status = "有效" if is_valid else "无效" + + if "genre" in line.lower() or status == "有效": + results.append((channel_name.strip(), channel_url.strip(), status)) + + # 写入文件 + with open(输出, "w", encoding="utf-8") as output_file: + for channel_name, channel_url, status in results: + output_file.write(f"{channel_name},{channel_url}\n") + + print(f"任务完成, 有效源数量: {len([x for x in results if x[2] == '有效'])}, 无效源数量: {len([x for x in results if x[2] == '无效'])}") + if __name__ == "__main__": - source_file_path = 'gat.txt' # 替换为你的源文件路径 - output_file_path = 'gat' # 替换为你的输出文件路径,不要后缀名 - main(source_file_path, output_file_path) -# 无需再打印酒店源，因为这里是对所有URL进行检测，而不是基于IP分组检测 + 输入 = "gat.txt" #input('请输入utf-8编码的直播源文件路径:') + 输出 = "gat.txt" + main(输入, 输出) + + @@ -685,10 +649,10 @@ def remove_duplicates(input_file, output_file): print("去重后的行数：", len(output_lines)) # 使用方法 -remove_duplicates('去重.txt', 'GAT.txt') +remove_duplicates('去重.txt', 'gat.txt') # 打开文档并读取所有行 -with open('GAT.txt', 'r', encoding="utf-8") as file: +with open('gat.txt', 'r', encoding="utf-8") as file: lines = file.readlines() # 使用列表来存储唯一的行的顺序 @@ -720,7 +684,7 @@ for file in files_to_remove: else: # 如果文件不存在，则提示异常并打印提示信息 print(f"文件 {file} 不存在，跳过删除。") -print("任务运行完毕，GAT频道列表可查看文件夹内综合源.txt文件！") +print("任务运行完毕，gat频道列表可查看文件夹内综合源.txt文件！") def append_text_between_files(file1_path, file2_path): @@ -745,7 +709,7 @@ def append_text_between_files(file1_path, file2_path): combined_lines = unique_lines2 + unique_lines1 with open(file2_path, 'w', encoding='utf-8') as file2: file2.write('\n'.join(combined_lines)) -file_path1 = 'GAT.txt' +file_path1 = 'gat.txt' file_path2 = '综合源.txt' append_text_between_files(file_path1, file_path2) @@ -830,4 +794,4 @@ print("任务运行完毕，频道列表可查看文件夹内源.txt文件！") -print("任务运行完毕，GAT频道列表可查看文件夹内综合源.txt文件！") +print("任务运行完毕，gat频道列表可查看文件夹内综合源.txt文件！")