174 lines
6.8 KiB
Python
174 lines
6.8 KiB
Python
import re
|
||
import requests
|
||
import os
|
||
from collections import defaultdict
|
||
|
||
# ================== 配置区域 ==================
|
||
# 需要删除的分组 (这些分组下的频道会被过滤)
|
||
DELETE_GROUPS = ["4K频道", "8K频道"]
|
||
PROVINCE_GROUPS = ["北京", "安徽", "甘肃", "广东", "贵州", "海南", "河北", "河南", "黑龙江", "湖北", "湖南",
|
||
"吉林", "江苏", "江西", "辽宁", "青海", "山东", "上海", "四川", "云南", "浙江", "重庆", "香港"]
|
||
GROUP_REPLACEMENTS = {"央视": "央视频道", "卫视": "卫视频道", "其他": "其他频道"}
|
||
DELETE_CHARS = ["iHOT-", "NewTV-", "SiTV-", "-HEVC", "-50-FPS", "-高码", "-4K", "-BPTV", "咪咕视频_8M1080_"]
|
||
GROUP_ORDER = ["收藏频道", "央视频道", "卫视频道", "其他频道", "地方频道"]
|
||
|
||
M3U_SOURCES = [
|
||
{"name": "aktv", "url": "https://git.gra.phite.ro/alantang/tvbs/raw/branch/main/output/result.m3u", "ua": "okhttp/4.12.0"},
|
||
{"name": "mytv", "url": "https://codeberg.org/sy147258/iptv/raw/branch/main/电视", "ua": "okhttp/4.12.0"},
|
||
{"name": "自用收藏", "url": "http://aktv.space/live.m3u", "ua": "okhttp/4.12.0"},
|
||
{"name": "big", "url": "https://git.gra.phite.ro/alantang/auto-iptv/raw/branch/main/live_ipv4.m3u", "ua": "okhttp/4.12.0"},
|
||
{"name": "xhztv", "url": "http://xhztv.top/new.txt", "ua": "okhttp/4.12.0"},
|
||
{"name": "top", "url": "http://tot.totalh.net/tttt.txt", "ua": "okhttp/4.12.0"},
|
||
{"name": "zbds", "url": "https://live.zbds.top/tv/iptv6.txt", "ua": "okhttp/4.12.0"},
|
||
{"name": "野火", "url": "https://gh.tryxd.cn/https://raw.githubusercontent.com/tianya7981/jiekou/main/野火959", "ua": "okhttp/4.12.0"},
|
||
{"name": "jundie", "url": "http://home.jundie.top:81/Cat/tv/live.txt", "ua": "okhttp/4.12.0"},
|
||
{"name": "MyIPTV", "url": "https://git.gra.phite.ro/alantang/auto-iptv/raw/branch/main/live_ipv6.m3u", "ua": "okhttp/4.12.0"}
|
||
]
|
||
|
||
# ================== 核心功能 ==================
|
||
def robust_download(url, ua, max_retries=3):
|
||
headers = {'User-Agent': ua}
|
||
for attempt in range(max_retries):
|
||
try:
|
||
response = requests.get(url, headers=headers, timeout=15)
|
||
response.raise_for_status()
|
||
response.encoding = response.apparent_encoding
|
||
return response.text
|
||
except Exception as e:
|
||
if attempt == max_retries - 1: raise
|
||
print(f"正在重试 {url} (第 {attempt+1} 次)")
|
||
|
||
def process_channel(line):
|
||
if any(f'group-title="{g}"' in line for g in DELETE_GROUPS):
|
||
return None
|
||
|
||
def process_channel(line):
|
||
"""频道信息处理流水线"""
|
||
# 过滤不需要的分组
|
||
if any(f'group-title="{g}"' in line for g in DELETE_GROUPS):
|
||
return None
|
||
|
||
# 分组名称替换
|
||
for old, new in GROUP_REPLACEMENTS.items():
|
||
line = line.replace(f'group-title="{old}"', f'group-title="{new}"')
|
||
|
||
# 省份频道处理
|
||
for province in PROVINCE_GROUPS:
|
||
if f'group-title="{province}"' in line:
|
||
new_group = '地方频道' if "卫视" not in line else '卫视频道'
|
||
line = line.replace(f'group-title="{province}"', f'group-title="{new_group}"')
|
||
|
||
# 特殊频道修正
|
||
if "凤凰卫视" in line:
|
||
line = line.replace('group-title="地方频道"', 'group-title="卫视频道"')
|
||
|
||
# 清洗频道名称
|
||
for char in DELETE_CHARS:
|
||
line = line.replace(char, "")
|
||
|
||
# 标准化CCTV写法
|
||
line = re.sub(r'cctv-?', 'CCTV', line, flags=re.IGNORECASE)
|
||
|
||
return line
|
||
|
||
def parse_m3u(content):
|
||
channels = []
|
||
current_channel = {}
|
||
for line in content.splitlines():
|
||
line = line.strip()
|
||
if line.startswith("#EXTINF"):
|
||
current_channel = {"meta": line, "url": ""}
|
||
elif line.startswith("http"):
|
||
current_channel["url"] = line
|
||
channels.append(current_channel)
|
||
current_channel = {}
|
||
return channels
|
||
|
||
def generate_m3u_output(channels):
|
||
"""生成M3U格式内容"""
|
||
group_dict = defaultdict(list)
|
||
for channel in channels:
|
||
if match := re.search(r'group-title="([^"]+)"', channel["meta"]):
|
||
group = match.group(1)
|
||
group_dict[group].append(channel)
|
||
|
||
ordered_groups = []
|
||
for group in GROUP_ORDER:
|
||
if group in group_dict:
|
||
ordered_groups.append((group, group_dict.pop(group)))
|
||
for group in sorted(group_dict.keys()):
|
||
ordered_groups.append((group, group_dict[group]))
|
||
|
||
output = ["#EXTM3U"]
|
||
for group, items in ordered_groups:
|
||
for item in items:
|
||
output.append(item["meta"])
|
||
output.append(item["url"])
|
||
return "\n".join(output)
|
||
|
||
def generate_txt_output(channels):
|
||
"""生成TXT格式内容（分组名称,频道名称,URL）"""
|
||
txt_lines = []
|
||
for channel in channels:
|
||
# 提取分组名称
|
||
group_match = re.search(r'group-title="([^"]+)"', channel["meta"])
|
||
group = group_match.group(1) if group_match else "未知分组"
|
||
|
||
# 提取频道名称（最后一个逗号后的内容）
|
||
channel_name = re.split(r',(?![^"]*\"\,)', channel["meta"])[-1].strip()
|
||
|
||
# 清洗频道名称
|
||
for char in DELETE_CHARS:
|
||
channel_name = channel_name.replace(char, "")
|
||
|
||
# 添加条目
|
||
txt_lines.append(f"{group},{channel_name},{channel['url']}")
|
||
|
||
return "\n".join(txt_lines)
|
||
|
||
def save_file(content, filename):
|
||
"""通用文件保存函数"""
|
||
try:
|
||
with open(filename, "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
print(f"✅ 成功生成 {filename} 文件")
|
||
return True
|
||
except PermissionError:
|
||
print(f"❌ 无写入权限: {filename}")
|
||
except Exception as e:
|
||
print(f"❌ 保存文件失败: {str(e)}")
|
||
return False
|
||
|
||
def main():
|
||
all_channels = []
|
||
|
||
print("开始下载和处理数据源...")
|
||
for source in M3U_SOURCES:
|
||
try:
|
||
content = robust_download(source["url"], source["ua"])
|
||
channels = parse_m3u(content)
|
||
|
||
processed = []
|
||
for ch in channels:
|
||
if cleaned_meta := process_channel(ch["meta"]):
|
||
processed.append({"meta": cleaned_meta, "url": ch["url"]})
|
||
|
||
all_channels.extend(processed)
|
||
print(f"[✓] {source['name']} 处理完成 ({len(processed)} 频道)")
|
||
except Exception as e:
|
||
print(f"[×] {source['name']} 失败: {str(e)}")
|
||
|
||
print("\n生成最终文件...")
|
||
# 生成M3U文件
|
||
m3u_content = generate_m3u_output(all_channels)
|
||
save_file(m3u_content, "live.m3u")
|
||
|
||
# 生成TXT文件
|
||
txt_content = generate_txt_output(all_channels)
|
||
save_file(txt_content, "live.txt")
|
||
|
||
print(f"\n处理完成！有效频道总数: {len(all_channels)}")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|