84 lines
2.5 KiB
Python
84 lines
2.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""主程序：IPTV 直播源采集、检测、去重"""
|
|
|
|
import os
|
|
import sys
|
|
import argparse
|
|
from collector import collect_all
|
|
from checker import check_all
|
|
from deduplicator import deduplicate, group_by_category
|
|
from config import OUTPUT_DIR, OUTPUT_FILE
|
|
|
|
|
|
def write_m3u(channels, filepath):
|
|
"""写入 M3U 文件"""
|
|
os.makedirs(os.path.dirname(filepath) or ".", exist_ok=True)
|
|
with open(filepath, "w", encoding="utf-8") as f:
|
|
f.write("#EXTM3U\n")
|
|
for ch in channels:
|
|
name = ch.get("name", "未知频道")
|
|
group = ch.get("group", "其他") or "其他"
|
|
tvg_name = ch.get("tvg_name", name)
|
|
url = ch.get("url", "")
|
|
f.write(
|
|
f'#EXTINF:-1 tvg-name="{tvg_name}" group-title="{group}",{name}\n'
|
|
)
|
|
f.write(f"{url}\n")
|
|
print(f"[输出] 已写入 {filepath}，共 {len(channels)} 个频道")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="IPTV 直播源采集检测去重工具")
|
|
parser.add_argument("--no-check", action="store_true", help="跳过检测步骤")
|
|
parser.add_argument("--output", default=OUTPUT_FILE, help="输出文件名")
|
|
parser.add_argument("--threads", type=int, default=None, help="并发线程数")
|
|
args = parser.parse_args()
|
|
|
|
if args.threads:
|
|
import config
|
|
config.MAX_WORKERS = args.threads
|
|
|
|
print("=" * 50)
|
|
print("IPTV 直播源采集检测去重工具")
|
|
print("=" * 50)
|
|
|
|
# 1. 采集
|
|
print("\n[步骤 1/4] 采集直播源...")
|
|
channels = collect_all()
|
|
if not channels:
|
|
print("未采集到任何频道，退出")
|
|
sys.exit(1)
|
|
|
|
# 2. 检测
|
|
if args.no_check:
|
|
print("\n[步骤 2/4] 跳过检测")
|
|
valid = channels
|
|
else:
|
|
print("\n[步骤 2/4] 检测直播源可用性...")
|
|
valid = check_all(channels)
|
|
|
|
if not valid:
|
|
print("没有可用频道，退出")
|
|
sys.exit(1)
|
|
|
|
# 3. 去重
|
|
print("\n[步骤 3/4] 去重...")
|
|
unique = deduplicate(valid)
|
|
|
|
# 4. 输出
|
|
print("\n[步骤 4/4] 输出结果...")
|
|
output_path = os.path.join(OUTPUT_DIR, args.output)
|
|
write_m3u(unique, output_path)
|
|
|
|
# 统计
|
|
groups = group_by_category(unique)
|
|
print("\n分类统计：")
|
|
for g, items in sorted(groups.items(), key=lambda x: -len(x[1])):
|
|
print(f" {g}: {len(items)} 个频道")
|
|
|
|
print(f"\n完成！输出文件：{output_path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|