|
| 1 | +import xml.etree.ElementTree as ET |
| 2 | +from collections import defaultdict |
| 3 | +import aiohttp |
| 4 | +import asyncio |
| 5 | +from tqdm.asyncio import tqdm_asyncio # 引入 tqdm 的异步支持 |
| 6 | +from datetime import datetime, timezone, timedelta |
| 7 | +import gzip |
| 8 | +import shutil |
| 9 | +from xml.dom import minidom |
| 10 | +import re |
| 11 | +from opencc import OpenCC |
| 12 | +import os |
| 13 | +from tqdm import tqdm # 引入 tqdm 的同步支持 |
| 14 | + |
| 15 | +TZ_UTC_PLUS_8 = timezone(timedelta(hours=8)) |
| 16 | + |
| 17 | +# ============ EPG 源预处理规则 ============ |
| 18 | +# 在合并前对指定源的指定频道做预处理 |
| 19 | +# 每条规则: (源URL关键字, 频道名关键字, 预处理函数) |
| 20 | + |
| 21 | +def _adjust_timezone(programme, from_offset, to_offset): |
| 22 | + """将 programme 节点的 start/stop 时区从 from_offset 替换为 to_offset""" |
| 23 | + for attr in ('start', 'stop'): |
| 24 | + val = programme.get(attr, '') |
| 25 | + if from_offset in val: |
| 26 | + programme.set(attr, val.replace(from_offset, to_offset)) |
| 27 | + |
| 28 | +def _make_tz_rule(channel_keyword, from_offset, to_offset): |
| 29 | + """生成一个时区调整规则函数""" |
| 30 | + def rule(channel_name, programme): |
| 31 | + if channel_keyword in channel_name: |
| 32 | + _adjust_timezone(programme, from_offset, to_offset) |
| 33 | + return rule |
| 34 | + |
| 35 | +# 预处理规则列表: (源URL包含的关键字, 规则函数) |
| 36 | +PREPROCESS_RULES = [ |
| 37 | + # 天映经典频道: 时区 +0800 → +0900 (延迟一小时) |
| 38 | + ("kuke31/xmlgz", _make_tz_rule("天映经典", "+0800", "+0700")), |
| 39 | +] |
| 40 | + |
| 41 | +def preprocess_epg(url, epg_content): |
| 42 | + """对 epg_content XML 字符串按规则做预处理，返回处理后的字符串""" |
| 43 | + matched_rules = [rule for keyword, rule in PREPROCESS_RULES if keyword in url] |
| 44 | + if not matched_rules: |
| 45 | + return epg_content |
| 46 | + |
| 47 | + try: |
| 48 | + parser = ET.XMLParser(encoding='UTF-8') |
| 49 | + root = ET.fromstring(epg_content, parser=parser) |
| 50 | + except ET.ParseError: |
| 51 | + return epg_content |
| 52 | + |
| 53 | + # 建立 channel_id -> display_name 的映射 |
| 54 | + channel_names = {} |
| 55 | + for channel in root.findall('channel'): |
| 56 | + cid = channel.get('id', '') |
| 57 | + names = [n.text for n in channel.findall('display-name') if n.text] |
| 58 | + channel_names[cid] = ' '.join(names) + ' ' + cid |
| 59 | + |
| 60 | + for programme in root.findall('programme'): |
| 61 | + cid = programme.get('channel', '') |
| 62 | + name_str = channel_names.get(cid, cid) |
| 63 | + for rule in matched_rules: |
| 64 | + rule(name_str, programme) |
| 65 | + |
| 66 | + return ET.tostring(root, encoding='unicode') |
| 67 | +# ============ 预处理规则结束 ============ |
| 68 | + |
| 69 | + |
| 70 | +def transform2_zh_hans(string): |
| 71 | + cc = OpenCC("t2s") |
| 72 | + new_str = cc.convert(string) |
| 73 | + return new_str |
| 74 | + |
| 75 | + |
| 76 | +async def fetch_epg(url): |
| 77 | + connector = aiohttp.TCPConnector(limit=16, ssl=False) |
| 78 | + headers = { |
| 79 | + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36" |
| 80 | + } |
| 81 | + try: |
| 82 | + async with aiohttp.ClientSession(connector=connector, trust_env=True, headers=headers) as session: |
| 83 | + async with session.get(url) as response: |
| 84 | + if url.endswith('.gz'): |
| 85 | + compressed_data = await response.read() |
| 86 | + return gzip.decompress(compressed_data).decode('utf-8', errors='ignore') |
| 87 | + else: |
| 88 | + return await response.text(encoding='utf-8') |
| 89 | + except aiohttp.ClientError as e: |
| 90 | + print(f"{url}HTTP请求错误: {e}") |
| 91 | + except asyncio.TimeoutError: |
| 92 | + print("{url}请求超时") |
| 93 | + except Exception as e: |
| 94 | + print(f"{url}其他错误: {e}") |
| 95 | + return None |
| 96 | + |
| 97 | +def process_display_name(display_name): |
| 98 | + if display_name.endswith('高清'): |
| 99 | + display_name = display_name[:-2] |
| 100 | + return display_name |
| 101 | + |
| 102 | +def parse_epg(epg_content): |
| 103 | + try: |
| 104 | + parser = ET.XMLParser(encoding='UTF-8') |
| 105 | + root = ET.fromstring(epg_content, parser=parser) |
| 106 | + except ET.ParseError as e: |
| 107 | + print(f"Error parsing XML: {e}") |
| 108 | + print(f"Problematic content: {epg_content[:500]}") |
| 109 | + return {}, defaultdict(list) |
| 110 | + |
| 111 | + channels = {} |
| 112 | + programmes = defaultdict(list) |
| 113 | + |
| 114 | + for channel in root.findall('channel'): |
| 115 | + channel_id = transform2_zh_hans(channel.get('id')) |
| 116 | + channel_display_names = [] |
| 117 | + for name in channel.findall('display-name'): |
| 118 | + t_name = transform2_zh_hans(name.text) |
| 119 | + t_name = process_display_name(t_name) |
| 120 | + channel_display_names.append([t_name, name.get('lang', 'zh')]) |
| 121 | + if not channel_id.isdigit() and channel_id not in channel_display_names: |
| 122 | + channel_display_names.append([channel_id, 'zh']) |
| 123 | + channels[channel_id] = channel_display_names |
| 124 | + |
| 125 | + today = datetime.now(TZ_UTC_PLUS_8).date() |
| 126 | + valid_channels = set() |
| 127 | + |
| 128 | + for programme in root.findall('programme'): |
| 129 | + channel_id = transform2_zh_hans(programme.get('channel')) |
| 130 | + channel_start = datetime.strptime( |
| 131 | + re.sub(r'\s+', '', programme.get('start')), "%Y%m%d%H%M%S%z") |
| 132 | + channel_stop = datetime.strptime( |
| 133 | + re.sub(r'\s+', '', programme.get('stop')), "%Y%m%d%H%M%S%z") |
| 134 | + channel_start = channel_start.astimezone(TZ_UTC_PLUS_8) |
| 135 | + channel_stop = channel_stop.astimezone(TZ_UTC_PLUS_8) |
| 136 | + |
| 137 | + if channel_stop.date() == today: |
| 138 | + valid_channels.add(channel_id) |
| 139 | + |
| 140 | + channel_elem = ET.SubElement( |
| 141 | + root, 'programme', attrib={"start": channel_start.strftime("%Y%m%d%H%M%S %z"), "stop": channel_stop.strftime("%Y%m%d%H%M%S %z")}) |
| 142 | + for title in programme.findall('title'): |
| 143 | + if title.text is None: |
| 144 | + channel_title = "精彩节目" |
| 145 | + else: |
| 146 | + channel_title = title.text.strip() |
| 147 | + langattr = title.get('lang') |
| 148 | + if langattr == 'zh' or langattr is None: |
| 149 | + channel_title = transform2_zh_hans(channel_title) |
| 150 | + channel_elem_t = ET.SubElement( |
| 151 | + channel_elem, 'title') |
| 152 | + channel_elem_t.text = channel_title |
| 153 | + if langattr is not None: |
| 154 | + channel_elem_t.set('lang', langattr) |
| 155 | + for desc in programme.findall('desc'): |
| 156 | + if desc.text is None: |
| 157 | + continue |
| 158 | + langattr = desc.get('lang') |
| 159 | + channel_desc = desc.text.strip() |
| 160 | + if langattr == 'zh' or langattr is None: |
| 161 | + channel_desc = transform2_zh_hans(channel_desc) |
| 162 | + channel_elem_d = ET.SubElement( |
| 163 | + channel_elem, 'desc') |
| 164 | + channel_elem_d.text = channel_desc.strip() |
| 165 | + if langattr is not None: |
| 166 | + channel_elem_d.set('lang', langattr) |
| 167 | + programmes[channel_id].append(channel_elem) |
| 168 | + |
| 169 | + # Filter channels that don't have any program ending today |
| 170 | + channels = {k: v for k, v in channels.items() if k in valid_channels} |
| 171 | + # Optional: Filter programmes as well to keep data consistent, |
| 172 | + # though only valid channels are returned so main loop might be fine. |
| 173 | + # But filtering programmes dict saves memory and ensures correctness if main iterates programmes keys logic changes. |
| 174 | + programmes = {k: v for k, v in programmes.items() if k in valid_channels} |
| 175 | + |
| 176 | + return channels, programmes |
| 177 | + |
| 178 | + |
| 179 | +def write_to_xml(channels_id, channels_names, programmes, filename): |
| 180 | + # 目录不存在 |
| 181 | + if not os.path.exists('output'): |
| 182 | + os.makedirs('output') |
| 183 | + current_time = datetime.now(TZ_UTC_PLUS_8).strftime("%Y%m%d%H%M%S %z") |
| 184 | + root = ET.Element('tv', attrib={'date': current_time}) |
| 185 | + for channel_id in channels_id: |
| 186 | + channel_elem = ET.SubElement( |
| 187 | + root, 'channel', attrib={"id": channel_id}) |
| 188 | + for display_name_node in channels_names[channel_id]: |
| 189 | + display_name = display_name_node[0] |
| 190 | + langattr = display_name_node[1] |
| 191 | + display_name_elem = ET.SubElement( |
| 192 | + channel_elem, 'display-name', attrib={"lang": langattr}) |
| 193 | + display_name_elem.text = display_name |
| 194 | + for prog in programmes[channel_id]: |
| 195 | + prog.set('channel', channel_id) # 设置 programme 的 channel 属性 |
| 196 | + root.append(prog) |
| 197 | + |
| 198 | + # Beautify the XML output |
| 199 | + rough_string = ET.tostring(root, 'utf-8') |
| 200 | + reparsed = minidom.parseString(rough_string) |
| 201 | + with open(filename, 'w', encoding='utf-8') as f: |
| 202 | + f.write(reparsed.toprettyxml(indent='\t', newl='\n')) |
| 203 | + |
| 204 | + |
| 205 | +def compress_to_gz(input_filename, output_filename): |
| 206 | + with open(input_filename, 'rb') as f_in: |
| 207 | + with gzip.open(output_filename, 'wb') as f_out: |
| 208 | + shutil.copyfileobj(f_in, f_out) |
| 209 | + |
| 210 | + |
| 211 | +def get_urls(): |
| 212 | + urls = [] |
| 213 | + with open('config.txt', 'r', encoding='utf-8') as file: |
| 214 | + for line in file: |
| 215 | + line = line.strip() |
| 216 | + if line and not line.startswith('#'): |
| 217 | + urls.append(line) |
| 218 | + return urls |
| 219 | + |
| 220 | + |
| 221 | +async def main(): |
| 222 | + urls = get_urls() |
| 223 | + tasks = [fetch_epg(url) for url in urls] |
| 224 | + print("Fetching EPG data...") |
| 225 | + epg_contents = await tqdm_asyncio.gather(*tasks, desc="Fetching URLs") |
| 226 | + all_channels_map = {} |
| 227 | + all_channel_id = set() |
| 228 | + all_channel_names = defaultdict(list) |
| 229 | + all_programmes = defaultdict(list) |
| 230 | + print("Finished.") |
| 231 | + i = 0 |
| 232 | + for epg_content in epg_contents: |
| 233 | + i += 1 |
| 234 | + print(f"Processing EPG source...{i}/{len(epg_contents)}") |
| 235 | + if epg_content is None: |
| 236 | + continue |
| 237 | + print("Parsing EPG data...") |
| 238 | + epg_content = preprocess_epg(urls[i - 1], epg_content) |
| 239 | + channels, programmes = parse_epg(epg_content) |
| 240 | + print("Finished.") |
| 241 | + with tqdm(total=len(channels), desc="Merging EPG", unit="file") as pbar: |
| 242 | + for channel_id, display_names in channels.items(): |
| 243 | + if len(programmes[channel_id]) == 0: |
| 244 | + continue |
| 245 | + is_in_map = False |
| 246 | + map_id = "" |
| 247 | + for display_name_node in display_names: |
| 248 | + if is_in_map: |
| 249 | + break |
| 250 | + display_name = display_name_node[0] |
| 251 | + is_in_map = display_name in all_channels_map |
| 252 | + map_id = display_name |
| 253 | + map_id = all_channels_map.get(map_id, channel_id) |
| 254 | + if not is_in_map: |
| 255 | + all_channel_id.add(map_id) |
| 256 | + all_channel_names[map_id] = display_names |
| 257 | + all_programmes[map_id] = programmes[channel_id] |
| 258 | + for display_name_node in display_names: |
| 259 | + display_name = display_name_node[0] |
| 260 | + all_channels_map[display_name] = map_id |
| 261 | + else: |
| 262 | + if len(all_programmes[map_id]) < len(programmes[channel_id]): |
| 263 | + all_programmes[map_id] = programmes[channel_id] |
| 264 | + for display_name_node in display_names: |
| 265 | + display_name = display_name_node[0] |
| 266 | + if display_name not in all_channels_map: |
| 267 | + all_channel_names[map_id].append(display_name_node) |
| 268 | + all_channels_map[display_name] = map_id |
| 269 | + pbar.update(1) # 更新进度条 |
| 270 | + print("Writing to XML...") |
| 271 | + write_to_xml(all_channel_id, all_channel_names, |
| 272 | + all_programmes, 'output/epg.xml') |
| 273 | + compress_to_gz('output/epg.xml', 'output/epg.gz') |
| 274 | + |
| 275 | +if __name__ == '__main__': |
| 276 | + asyncio.run(main()) |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?v=4&size=40){kind=avatar}
0 commit comments