import asyncio import json from datetime import datetime, timedelta from functools import partial from pathlib import Path from urllib.parse import urljoin import httpx from playwright.async_api import async_playwright from selectolax.parser import HTMLParser from .utils import ( LOGOS, TZ, capture_req, firefox, get_base, get_logger, load_cache, now, safe_process_event, ) log = get_logger(__name__) urls: dict[str, dict[str, str | float]] = {} CACHE_FILE = Path(__file__).parent / "caches" / "streameast.json" MIRRORS = [ "https://streameast.ga", "https://streameast.tw", "https://streameast.ph", "https://streameast.sg", "https://streameast.ch", "https://streameast.ec", "https://streameast.fi", "https://streameast.ms", "https://streameast.ps", "https://streameast.cf", "https://streameast.sk", "https://thestreameast.co", "https://thestreameast.fun", "https://thestreameast.ru", "https://thestreameast.su", ] LOGOS["CFB"] = LOGOS["NCAAF"] LOGOS["CBB"] = LOGOS["NCAAB"] async def process_event(url: str, url_num: int) -> str | None: async with async_playwright() as p: browser, context = await firefox(p) page = await context.new_page() captured: list[str] = [] got_one = asyncio.Event() handler = partial(capture_req, captured=captured, got_one=got_one) page.on("request", handler) try: await page.goto(url, wait_until="domcontentloaded", timeout=15_000) wait_task = asyncio.create_task(got_one.wait()) try: await asyncio.wait_for(wait_task, timeout=10) except asyncio.TimeoutError: log.warning(f"URL {url_num}) Timed out waiting for M3U8.") return finally: if not wait_task.done(): wait_task.cancel() try: await wait_task except asyncio.CancelledError: pass if captured: log.info(f"URL {url_num}) Captured M3U8") return captured[-1] log.warning(f"URL {url_num}) No M3U8 captured after waiting.") return except Exception as e: log.warning(f"URL {url_num}) Exception while processing: {e}") return finally: page.remove_listener("request", handler) await page.close() await browser.close() async def get_events( client: httpx.AsyncClient, url: str, cached_keys: list[str], ) -> list[dict[str, str]]: try: r = await client.get(url) r.raise_for_status() except Exception as e: log.error(f'Failed to fetch "{url}"\n{e}') return [] soup = HTMLParser(r.text) events = [] start_dt = now - timedelta(minutes=30) end_dt = now + timedelta(minutes=30) for li in soup.css("li.f1-podium--item"): a = li.css_first("a.f1-podium--link") if not a: continue href = urljoin(url, a.attributes.get("href", "")) sport = a.css_first(".MacBaslikKat").text(strip=True) name = a.css_first(".MacIsimleri").text(strip=True) time_span = a.css_first(".f1-podium--time") time_text = time_span.text(strip=True) timestamp = int(time_span.attributes.get("data-zaman")) key = f"[{sport}] {name}" if key in cached_keys: continue event_dt = datetime.fromtimestamp(timestamp, TZ) if time_text == "LIVE" or (start_dt <= event_dt < end_dt): events.append( { "sport": sport, "event": name, "link": href, "logo": LOGOS.get( sport, "https://i.gyazo.com/ec27417a9644ae517196494afa72d2b9.png", ), } ) return events async def main(client: httpx.AsyncClient) -> None: cached_urls = load_cache(CACHE_FILE, exp=14400) cached_count = len(cached_urls) urls.update(cached_urls) log.info(f"Collected {cached_count} event(s) from cache") if not (base_url := await get_base(client, MIRRORS)): log.warning("No working StreamEast mirrors") return log.info(f'Scraping from "{base_url}"') events = await get_events( client, base_url, set(cached_urls.keys()), ) log.info(f"Processing {len(events)} new URL(s)") for i, ev in enumerate(events, start=1): url = await safe_process_event( lambda: process_event(ev["link"], url_num=i), url_num=i, log=log, ) if url: entry = { "url": url, "logo": ev["logo"], "timestamp": now.timestamp(), } key = f"[{ev['sport']}] {ev['event']}" urls[key] = cached_urls[key] = entry if new_count := len(cached_urls) - cached_count: log.info(f"Collected and cached {new_count} new event(s)") else: log.info("No new events found") CACHE_FILE.write_text(json.dumps(cached_urls, indent=2), encoding="utf-8")