Update iptv流畅度检测.py

py/iptv流畅度检测.py

@@ -715,7 +715,7 @@ with open(output_file_path, 'w', encoding='utf-8') as output_file:
                 start_time = time.time()
                 frame_count = 0
                 # 尝试捕获5秒内的帧
-                while frame_count < 72 and (time.time() - start_time) < 3:#//////////////////////////////////////////////////////////////////////////////////////###########
+                while frame_count < 66 and (time.time() - start_time) < 3:#//////////////////////////////////////////////////////////////////////////////////////###########
                     ret, frame = cap.read()
                     if not ret:
                         break
@@ -723,7 +723,7 @@ with open(output_file_path, 'w', encoding='utf-8') as output_file:
                 # 释放资源
                 cap.release()
                 # 根据捕获的帧数判断状态并记录结果#////////////////////////////////////////////////////////////////////////////////////////////////////////////////###########
-                if frame_count >= 72:  #5秒内超过100帧则写入#/////////////////////////////////////////////////////////////////////////////////////////////////////###########
+                if frame_count >= 66:  #5秒内超过100帧则写入#/////////////////////////////////////////////////////////////////////////////////////////////////////###########
                     detected_ips[ip_key] = {'status': 'ok'}
                     output_file.write(line)  # 写入检测通过的行
                 else:
@@ -734,141 +734,6 @@ for ip_key, result in detected_ips.items():
 
 
 
-##########################################################IP段去重,保留最后一个IP段，防止高峰拥堵，也减少不必要的检测行
-import re
-def deduplicate_lines(input_file_path, output_file_path):
-    seen_combinations = {}
-    unique_lines = []
-    with open(input_file_path, 'r', encoding='utf-8') as file:
-        for line in file:
-            # 使用正则表达式查找行中的所有URL，并捕获IP地址、端口号和端口号之后的部分
-            urls = re.findall(r'http://([\d.]+):(\d+)(/.*)?', line)
-            # 为每个URL生成一个去重键
-            for full_url in urls:
-                ip, port, path = full_url
-                ip_parts = ip.split('.')
-                if len(ip_parts) < 3:
-                    continue
-                # 使用IP的前三个字段和端口号之后的部分生成去重键
-                combination_key = f"{ip_parts[0]}.{ip_parts[1]}.{ip_parts[2]}-{port}-{path or ''}"
-                # 检查这个组合是否已经出现过
-                if combination_key not in seen_combinations:
-                    # 如果没有出现过，记录当前行和去重键
-                    seen_combinations[combination_key] = line.strip()
-                else:
-                    # 如果已经出现过，更新为最后出现的行
-                    seen_combinations[combination_key] = line.strip()
-    # 将去重后的所有唯一行写入新文件
-    with open(output_file_path, 'w', encoding='utf-8') as file:
-        for line in seen_combinations.values():
-            file.write(line + '\n')
-print(f"IP段去重完成")            
-# 调用函数
-input_file_path = '酒店优选.txt'
-output_file_path = '酒店优选.txt'
-deduplicate_lines(input_file_path, output_file_path)
-################################################################################
-
-
-from concurrent.futures import ThreadPoolExecutor
-from tqdm import tqdm
-from pypinyin import lazy_pinyin
-from opencc import OpenCC
-import base64
-import cv2
-#################################################### 对整理好的频道列表测试HTTP连接
-# 函数：获取视频分辨率#$
-def get_video_resolution(video_path, timeout=2):#$
-    # 使用OpenCV创建视频捕获对象#$
-    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(',')#$
-    # 如果行包含特定的标签'#genre#'，则直接写入新文件#$
-    if '#genre#' in line:#$
-        with threading.Lock():  # 使用线程锁保证写入操作的原子性#$
-            output_file.write(line)#$
-            print(f"已写入genre行：{line.strip()}")#$
-    # 如果分割后的部分数量为2，则继续处理#$
-    elif len(parts) == 2:#$
-        channel_name, channel_url = parts#$
-        # 获取视频的分辨率#$
-        resolution = get_video_resolution(channel_url, timeout=2)#$
-        # 如果分辨率有效且高度大于等于720p#$
-        if resolution and resolution[1] >= 720:#$
-            with threading.Lock():  # 使用线程锁#$
-                output_file.write(f"{channel_name}[{resolution[1]}p],{channel_url}\n")#$
-                # 将频道名、分辨率和URL添加到列表中#$
-                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:#$
-            # 从队列中获取任务，超时时间为1秒#$
-            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):#$
-    # 初始化列表和计数器#$
-    order_list = []#$
-    valid_count = [0]#$
-    invalid_count = [0]#$
-    task_queue = Queue()#$
-    # 使用with语句打开源文件并读取所有行#$
-    with open(source_file_path, 'r', encoding='utf-8') as source_file:#$
-        lines = source_file.readlines()#$
-    # 使用with语句打开输出文件准备写入#$
-    with open(output_file_path, 'w', encoding='utf-8') as output_file:#$
-        # 创建线程池，最大工作线程数为64#$
-        with ThreadPoolExecutor(max_workers=264) as executor:#$
-            # 为线程池中的每个线程提交worker函数#$
-            for _ in range(264):#$
-                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__":#$
-    # 定义源文件和输出文件的路径#$
-    source_file_path = '酒店优选.txt'  # 替换为你的源文件路径#$
-    output_file_path = '酒店优选.txt'  # 替换为你的输出文件路径#$
-    # 调用主函数#$
-    main(source_file_path, output_file_path)#$
 
 
 ####################### 提示用户输入文件名（拖入文件操作）打开用户指定的文件对不规范频道名再次替换