Update iptv-check.py

This commit is contained in:
frxz751113
2024-08-10 03:47:18 +08:00
committed by GitHub
parent 069f5bee4b
commit 9f54e31200
+1 -77
View File
@@ -1,4 +1,4 @@
#本程序只适用于酒店源的检测,请勿移植他用
#本程序只对酒店源进行了720p以上分辨率过滤,IP段去重。组播和自定义源请自行从源文件过滤
import time
import concurrent.futures
from selenium import webdriver
@@ -25,7 +25,6 @@ from translate import Translator # 导入Translator类,用于文本翻译
# 搜素关键词:"iptv/live/zh_cn.js" && country="CN" && region="Hunan" && city="changsha"
# 搜素关键词:"ZHGXTV" && country="CN" && region="Hunan" && city="changsha"
#"isShowLoginJs"智能KUTV管理
#定义ZHGXTV采集地址
urls = [
#"https://fofa.info/result?qbase64=IlpIR1hUViIgJiYgcmVnaW9uPSJndWFuZ2Rvbmci",#广东
@@ -58,8 +57,6 @@ def is_url_accessible(url):
except requests.exceptions.RequestException:
pass
return None
results = []
for url in urls:
# 创建一个Chrome WebDriver实例
@@ -75,7 +72,6 @@ for url in urls:
page_content = driver.page_source
# 关闭WebDriver
driver.quit()
# 查找所有符合指定格式的网址
pattern = r"http://\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d+" # 设置匹配的格式,如http://8.8.8.8:8888
urls_all = re.findall(pattern, page_content)
@@ -97,7 +93,6 @@ for url in urls:
x_url = f"{base_url}{modified_ip}{port}"
x_urls.append(x_url)
urls = set(x_urls) # 去重得到唯一的URL列表
valid_urls = []
# 多线程获取可用url
with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
@@ -135,7 +130,6 @@ for url in urls:
else:
urld = (f"{urls[0]}//{url_data[2]}")
print(f"{name},{urld}")
if name and urld:
name = name.replace("高清电影", "影迷电影")
name = name.replace("中央", "CCTV")
@@ -256,7 +250,6 @@ with open("iptv.txt", 'w', encoding='utf-8') as file:
file.write(result + "\n")
print(result)
print("频道列表文件iptv.txt获取完成!")
for line in fileinput.input("iptv.txt", inplace=True): #打开文件,并对其进行关键词原地替换
line = line.replace("河南河南", "河南")
line = line.replace("河南河南", "河南")
@@ -276,7 +269,6 @@ for line in fileinput.input("iptv.txt", inplace=True): #打开文件,并对
line = line.replace("都市生活", "都市")
line = line.replace("都市生活6", "都市")
print(line, end="") #设置end="",避免输出多余的换行符
#定义智慧桌面采集地址
urls = [
#"https://fofa.info/result?qbase64=ImlwdHYvbGl2ZS96aF9jbi5qcyIgJiYgcG9ydD0iMTExMSI%3D", # 1111
@@ -516,8 +508,6 @@ with open("iptv.txt", 'a', encoding='utf-8') as file: #打开文本以
file.write(result + "\n")
print(result)
print("频道列表文件iptv.txt追加写入成功!")
#这里排序IP段去重放在了原文件头
#这里排序IP段去重放在了原文件头
#这里排序IP段去重放在了原文件头
@@ -525,9 +515,6 @@ print("频道列表文件iptv.txt追加写入成功!")
#这里排序IP段去重放在了原文件头
#这里排序IP段去重放在了原文件头
#这里排序IP段去重放在了原文件头
###################################################去除列表中的组播地址以及CCTV和卫视
def filter_lines(input_file, output_file):
with open(input_file, 'r', encoding='utf-8') as file:
@@ -542,8 +529,6 @@ def filter_lines(input_file, output_file):
filter_lines("iptv.txt", "iptv.txt")
print(f"文件已过滤完成")
########################################################################################################################################################################################
########################################################################################################################################################################################
##########################################################IP段去重
import re
@@ -579,7 +564,6 @@ input_file_path = 'iptv.txt'
output_file_path = 'iptv.txt'
deduplicate_lines(input_file_path, output_file_path)
################################################################################
#################################################### 对整理好的频道列表测试HTTP连接
# 函数:获取视频分辨率
def get_video_resolution(video_path, timeout=0.8):
@@ -590,7 +574,6 @@ def get_video_resolution(video_path, timeout=0.8):
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(',')
@@ -612,7 +595,6 @@ def process_line(line, output_file, order_list, valid_count, invalid_count, tota
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:
@@ -623,34 +605,27 @@ def worker(task_queue, output_file, order_list, valid_count, invalid_count, tota
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 open(source_file_path, 'r', encoding='utf-8') as source_file:
lines = source_file.readlines()
with open(output_file_path, 'w', encoding='utf-8') as output_file:
# 创建线程池
with ThreadPoolExecutor(max_workers=64) as executor:
# 创建并启动工作线程
for _ in range(64):
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 = 'iptv.txt' # 替换为你的源文件路径
output_file_path = '检测结果.txt' # 替换为你的输出文件路径,不要后缀名
@@ -874,24 +849,18 @@ def check_and_write_file(input_file, output_file, keywords):
check_and_write_file('酒店源.txt', 'a0.txt', keywords="央视频道, 8K, 4K, 4k")
check_and_write_file('酒店源.txt', 'a.txt', keywords="央视频道, CCTV, 8K, 4K, 爱上4K, 纯享, 风云剧场, 怀旧剧场, 影迷, 高清电影, 动作电影, 每日影院, 全球大片, 第一剧场, 家庭影院, 影迷电影, 星光, 华语, 美国大片, 峨眉")
check_and_write_file('酒店源.txt', 'a1.txt', keywords="央视频道, 风云音乐, 女性时尚, 地理世界, 音乐现场")
check_and_write_file('酒店源.txt', 'b.txt', keywords="卫视频道, 卫视, 凤凰, 星空")
check_and_write_file('酒店源.txt', 'c.txt', keywords="影视频道, 爱情喜剧, 爱喜喜剧, 风云剧场, 怀旧剧场, 影迷, 高清电影, 动作电影, 每日影院, 全球大片, 第一剧场, 家庭影院, 影迷电影, 星光, 华语, 美国大片, 峨眉, \
电影, 惊嫊悬疑, 东北热剧, 无名, 都市剧场, iHOT, 剧场, 欢笑剧场, 重温经典, 明星大片, 中国功夫, 军旅, 东北热剧, 中国功夫, 军旅剧场, 古装剧场, \
家庭剧场, 惊悚悬疑, 欢乐剧场, 潮妈辣婆, 爱情喜剧, 精品大剧, 超级影视, 超级电影, 黑莓动画, 黑莓电影, 海外剧场, 精彩影视, 无名影视, 潮婆辣妈, 超级剧, 热播精选")
check_and_write_file('酒店源.txt', 'c1.txt', keywords="影视频道, 求索动物, 求索, 求索科学, 求索记录, 爱谍战, 爱动漫, 爱科幻, 爱青春, 爱自然, 爱科学, 爱浪漫, 爱历史, 爱旅行, 爱奇谈, 爱怀旧, 爱赛车, 爱都市, 爱体育, 爱经典, \
爱玩具, 爱喜剧, 爱悬疑, 爱幼教, 爱院线")
check_and_write_file('酒店源.txt', 'c2.txt', keywords="影视频道, 军事评论, 农业致富, 哒啵赛事, 怡伴健康, 武博世界, 超级综艺, 哒啵, HOT, 炫舞未来, 精品体育, 精品萌宠, 精品记录, 超级体育, 金牌, 武术世界, 精品纪录")
check_and_write_file('酒店源.txt', 'd.txt', keywords="少儿频道, 少儿, 卡通, 动漫, 宝贝, 哈哈")
check_and_write_file('酒店源.txt', 'e.txt', keywords="港澳频道, TVB, 珠江台, 澳门, 龙华, 广场舞, 动物杂技, 民视, 中视, 华视, AXN, MOMO, 采昌, 耀才, 靖天, 镜新闻, 靖洋, 莲花, 年代, 爱尔达, 好莱坞, 华丽, 非凡, 公视, \
寰宇, 无线, EVEN, MoMo, 爆谷, 面包, momo, 唐人, 中华小, 三立, CNA, FOX, RTHK, Movie, 八大, 中天, 中视, 东森, 凤凰, 天映, 美亚, 环球, 翡翠, 亚洲, 大爱, 大愛, 明珠, 半岛, AMC, 龙祥, 台视, 1905, 纬来, 神话, 经典都市, 视界, \
番薯, 私人, 酒店, TVB, 凤凰, 半岛, 星光视界, 大愛, 新加坡, 星河, 明珠, 环球, 翡翠台")
check_and_write_file('酒店源.txt', 'f.txt', keywords="省市频道, 湖北, 武汉, 河北, 广东, 河南, 陕西, 四川, 湖南, 广西, 石家庄, 南宁, 汕头, 揭阳, 普宁, 福建, 辽宁")
check_and_write_file('酒店源.txt', 'o1.txt', keywords="其他频道, 新闻, 综合, 文艺, 电视, 公共, 科教, 教育, 民生, 轮播, 套, 法制, 文化, 经济, 生活")
check_and_write_file('酒店源.txt', 'o.txt', keywords="其他频道, , ")
#
@@ -909,7 +878,6 @@ for file_path in file_paths:
with open("去重.txt", "w", encoding="utf-8") as output:
output.write('\n'.join(file_contents))
#
##################################################################### 打开文档并读取所有行 ,对提取后重复的频道去重
with open('去重.txt', 'r', encoding="utf-8") as file:
lines = file.readlines()
@@ -934,42 +902,6 @@ for file in files_to_remove:
print("任务运行完毕,酒店源频道列表可查看文件夹内txt文件!")
# 获取rtp目录下的文件名
files = os.listdir('rtp')
files_name = []
@@ -1207,9 +1139,6 @@ print('提取完成,结果已保存到:', output_file_path)
url = "https://raw.githubusercontent.com/frxz751113/AAAAA/main/TW.txt" #源采集地址
r = requests.get(url)
open('港澳.txt','wb').write(r.content) #打开源文件并临时写入
#
#从整理好的文本中按类别进行特定关键词提取#
with open('酒店源.txt', 'r', encoding='utf-8') as f: #打开文件,并对其进行关键词提取 #
@@ -1223,9 +1152,6 @@ with open('酒店源.txt', 'r', encoding='utf-8') as f: #打开文件,并对
if re.search(pattern, line): # 如果行中有任意关键字 #
b.write(line) # 将该行写入输出文件 #
#
#从整理好的文本中进行特定关键词替换以规范频道名#
for line in fileinput.input("b.txt", inplace=True): #打开文件,并对其进行关键词原地替换 #
#line = line.replace("央视频道,#genre#", "") #
@@ -1312,8 +1238,6 @@ with open(deleted_lines_file_path, 'w', encoding='utf-8') as file:
file.writelines(deleted_lines)
print('过滤完成,结果已保存到:', output_file_path)
print('删除的行已保存到:', deleted_lines_file_path)
#
#从文本中截取少儿段并生成两个新文件#
# 定义关键词