"""解析 m3u / txt 格式的直播源文本,提取频道条目。""" import re from dataclasses import dataclass, field @dataclass class Channel: """单个频道条目。""" name: str url: str logo: str = "" group: str = "" tvg_id: str = "" extra: dict = field(default_factory=dict) _ATTR_RE = re.compile(r'([\w-]+)="([^"]*)"') def _parse_extinf(line: str) -> dict: """解析 #EXTINF 行中的属性。""" attrs = {} for key, value in _ATTR_RE.findall(line): attrs[key.lower()] = value # 逗号后面的名称 if "," in line: attrs["name"] = line.rsplit(",", 1)[1].strip() return attrs def parse_m3u(text: str) -> list[Channel]: """解析 m3u 文本。""" channels: list[Channel] = [] lines = [ln.strip() for ln in text.splitlines()] i = 0 while i < len(lines): line = lines[i] if line.startswith("#EXTINF"): attrs = _parse_extinf(line) # 下一行非注释的即为 URL j = i + 1 while j < len(lines) and (not lines[j] or lines[j].startswith("#")): j += 1 if j < len(lines): url = lines[j] name = attrs.get("name") or attrs.get("tvg-name") or "" if name and url: channels.append( Channel( name=name, url=url, logo=attrs.get("tvg-logo", ""), group=attrs.get("group-title", ""), tvg_id=attrs.get("tvg-id", ""), ) ) i = j + 1 continue i += 1 return channels def parse_txt(text: str) -> list[Channel]: """解析常见 txt 格式:频道名,URL 或 频道名 URL。""" channels: list[Channel] = [] for raw in text.splitlines(): line = raw.strip() if not line or line.startswith("#"): continue # 支持 , 或空格分隔 if "," in line: name, url = line.split(",", 1) elif " " in line: name, url = line.split(" ", 1) else: continue name, url = name.strip(), url.strip() if name and url.startswith(("http://", "https://", "rtmp://", "rtsp://")): channels.append(Channel(name=name, url=url)) return channels def parse(text: str) -> list[Channel]: """自动识别格式并解析。""" if "#EXTM3U" in text or "#EXTINF" in text: return parse_m3u(text) return parse_txt(text)