Update GAT.py

py/GAT.py

@@ -174,63 +174,72 @@ with open('gat.txt', 'w', encoding='utf-8') as new_file:
 
 
 
-import requests
+# 函数：获取视频分辨率
-from tqdm import tqdm
+def get_video_resolution(video_path, timeout=0.8):
-import threading
+    cap = cv2.VideoCapture(video_path)
-import re
+    if not cap.isOpened():
-def test_connectivity(url, max_attempts=2):
+        return None
-    video_formats = ["m3u", "/", "rtsp"]
+    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
-    if not any(re.search(keyword, url, re.I) for keyword in video_formats):
+    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
-        print("\n特殊网址: 跳过检测")
+    cap.release()
-        return False
+    return (width, height)
-    for _ in range(max_attempts):
+# 函数：处理每一行
-        try:
+def process_line(line, output_file, order_list, valid_count, invalid_count, total_lines):
-            response = requests.get(url, timeout=0.3)
+    parts = line.strip().split(',')
-            return response.status_code == 200
+    if '#genre#' in line:
-        except requests.RequestException:
+        # 如果行包含 '#genre#'，直接写入新文件
-            pass
-    return False
-# 处理每一行的函数
-def process_line(line, output_file, valid_count, invalid_count):
-    parts = line.strip().split(",")
-    if len(parts) == 2:
-        channel_name, channel_url = parts
-        if "genre" in line.lower():
-            with threading.Lock():
-                output_file.write("\n" +line)  # 直接写入原始行
-        elif test_connectivity(channel_url):
-            with threading.Lock():
-                output_file.write(f"{channel_name},{channel_url}\n")
-                valid_count[0] += 1
-        else:
-            with threading.Lock():
-                invalid_count[0] += 1
-    else:
         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:
+        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()
+
 # 主函数
 def main(source_file_path, output_file_path):
-    with open(source_file_path, "r", encoding="utf-8") as source_file:
+    order_list = []
-        lines = source_file.readlines()
     valid_count = [0]
     invalid_count = [0]
-    with open(output_file_path, "w", encoding="utf-8") as output_file:
+    task_queue = Queue()
-        threads = []
+    # 读取源文件
-        for line in tqdm(lines, desc="地址有效"):
+    with open(source_file_path, 'r', encoding='utf-8') as source_file:
-            thread = threading.Thread(target=process_line, args=(line, output_file, valid_count, invalid_count))
+        lines = source_file.readlines()
-            thread.start()
+    with open(output_file_path + '.txt', 'w', encoding='utf-8') as output_file:
-            threads.append(thread)
+        # 创建线程池
-        for thread in threads:
+        with ThreadPoolExecutor(max_workers=64) as executor:
-            thread.join()
+            # 创建并启动工作线程
-    print(f"任务完成,有效源数量: {valid_count[0]}, 无效源数量: {invalid_count[0]}")
+            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)}")
 if __name__ == "__main__":
-    try:
+    source_file_path = 'gat.txt'  # 替换为你的源文件路径
-        source_file_path = "gat.txt"
+    output_file_path = 'gat'  # 替换为你的输出文件路径,不要后缀名
-        output_file_path = "gat.txt"
+    main(source_file_path, output_file_path)
-        main(source_file_path, output_file_path)
-    except Exception as e:
-        print(f"程序发生错误: {e}")
-
 # 无需再打印酒店源，因为这里是对所有URL进行检测，而不是基于IP分组检测