Update app.py

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"])
-            content = robust_download(source["url"], source["ua"])
+        if not content:
-            channels = parse_m3u(content)
+            print(f"[×] {source['name']} 无法处理，跳过")
-            
+            continue
-            processed = []
+        
-            for ch in channels:
+        channels = parse_m3u(content)
-                if cleaned_meta := process_channel(ch["meta"]):
+        processed = []
-                    processed.append({"meta": cleaned_meta, "url": ch["url"]})
+        for ch in channels:
-            
+            if cleaned_meta := process_channel(ch["meta"]):
-            all_channels.extend(processed)
+                processed.append({"meta": cleaned_meta, "url": ch["url"]})
-            print(f"[✓] {source['name']} 处理完成 ({len(processed)} 频道)")
+        
-        except Exception as e:
+        all_channels.extend(processed)
-            print(f"[×] {source['name']} 失败: {str(e)}")
+        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")