175 lines
7.5 KiB
Python
175 lines
7.5 KiB
Python
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) |