global_func/contest_lobby_cache.py

"""Persist pricing-step contest lists in Mongo (Contest_Information DB)."""
from datetime import date, datetime
from typing import List, Optional

import pandas as pd
import pytz

COLLECTION_NAME = "DFS_PM_pricing_contest_cache"
_INDEX_ENSURED = False
_EASTERN = pytz.timezone("US/Eastern")


def _eastern_today() -> date:
    return datetime.now(_EASTERN).date()


def _parse_contest_slate_date(contest: dict) -> Optional[date]:
    """Slate date from cached contest dict (DRAFTKINGS YYYYMMDD or YYYY-MM-DD display), or None if unknown."""
    cd = contest.get("contest_date")
    if isinstance(cd, str) and len(cd) == 8 and cd.isdigit():
        try:
            return datetime.strptime(cd, "%Y%m%d").date()
        except ValueError:
            pass
    disp = contest.get("contest_date_display")
    if isinstance(disp, str):
        s = disp.strip()
        if len(s) >= 10:
            try:
                return datetime.strptime(s[:10], "%Y-%m-%d").date()
            except ValueError:
                pass
    return None


def filter_contests_from_today_forward(contests: List[dict]) -> List[dict]:
    """
    Keep contests whose slate date is today (US/Eastern) or later.
    Entries with no parseable date are kept (e.g. FanDuel CSV uploads without a date column).
    """
    today = _eastern_today()
    out: List[dict] = []
    for c in contests:
        d = _parse_contest_slate_date(c)
        if d is None or d >= today:
            out.append(c)
    return out


def _collection(db):
    return db[COLLECTION_NAME]


def ensure_contest_cache_indexes(db) -> None:
    global _INDEX_ENSURED
    if _INDEX_ENSURED:
        return
    try:
        _collection(db).create_index(
            [("site", 1), ("sport", 1), ("type", 1)],
            unique=True,
            name="site_sport_type_unique",
        )
    except Exception:
        pass
    _INDEX_ENSURED = True


def cache_has_contests(db, site: str, sport: str, game_type: str) -> bool:
    """True if cache has at least one non-expired contest (slate date >= today US/Eastern)."""
    cached = get_cached_contests(db, site, sport, game_type)
    return cached is not None and len(cached) > 0


def get_cached_contests(db, site: str, sport: str, game_type: str) -> Optional[List[dict]]:
    """Return cached contest list (today-forward only), or None if no document exists."""
    ensure_contest_cache_indexes(db)
    key = {"site": site, "sport": sport, "type": game_type}
    doc = _collection(db).find_one(key)
    if doc is None:
        return None
    contests = doc.get("contests")
    if contests is None:
        return []
    raw = list(contests)
    filtered = filter_contests_from_today_forward(raw)
    if len(filtered) != len(raw):
        _collection(db).update_one(
            key,
            {"$set": {"contests": filtered, "updated_at": datetime.utcnow()}},
        )
    return filtered


def save_contests_cache(db, site: str, sport: str, game_type: str, contests: List[dict]) -> None:
    """Replace cached contests with the payload after removing prior-day slates (US/Eastern)."""
    ensure_contest_cache_indexes(db)
    to_store = filter_contests_from_today_forward(list(contests))
    _collection(db).update_one(
        {"site": site, "sport": sport, "type": game_type},
        {
            "$set": {
                "contests": to_store,
                "updated_at": datetime.utcnow(),
            }
        },
        upsert=True,
    )


def parse_fd_contest_id_csv(df: pd.DataFrame) -> List[dict]:
    """Build contest dicts for the pricing dropdown from a FanDuel-oriented upload."""
    if df is None or df.empty:
        return []

    cols = {str(c).strip(): c for c in df.columns}
    id_key = None
    for label in ("contest_id", "Contest ID", "contestId", "ID", "Id", "id"):
        if label in cols:
            id_key = cols[label]
            break
    if id_key is None:
        first = list(df.columns)[0]
        id_key = first

    name_key = None
    for label in ("contest_name", "Contest Name", "Contest", "Name", "n", "nickname"):
        if label in cols:
            name_key = cols[label]
            break

    out: List[dict] = []
    for _, row in df.iterrows():
        raw_id = row.get(id_key)
        if pd.isna(raw_id):
            continue
        s = str(raw_id).strip().replace(",", "")
        if not s:
            continue
        try:
            cid = int(float(s))
        except (ValueError, TypeError):
            cid = s
        name = ""
        if name_key is not None:
            v = row.get(name_key)
            if pd.notna(v):
                name = str(v).strip()
        if not name:
            name = f"Contest {cid}"
        out.append(
            {
                "contest_name": name,
                "contest_id": cid,
                "contest_date": "",
                "contest_date_display": "",
                "game_type": "",
                "prize_pool": 0,
            }
        )
    return out