80 lines
2.2 KiB
Python
80 lines
2.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
直播源采集与更新脚本
|
|
生成标准 M3U 格式文件到 output/tubo.m3u
|
|
"""
|
|
|
|
import os
|
|
import time
|
|
import requests
|
|
from datetime import datetime
|
|
from sources import CHANNELS
|
|
|
|
OUTPUT_DIR = "output"
|
|
OUTPUT_FILE = os.path.join(OUTPUT_DIR, "tubo.m3u")
|
|
|
|
# 请求头，模拟常见播放器
|
|
HEADERS = {
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
|
"Accept": "*/*",
|
|
}
|
|
|
|
|
|
def check_url(url, timeout=5):
|
|
"""检测直播源是否可用"""
|
|
try:
|
|
resp = requests.get(url, headers=HEADERS, timeout=timeout, stream=True)
|
|
return resp.status_code == 200
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def build_m3u(channels):
|
|
"""根据频道列表构建 m3u 内容"""
|
|
lines = ["#EXTM3U"]
|
|
lines.append(f"# 途播播放器直播源 - 更新时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
lines.append(f"# 频道总数: {len(channels)}")
|
|
lines.append("")
|
|
|
|
for ch in channels:
|
|
name = ch["name"]
|
|
group = ch.get("group", "其他")
|
|
url = ch["url"]
|
|
# tvg-name / group-title 便于播放器分组
|
|
lines.append(f'#EXTINF:-1 tvg-name="{name}" group-title="{group}",{name}')
|
|
lines.append(url)
|
|
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def update():
|
|
"""执行一次更新"""
|
|
os.makedirs(OUTPUT_DIR, exist_ok=True)
|
|
|
|
print(f"[{datetime.now()}] 开始更新直播源...")
|
|
valid_channels = []
|
|
for ch in CHANNELS:
|
|
ok = check_url(ch["url"])
|
|
status = "✓" if ok else "✗"
|
|
print(f" {status} {ch['name']}")
|
|
if ok:
|
|
valid_channels.append(ch)
|
|
time.sleep(0.1) # 避免请求过频
|
|
|
|
# 如果全部失效，则保留全部频道（避免空文件）
|
|
if not valid_channels:
|
|
print(" 警告：所有源检测失败，保留全部频道")
|
|
valid_channels = CHANNELS
|
|
|
|
content = build_m3u(valid_channels)
|
|
with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
|
|
f.write(content)
|
|
|
|
print(f"[{datetime.now()}] 更新完成，共 {len(valid_channels)} 个频道 -> {OUTPUT_FILE}")
|
|
return OUTPUT_FILE
|
|
|
|
|
|
if __name__ == "__main__":
|
|
update()
|