(url: str, classifier: ChannelClassifier, corrections: dict)
| 282 | return txt_lines |
| 283 | |
| 284 | def process_remote_url(url: str, classifier: ChannelClassifier, corrections: dict): |
| 285 | classifier.other_lines.append(f"{url},#genre#") |
| 286 | try: |
| 287 | headers = {'User-Agent': USER_AGENT} |
| 288 | req = urllib.request.Request(safe_quote_url(url), headers=headers) |
| 289 | with urllib.request.urlopen(req, timeout=URL_FETCH_TIMEOUT) as resp: |
| 290 | data = resp.read() |
| 291 | text = None |
| 292 | for encoding in ['utf-8', 'gbk', 'gb2312', 'iso-8859-1']: |
| 293 | try: |
| 294 | text = data.decode(encoding) |
| 295 | break |
| 296 | except UnicodeDecodeError: |
| 297 | continue |
| 298 | if not text: |
| 299 | print(f"[ERROR] 远程源 {url} 解码失败") |
| 300 | return |
| 301 | if is_m3u_content(text): |
| 302 | lines = convert_m3u_to_txt(text) |
| 303 | else: |
| 304 | lines = [line.strip() for line in text.split('\n') if line.strip()] |
| 305 | print(f"[PROCESS] 远程源 {url} 提取有效行: {len(lines)}") |
| 306 | for line in lines: |
| 307 | process_single_line(line, classifier, corrections) |
| 308 | classifier.other_lines.append('\n') |
| 309 | except Exception as e: |
| 310 | print(f"[ERROR] 处理远程源 {url} 失败: {str(e)}") |
| 311 | |
| 312 | def process_single_line(line: str, classifier: ChannelClassifier, corrections: dict): |
| 313 | if "#genre#" in line or "#EXTINF:" in line or "," not in line or "://" not in line: |
no test coverage detected