AI 生成: 一个自动构建 APTV 直播源的脚本工具,每天自动更新央视、卫视、地方台频道,支持源过滤、去重和可用性检测。
This commit is contained in:
@@ -0,0 +1,56 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""可用性检测与去重"""
|
||||
|
||||
import requests
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
|
||||
def check_url(url, timeout=5):
|
||||
"""检测 URL 是否可用"""
|
||||
try:
|
||||
resp = requests.get(url, timeout=timeout, stream=True, headers={
|
||||
"User-Agent": "Mozilla/5.0 (APTV-Live-Source-Builder)"
|
||||
})
|
||||
return resp.status_code == 200
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def filter_available(channels, timeout=5, max_workers=20):
|
||||
"""并发过滤可用源"""
|
||||
if not channels:
|
||||
return []
|
||||
|
||||
results = []
|
||||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||||
future_map = {
|
||||
executor.submit(check_url, ch["url"], timeout): ch
|
||||
for ch in channels
|
||||
}
|
||||
for future in as_completed(future_map):
|
||||
ch = future_map[future]
|
||||
try:
|
||||
ok = future.result()
|
||||
except Exception:
|
||||
ok = False
|
||||
if ok:
|
||||
results.append(ch)
|
||||
return results
|
||||
|
||||
|
||||
def dedup(channels):
|
||||
"""按 URL 和频道名去重"""
|
||||
seen_urls = set()
|
||||
seen_names = set()
|
||||
result = []
|
||||
for ch in channels:
|
||||
url = ch["url"]
|
||||
name = ch["name"]
|
||||
if url in seen_urls:
|
||||
continue
|
||||
if name in seen_names:
|
||||
continue
|
||||
seen_urls.add(url)
|
||||
seen_names.add(name)
|
||||
result.append(ch)
|
||||
return result
|
||||
Reference in New Issue
Block a user