52 lines
1.3 KiB
Python
52 lines
1.3 KiB
Python
"""AVTV 主入口：抓取 -> 解析 -> 过滤去重 -> 生成 m3u。"""
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
from avtv.fetcher import fetch_all
|
|
from avtv.filter import process
|
|
from avtv.generator import write_m3u
|
|
from avtv.parser import Channel, parse
|
|
from avtv.sources import SOURCES
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
|
)
|
|
log = logging.getLogger("avtv")
|
|
|
|
OUTPUT = Path("output/avtv.m3u")
|
|
|
|
|
|
def collect() -> list[Channel]:
|
|
"""抓取并解析所有源。"""
|
|
raw_texts = fetch_all(SOURCES)
|
|
log.info("successfully fetched %d/%d sources", len(raw_texts), len(SOURCES))
|
|
|
|
all_channels: list[Channel] = []
|
|
for text in raw_texts:
|
|
try:
|
|
channels = parse(text)
|
|
all_channels.extend(channels)
|
|
except Exception as exc: # noqa: BLE001
|
|
log.warning("parse failed: %s", exc)
|
|
log.info("parsed %d raw channels", len(all_channels))
|
|
return all_channels
|
|
|
|
|
|
def run() -> None:
|
|
raw = collect()
|
|
cleaned = process(raw)
|
|
log.info("after filter/dedupe: %d channels", len(cleaned))
|
|
|
|
if not cleaned:
|
|
log.error("no valid channels found, skip writing")
|
|
return
|
|
|
|
out = write_m3u(cleaned, OUTPUT)
|
|
log.info("written -> %s", out)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|