Spaces:
Build error
Build error
File size: 4,978 Bytes
212b005 50daa3e 212b005 50daa3e 212b005 50daa3e 212b005 50daa3e 212b005 50daa3e 212b005 50daa3e 212b005 50daa3e 212b005 50daa3e 212b005 50daa3e 212b005 50daa3e 212b005 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | """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
|