import re import requests from collections import OrderedDict from datetime import datetime from concurrent.futures import ThreadPoolExecutor, as_completed tv_urls = [ "https://qu.ax/vUBde.txt", "http://bxtv.3a.ink/live.m3u", "http://live.nctv.top/x.txt", "https://aktv.space/live.m3u", "http://tot.totalh.net/tttt.txt", "https://m3u.ibert.me/fmml_ipv6.m3u", "https://json.doube.eu.org/XingHuo.txt", "https://raw.githubusercontent.com/zwc456baby/iptv_alive/master/live.txt", "https://raw.githubusercontent.com/zwc456baby/iptv_alive/master/live.m3u", "https://raw.githubusercontent.com/BurningC4/Chinese-IPTV/master/TV-IPV4.m3u", "https://raw.githubusercontent.com/YueChan/Live/refs/heads/main/APTV.m3u", "https://raw.githubusercontent.com/Wirili/IPTV/refs/heads/main/live.m3u", "https://raw.githubusercontent.com/wwb521/live/refs/heads/main/tv.m3u", "https://raw.githubusercontent.com/Kimentanm/aptv/master/m3u/iptv.m3u", "https://raw.githubusercontent.com/Ftindy/IPTV-URL/main/IPV6.m3u", "https://live.zbds.top/tv/iptv4.m3u", "https://live.zbds.top/tv/iptv6.m3u", "https://raw.githubusercontent.com/wind005/TVlive/refs/heads/main/m3u/%E6%B9%96%E5%8D%97%E7%A7%BB%E5%8A%A8.m3u", "https://raw.githubusercontent.com/hanhan8127/TVBox/refs/heads/main/live.txt", "https://raw.githubusercontent.com/hujingguang/ChinaIPTV/main/cnTV_AutoUpdate.m3u8", "https://raw.githubusercontent.com/suxuang/myIPTV/refs/heads/main/ipv4.m3u", "https://raw.githubusercontent.com/suxuang/myIPTV/refs/heads/main/ipv6.m3u", "https://raw.githubusercontent.com/Guovin/iptv-api/gd/output/ipv4/result.m3u", "https://raw.githubusercontent.com/Guovin/iptv-api/gd/output/ipv6/result.m3u", ] def parse_template(template_file): template_channels = OrderedDict() current_category = None with open(template_file, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line and not line.startswith("#"): if "#genre#" in line: current_category = line.split(",")[0].strip() template_channels[current_category] = [] elif current_category: channel_name = line.split(",")[0].strip() template_channels[current_category].append(channel_name) return template_channels def fetch_channels(url): channels = OrderedDict() try: response = requests.get(url, timeout=120) response.raise_for_status() response.encoding = "utf-8" lines = response.text.split("\n") is_m3u = any("#EXTINF" in line for line in lines[:5]) current_category = None if is_m3u: channel_name = "" for line in lines: line = line.strip() if line.startswith("#EXTINF"): match = re.search(r'group-title="(.*?)",(.*)', line) if match: current_category = match.group(1).strip() channel_name = match.group(2).strip() if current_category not in channels: channels[current_category] = [] elif line and not line.startswith("#"): if current_category and channel_name: channels[current_category].append((channel_name, line)) channel_name = "" else: for line in lines: line = line.strip() if "#genre#" in line: current_category = line.split(",")[0].strip() channels[current_category] = [] elif current_category and line: parts = line.split(",", 1) if len(parts) == 2: name, url = parts channels[current_category].append((name.strip(), url.strip())) return channels except Exception as e: print(f"Error fetching {url}: {str(e)}") return OrderedDict() def match_channels(template_channels, all_channels): matched = OrderedDict() for category, names in template_channels.items(): matched[category] = OrderedDict() for name in names: primary_name = name.split("|")[0] for src_category, channels in all_channels.items(): for chan_name, chan_url in channels: if chan_name in name.split("|"): matched[category].setdefault(primary_name, []).append(chan_url) return matched def is_ipv6(url): return re.match(r"^http:\/\/\[[0-9a-fA-F:]+\]", url) is not None def generate_outputs(channels, template_channels): written_urls = set() current_date = datetime.now().strftime("%Y-%m-%d") with open("lib/iptv.m3u", "w", encoding="utf-8") as m3u, \ open("lib/iptv.txt", "w", encoding="utf-8") as txt: # Write channel list total_count = 0 for category in template_channels: if category not in channels: continue txt.write(f"\n{category},#genre#\n") for name in template_channels[category]: primary_name = name.split("|")[0] urls = channels[category].get(primary_name, []) # URL filtering and sorting filtered = [ url for url in urls if url and url not in written_urls # and not any(b in url for b in config.url_blacklist) ] if not filtered: continue # IP version priority sorting # filtered.sort(key=lambda x: is_ipv6(x) != (config.ip_version_priority == "ipv6")) # Format URLs total = len(filtered) for idx, url in enumerate(filtered, 1): base_url = url.split("$")[0] suffix = "$LR•" + ("IPV6" if is_ipv6(url) else "IPV4") if total > 1: suffix += f"•{total}『线路{idx}』" final_url = f"{base_url}{suffix}" m3u.write(f'#EXTINF:-1 tvg-id="{idx}" tvg-name="{primary_name}" ' #f'tvg-logo="https://example.com/logo/{primary_name}.png" ' f'group-title="{category}",{primary_name}\n') m3u.write(f"{final_url}\n") txt.write(f"{primary_name},{final_url}\n") written_urls.add(url) total_count += 1 print(f"频道处理完成，总计有效频道数：{total_count}") def filter_sources(template_file): template = parse_template(template_file) all_channels = OrderedDict() with ThreadPoolExecutor(max_workers=5) as executor: futures = {executor.submit(fetch_channels, url): url for url in tv_urls} for future in as_completed(futures): url = futures[future] try: result = future.result() for cat, chans in result.items(): all_channels.setdefault(cat, []).extend(chans) except Exception as e: print(f"处理源 {url} 时出错: {str(e)}") return match_channels(template, all_channels), template if __name__ == "__main__": matched_channels, template = filter_sources("py/config/iptv.txt") generate_outputs(matched_channels, template)