utils.py

from curl_cffi import requests
from playwright.sync_api import sync_playwright
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

IST = ZoneInfo("Asia/Kolkata")
ET = ZoneInfo("America/New_York")
from config import HTTP_REQUEST_TIMEOUT_SECONDS, MARKET_CONFIG
from logger import logger


def fetch_html(url: str):
    """
    Fetch HTML using requests first.
    Fallback to Playwright for JS-heavy or protected sites.

    Returns:
        (html, error)
    """

    headers = {
        "User-Agent": (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/136.0.0.0 Safari/537.36"
        ),
        "Accept": (
            "text/html,application/xhtml+xml,"
            "application/xml;q=0.9,image/avif,"
            "image/webp,*/*;q=0.8"
        ),
        "Accept-Language": "en-US,en;q=0.5",
        "Connection": "keep-alive",
    }

    #
    # First try direct HTTP
    #
    try:
        logger.info(f"HTTP GET {url}")

        response = requests.get(
            url,
            headers=headers,
            timeout=HTTP_REQUEST_TIMEOUT_SECONDS,
            impersonate="chrome",
        )

        response.raise_for_status()

        html = response.text

        if html and len(html) > 1000:
            logger.info(f"HTTP succeeded ({len(html)} chars)")
            return html, None

        logger.warning(
            "HTTP returned suspiciously small response. " "Falling back to Playwright."
        )

    except Exception as e:
        logger.warning(f"HTTP fetch failed ({e}). " "Falling back to Playwright.")

    #
    # Fallback: Playwright
    #
    try:
        with sync_playwright() as p:

            browser = p.chromium.launch(
                headless=True,
                args=[
                    "--disable-http2",
                    "--disable-blink-features=AutomationControlled",
                    "--window-size=1920,1080",
                ],
            )

            context = browser.new_context(
                user_agent=headers["User-Agent"],
                viewport={"width": 1920, "height": 1080},
            )

            page = context.new_page()

            if "nseindia.com" in url:
                page.goto(
                    "https://www.nseindia.com",
                    wait_until="domcontentloaded",
                    timeout=30000,
                )
                page.wait_for_timeout(2000)

            page.goto(
                url,
                wait_until="domcontentloaded",
                timeout=45000,
            )

            page.wait_for_timeout(1000)

            html = page.content()

            logger.info(f"Playwright succeeded ({len(html)} chars)")

            return html, None

    except Exception as e:
        logger.exception(f"Playwright fetch failed for {url}")
        return None, str(e)

    finally:
        try:
            context.close()
            browser.close()
        except Exception:
            pass


def seconds_until_next_monday():
    now = datetime.now(IST)

    # Monday 00:00 IST
    days_until_monday = (7 - now.weekday()) % 7

    if days_until_monday == 0:
        days_until_monday = 7

    next_monday = now.replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(
        days=days_until_monday
    )

    return max(1, int((next_monday - now).total_seconds()))


# =========================
# MARKET-CLOSE SCHEDULING
# =========================
# Generalizes seconds_until_next_monday() to "next weekday close for a market".
# The stock/macro workers wake at the soonest close across markets, fetch the
# rows for whichever market closed today, then sleep again to the next close.


def _market_tz(market: str) -> ZoneInfo:
    """ZoneInfo for a market key in MARKET_CONFIG (defaults to IST)."""

    cfg = MARKET_CONFIG.get(market)
    return ZoneInfo(cfg["tz"]) if cfg else IST


def _market_close_today(market: str, now: datetime) -> datetime:
    """Today's close datetime (tz-aware) for `market`, on the date of `now`."""

    hour, minute = MARKET_CONFIG[market]["close"]
    return now.replace(hour=hour, minute=minute, second=0, microsecond=0)


def market_already_closed_today(market: str) -> bool:
    """True if `market` is on a weekday and its close time has already passed."""

    now = datetime.now(_market_tz(market))

    # Saturday=5, Sunday=6 — markets shut.
    if now.weekday() in (5, 6):
        return False

    return now >= _market_close_today(market, now)


def seconds_until_next_market_close(market: str) -> int:
    """Seconds until `market`'s next close, skipping weekends.

    If today is a weekday and the close hasn't happened yet, that's the target.
    Otherwise advance day-by-day to the next weekday's close.
    """

    now = datetime.now(_market_tz(market))
    close = _market_close_today(market, now)

    # Move to the next valid close: future + on a weekday.
    while close <= now or close.weekday() in (5, 6):
        close = close + timedelta(days=1)

    return max(1, int((close - now).total_seconds()))


def get_market_day_start_ms(market: str) -> int:
    """Epoch-ms for the start (00:00) of the current day in `market`'s tz.

    Used to gate "already fetched today": a row whose last_fetched_at is >= this
    value has been fetched during the current market day and is skipped.
    """

    now = datetime.now(_market_tz(market))
    start = now.replace(hour=0, minute=0, second=0, microsecond=0)
    return int(start.timestamp() * 1000)