diff --git a/app.py b/app.py index fcb4a695..1eec4da9 100644 --- a/app.py +++ b/app.py @@ -4,7 +4,6 @@ import os from collections import defaultdict # ================== 配置区域 ================== -# 需要删除的分组 (这些分组下的频道会被过滤) DELETE_GROUPS = ["4K频道", "8K频道"] PROVINCE_GROUPS = ["北京", "安徽", "甘肃", "广东", "贵州", "海南", "河北", "河南", "黑龙江", "湖北", "湖南", "吉林", "江苏", "江西", "辽宁", "青海", "山东", "上海", "四川", "云南", "浙江", "重庆", "香港"] @@ -17,7 +16,7 @@ M3U_SOURCES = [ {"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": "https://gh.tryxd.cn/https://raw.githubusercontent.com/Meroser/IPTV/refs/heads/main/IPTV-demo.m3u", "ua": "okhttp/4.12.0"}, + {"name": "xhztv", "url": "https://gh.tryxd.cn/https://raw.githubusercontent.com/hjdhnx/hipy-sniffer/refs/heads/main/static/lives/lives.txt", "ua": "okhttp/4.12.0"}, {"name": "top", "url": "https://tv.iill.top/m3u/Gather", "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"}, @@ -35,38 +34,29 @@ def robust_download(url, ua, max_retries=3): response.encoding = response.apparent_encoding return response.text except Exception as e: - if attempt == max_retries - 1: raise + if attempt == max_retries - 1: + print(f"❌ 无法下载: {url}. 错误: {str(e)}") + return None 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 @@ -85,7 +75,6 @@ def parse_m3u(content): 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"]): @@ -106,40 +95,16 @@ def generate_m3u_output(channels): 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): - """通用文件保存函数""" - # 创建 live 文件夹,如果不存在 live_folder = 'live' if not os.path.exists(live_folder): os.makedirs(live_folder) file_path = os.path.join(live_folder, filename) try: - with open(filename, "w", encoding="utf-8") as f: + with open(file_path, "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 @@ -149,26 +114,28 @@ def main(): 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)}") - + content = robust_download(source["url"], source["ua"]) + if not content: + print(f"[×] {source['name']} 无法处理,跳过") + continue + + 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)} 频道)") + + if not all_channels: + print("❌ 没有可用的频道数据,终止生成文件") + return + 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")