iptv/M3U8/fetch.py

#!/usr/bin/env python3
import asyncio
from pathlib import Path

import httpx
from scrapers import fstv, livetvsx, ppv, streambtw, streameast, tvpass
from scrapers.utils import UA, get_logger

log = get_logger(__name__)

BASE_URL = "https://s.id/ePwXT"

M3U8_FILE = Path(__file__).parent / "TV.m3u8"

CLIENT = httpx.AsyncClient(
    timeout=5,
    follow_redirects=True,
    headers={"User-Agent": UA},
)


async def vanilla_fetch() -> tuple[list[str], int]:
    log.info("Fetching base M3U8")

    try:
        r = await CLIENT.get(BASE_URL)
        r.raise_for_status()
    except Exception as e:
        log.error(f'Failed to fetch "{BASE_URL}"\n{e}')
        raise SystemExit(e) from e

    last_chnl_num = int(r.text.split("tvg-chno=")[-1].split('"')[1])

    return r.text.splitlines()[1:], last_chnl_num


async def main() -> None:
    tasks = [
        asyncio.create_task(fstv.main(CLIENT)),
        asyncio.create_task(livetvsx.main(CLIENT)),
        asyncio.create_task(ppv.main(CLIENT)),
        asyncio.create_task(streambtw.main(CLIENT)),
        asyncio.create_task(streameast.main(CLIENT)),
        asyncio.create_task(tvpass.main(CLIENT)),
        vanilla_fetch(),
    ]

    results = await asyncio.gather(*tasks)

    base_m3u8, tvg_chno = results[-1]

    additions = (
        fstv.urls
        | livetvsx.urls
        | ppv.urls
        | streambtw.urls
        | streameast.urls
        | tvpass.urls
    )

    live_events = []

    for chnl_num, (event, info) in enumerate(
        sorted(additions.items()),
        start=tvg_chno + 1,
    ):
        live_events.extend(
            (
                f'#EXTINF:-1 tvg-chno="{chnl_num}" tvg-id="(N/A)" tvg-name="{event}" tvg-logo="{info["logo"]}" group-title="Live Events",{event}',
                f'#EXTVLCOPT:http-referrer={info["base"]}',
                f'#EXTVLCOPT:http-origin={info["base"]}',
                info["url"],
            )
        )

    m3u8_content = "\n".join(
        [
            '#EXTM3U url-tvg="https://raw.githubusercontent.com/doms9/iptv/refs/heads/default/EPG/TV.xml"'
        ]
        + base_m3u8
        + live_events
    )

    M3U8_FILE.write_text(m3u8_content, encoding="utf-8")

    log.info(f"M3U8 saved to {M3U8_FILE.name}")


if __name__ == "__main__":
    asyncio.run(main())