免费电视直播源的获取与使用

#EXTM3U x-tvg-url="https://epg.v1.mk/fy.xml" #EXTINF:-1,tvg-id="NOWSPORTS4K616" tvg-name="NOWSPORTS4K616" tvg-logo="https://epg.v1.mk/logo/NOWSPORTS4K616.png" group-title="其他",4K60PSDR-H264-AAC测试 https://cdn.jsdelivr.net/gh/feiyangdigital/testvideo/sdr4kvideo/index.m3u8 #EXTINF:-1,tvg-id="NOWSPORTS4K616" tvg-name="NOWSPORTS4K616" tvg-logo="https://epg.v1.mk/logo/NOWSPORTS4K616.png" group-title="其他",4K60PHLG-HEVC-EAC3测试 https://cdn.jsdelivr.net/gh/feiyangdigital/testvideo/hlg4kvideo/index.m3u8 #EXTINF:-1,tvg-id="CCTV1" tvg-name="CCTV1" tvg-logo="https://epg.v1.mk/logo/CCTV1.png" group-title="央视",CCTV1 http://ottrrs.hl.chinamobile.com/PLTV//224//index.m3u8

import os import glob import asyncio import aiohttp from urllib.parse import urlparse def is_url(string): try: result = urlparse(string) return all([result.scheme, result.netloc]) except ValueError: return False async def verify_url(session, url, semaphore): async with semaphore: try: async with session.head(url, timeout=10) as response: return response.status < 400 except Exception: return False def process_file(file): channels = [] with open(file, 'r', encoding='utf-8') as f: lines = f.readlines() i = 0 while i < len(lines): line = lines[i].strip() if line.startswith('#EXTINF'): title = line i += 1 while i < len(lines) and not is_url(lines[i].strip()): i += 1 if i < len(lines): url = lines[i].strip() if is_url(url): channels.append((title, url)) i += 1 return channels async def main(): m3u_files = glob.glob('*.m3u') all_channels = {} for file in m3u_files: channels = process_file(file) for title, url in channels: if title not in all_channels: all_channels[title] = set() # 使用集合存储不同的 URL all_channels[title].add(url) # 添加 URL 到集合中 print(f"Total unique channels found: {len(all_channels)}") valid_channels = [] semaphore = asyncio.Semaphore(50) # 限制并发数为50 async with aiohttp.ClientSession() as session: tasks = [] for title, urls in all_channels.items(): for url in urls: task = asyncio.ensure_future(verify_url(session, url, semaphore)) tasks.append((task, url, title)) for i, (task, url, title) in enumerate(tasks, 1): result = await task if result: valid_channels.append((title, url)) print(f"URL {i}/{len(tasks)} is valid: {url}") else: print(f"URL {i}/{len(tasks)} is not accessible: {url}") print(f"Total valid channels: {len(valid_channels)}") with open('merged_verified.m3u', 'w', encoding='utf-8') as f: f.write('#EXTM3U x-tvg-url="https://epg.v1.mk/fy.xml"\n') for title, url in valid_channels: f.write(f"{title}\n{url}\n") print("Merged and verified .m3u file has been created: merged_verified.m3u") if __name__ == "__main__": asyncio.run(main()) 

input_file = 'merged_verified.m3u' # 输入文件名 output_file = 'output.m3u' # 输出文件名 # 要删除的 group-title remove_groups = {"zonghe", "原创IP", "原创", "科技", "我的世界", "一起看", "4K频道", "少儿动画1",'IPTV少儿动画','韩国娱乐KMTV'} # 要删除的 IP 地址 remove_ip = "2409:8087:74F1:0021::0008" with open(input_file, 'r', encoding='utf-8') as infile, open(output_file, 'w', encoding='utf-8') as outfile: lines = infile.readlines() skip_next = False for line in lines: if line.startswith('#EXTINF'): # 检查 group-title 是否在要删除的列表中 if any(group in line for group in remove_groups): skip_next = True # 标记下一个链接行需要跳过 continue # 检查下一行是否包含要删除的 IP 地址 next_line_index = lines.index(line) + 1 if next_line_index < len(lines) and remove_ip in lines[next_line_index]: skip_next = True # 标记下一个链接行需要跳过 continue if skip_next: skip_next = False # 重置标记 continue outfile.write(line) # 写入保留的行 print("处理完成，结果保存在", output_file) 

#EXTINF:-1,tvg-id="天元围棋" tvg-name="天元围棋" group-title="我的",天元围棋 https://lv.weiqi.com.cn/live/5.m3u8