Compare commits
2 Commits
b2180a4a34
...
a7837cec0f
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a7837cec0f | ||
|
|
8cde441996 |
@@ -19,6 +19,7 @@ from urllib.request import urlopen, Request
|
||||
PAGE_SIZE = 50
|
||||
MAX_RETRIES = 5
|
||||
INITIAL_RETRY_DELAY = 2 # exponential backoff starts at 2s
|
||||
WIB = timezone(timedelta(hours=7)) # UTC+7 for Indonesian users
|
||||
|
||||
GAME_CATEGORIES = {
|
||||
"All Esports": "Esports",
|
||||
@@ -221,94 +222,79 @@ def format_spread(bid, ask):
|
||||
spread = ask - bid
|
||||
return f"{prob_to_cents(spread)}c"
|
||||
|
||||
def get_match_time_status(e):
|
||||
|
||||
def _get_time_data(e, tz=None):
|
||||
"""
|
||||
Return a human-readable match time status.
|
||||
Returns (status_str, urgency) where urgency is 0-3 (higher = more urgent/live).
|
||||
Uses startTime for actual match start time.
|
||||
Displays times in WIB (UTC+7 for Indonesian users).
|
||||
Unified time data extraction for event timestamps.
|
||||
|
||||
Uses startTime (preferred) or startDate as the event start time.
|
||||
Datetime parsing and all relative calculations are UTC-based.
|
||||
The tz parameter only affects the abs_time formatting.
|
||||
|
||||
Args:
|
||||
e: Event dict with 'startTime' or 'startDate' key.
|
||||
tz: datetime.timezone for abs_time formatting.
|
||||
Defaults to WIB (UTC+7).
|
||||
|
||||
Returns:
|
||||
{
|
||||
"time_status": str, # e.g. "LIVE", "In 6h", "12h ago"
|
||||
"time_urgency": int, # 0-3 (higher = more urgent/live)
|
||||
"abs_time": str, # e.g. "Mar 25, 19:00 WIB" or "TBD"
|
||||
}
|
||||
"""
|
||||
# Use startTime for actual match start, not startDate (which is market creation time)
|
||||
tz = tz or WIB
|
||||
start_str = e.get("startTime") or e.get("startDate", "")
|
||||
|
||||
if not start_str:
|
||||
return "TBD", 0
|
||||
return {"time_status": "TBD", "time_urgency": 0, "abs_time": "TBD"}
|
||||
|
||||
try:
|
||||
start_dt = datetime.fromisoformat(start_str.replace('Z', '+00:00'))
|
||||
now_utc = datetime.now(timezone.utc)
|
||||
utc7 = timezone(timedelta(hours=7))
|
||||
now = now_utc.astimezone(utc7)
|
||||
start_utc7 = start_dt.astimezone(utc7)
|
||||
|
||||
delta = start_dt - now_utc
|
||||
total_sec = delta.total_seconds()
|
||||
|
||||
if delta.total_seconds() < 0:
|
||||
# Started already
|
||||
hours_ago = abs(delta.total_seconds()) / 3600
|
||||
if total_sec < 0:
|
||||
# Event is in the past
|
||||
hours_ago = abs(total_sec) / 3600
|
||||
if hours_ago < 1:
|
||||
return "LIVE", 3
|
||||
time_status = "LIVE"
|
||||
time_urgency = 3
|
||||
elif hours_ago < 4:
|
||||
return f"LIVE {int(hours_ago)}h", 3
|
||||
time_status = f"LIVE {int(hours_ago)}h"
|
||||
time_urgency = 3
|
||||
elif hours_ago < 24:
|
||||
return f"Started {int(hours_ago)}h ago", 1
|
||||
time_status = f"{int(hours_ago)}h ago"
|
||||
time_urgency = 1
|
||||
else:
|
||||
days = int(hours_ago / 24)
|
||||
return f"{days}d ago", 0
|
||||
time_status = f"{days}d ago"
|
||||
time_urgency = 0
|
||||
else:
|
||||
# Starts in future
|
||||
hours_until = delta.total_seconds() / 3600
|
||||
if hours_until <= 0:
|
||||
return "LIVE", 3
|
||||
elif hours_until < 1:
|
||||
mins = int(delta.total_seconds() / 60)
|
||||
return f"In {mins}m", 3
|
||||
elif hours_until < 24:
|
||||
return f"In {int(hours_until)}h", 2
|
||||
# Event is in the future
|
||||
if total_sec < 3600:
|
||||
mins = int(total_sec / 60)
|
||||
time_status = f"In {mins}m"
|
||||
time_urgency = 3
|
||||
elif total_sec < 86400:
|
||||
hours_until = int(total_sec / 3600)
|
||||
time_status = f"In {hours_until}h"
|
||||
time_urgency = 2
|
||||
else:
|
||||
days = int(hours_until / 24)
|
||||
return f"In {days}d", 1
|
||||
except:
|
||||
return "", 0
|
||||
days = int(total_sec / 86400)
|
||||
time_status = f"In {days}d"
|
||||
time_urgency = 1
|
||||
|
||||
def get_match_time_str(e):
|
||||
"""
|
||||
Return just the time status string (e.g. 'LIVE', 'In 6h', 'In 1d').
|
||||
Uses startTime for actual match start time.
|
||||
"""
|
||||
start_str = e.get("startTime") or e.get("startDate", "")
|
||||
if not start_str:
|
||||
return "TBD"
|
||||
try:
|
||||
start_dt = datetime.fromisoformat(start_str.replace('Z', '+00:00'))
|
||||
now_utc = datetime.now(timezone.utc)
|
||||
delta = start_dt - now_utc
|
||||
abs_time = start_dt.astimezone(tz).strftime("%b %d, %H:%M ")
|
||||
if tz == WIB:
|
||||
abs_time += "WIB"
|
||||
else:
|
||||
abs_time += start_dt.astimezone(tz).strftime("%Z")
|
||||
return {"time_status": time_status, "time_urgency": time_urgency, "abs_time": abs_time}
|
||||
except Exception:
|
||||
return {"time_status": "", "time_urgency": 0, "abs_time": "TBD"}
|
||||
|
||||
if delta.total_seconds() < 0:
|
||||
hours_ago = abs(delta.total_seconds()) / 3600
|
||||
if hours_ago < 1:
|
||||
return "LIVE"
|
||||
elif hours_ago < 4:
|
||||
return f"LIVE {int(hours_ago)}h"
|
||||
elif hours_ago < 24:
|
||||
return f"{int(hours_ago)}h ago"
|
||||
else:
|
||||
days = int(hours_ago / 24)
|
||||
return f"{days}d ago"
|
||||
else:
|
||||
hours_until = delta.total_seconds() / 3600
|
||||
if hours_until <= 0:
|
||||
return "LIVE"
|
||||
elif hours_until < 1:
|
||||
mins = int(delta.total_seconds() / 60)
|
||||
return f"In {mins}m"
|
||||
elif hours_until < 24:
|
||||
return f"In {int(hours_until)}h"
|
||||
else:
|
||||
days = int(hours_until / 24)
|
||||
return f"In {days}d"
|
||||
except:
|
||||
return ""
|
||||
|
||||
def filter_events(events, tradeable_only=True):
|
||||
"""
|
||||
@@ -327,6 +313,7 @@ def filter_events(events, tradeable_only=True):
|
||||
|
||||
return match_events, non_match_events
|
||||
|
||||
|
||||
def sort_events(events):
|
||||
return sorted(events, key=get_ml_volume, reverse=True)
|
||||
|
||||
@@ -361,12 +348,12 @@ def format_event(e):
|
||||
best_bid = float(ml.get("bestBid", 0)) if ml else 0
|
||||
best_ask = float(ml.get("bestAsk", 0)) if ml else 0
|
||||
vol = get_ml_volume(e)
|
||||
time_status, urgency = get_match_time_status(e)
|
||||
td = _get_time_data(e)
|
||||
|
||||
return {
|
||||
"title": e.get("title", ""),
|
||||
"time_status": time_status,
|
||||
"time_urgency": urgency,
|
||||
"time_status": td["time_status"],
|
||||
"time_urgency": td["time_urgency"],
|
||||
"url": get_event_url(e),
|
||||
"livestream": e.get("resolutionSource"),
|
||||
"outcomes": outcomes,
|
||||
@@ -385,11 +372,12 @@ def format_detail_event(e):
|
||||
]
|
||||
active_markets = sorted(active_markets, key=lambda m: float(m.get("volume", 0)), reverse=True)
|
||||
|
||||
time_status, urgency = get_match_time_status(e)
|
||||
td = _get_time_data(e)
|
||||
|
||||
return {
|
||||
"title": e.get("title", ""),
|
||||
"time_status": time_status,
|
||||
"time_status": td["time_status"],
|
||||
"abs_time": td["abs_time"],
|
||||
"url": get_event_url(e),
|
||||
"livestream": e.get("resolutionSource"),
|
||||
"outcomes": json.loads(ml.get("outcomes", "[]")) if ml else [],
|
||||
@@ -416,48 +404,6 @@ def format_detail_event(e):
|
||||
# DISPLAY
|
||||
# ============================================================
|
||||
|
||||
def get_start_time_wib(e):
|
||||
"""Return (date_time_str, relative_str) for display."""
|
||||
start_str = e.get("startTime") or e.get("startDate", "")
|
||||
if not start_str:
|
||||
return "TBD", ""
|
||||
try:
|
||||
start_dt = datetime.fromisoformat(start_str.replace('Z', '+00:00'))
|
||||
now_utc = datetime.now(timezone.utc)
|
||||
utc7 = timezone(timedelta(hours=7))
|
||||
start_utc7 = start_dt.astimezone(utc7)
|
||||
|
||||
# Absolute: "Mar 25, 19:00 WIB"
|
||||
abs_str = start_utc7.strftime("%b %d, %H:%M WIB")
|
||||
|
||||
# Relative: "In 5h", "In 10h", "LIVE", etc.
|
||||
delta = start_dt - now_utc
|
||||
if delta.total_seconds() < 0:
|
||||
hours_ago = abs(delta.total_seconds()) / 3600
|
||||
if hours_ago < 1:
|
||||
rel_str = "LIVE"
|
||||
elif hours_ago < 24:
|
||||
rel_str = f"{int(hours_ago)}h ago"
|
||||
else:
|
||||
days = int(hours_ago / 24)
|
||||
rel_str = f"{days}d ago"
|
||||
else:
|
||||
hours_until = delta.total_seconds() / 3600
|
||||
if hours_until <= 0:
|
||||
rel_str = "LIVE"
|
||||
elif hours_until < 1:
|
||||
mins_until = int(delta.total_seconds() / 60)
|
||||
rel_str = f"In {mins_until}m"
|
||||
elif hours_until < 24:
|
||||
rel_str = f"In {int(hours_until)}h"
|
||||
else:
|
||||
days = int(hours_until / 24)
|
||||
rel_str = f"In {days}d"
|
||||
|
||||
return abs_str, rel_str
|
||||
except:
|
||||
return "TBD", ""
|
||||
|
||||
def get_header_date():
|
||||
"""Return current date string like 'Mar 25, 2026'"""
|
||||
now_utc = datetime.now(timezone.utc)
|
||||
@@ -513,7 +459,9 @@ def print_browse(match_events, non_match_events, category, total_raw, total_fetc
|
||||
vol = f["volume"]
|
||||
title = f["title"]
|
||||
url = f["url"]
|
||||
start_time_wib, rel_time = get_start_time_wib(e)
|
||||
td = _get_time_data(e)
|
||||
start_time_wib = td["abs_time"]
|
||||
rel_time = td["time_status"]
|
||||
|
||||
team_a = outcomes[0] if len(outcomes) > 0 else "?"
|
||||
team_b = outcomes[1] if len(outcomes) > 1 else "?"
|
||||
@@ -541,7 +489,9 @@ def print_browse(match_events, non_match_events, category, total_raw, total_fetc
|
||||
for i, e in enumerate(non_match_events[:non_matches_max], 1):
|
||||
title = e.get("title", "?")
|
||||
url = get_event_url(e)
|
||||
start_time_wib, rel_time = get_start_time_wib(e)
|
||||
td = _get_time_data(e)
|
||||
start_time_wib = td["abs_time"]
|
||||
rel_time = td["time_status"]
|
||||
|
||||
total_vol = sum(float(m.get("volume", 0)) for m in e.get("markets", []))
|
||||
market_count = len(e.get("markets", []))
|
||||
@@ -551,17 +501,11 @@ def print_browse(match_events, non_match_events, category, total_raw, total_fetc
|
||||
print(f" Markets: {market_count} | Total Vol: ${total_vol:,.0f}")
|
||||
|
||||
def print_detail(e, detail):
|
||||
from datetime import datetime, timezone, timedelta
|
||||
now_utc = datetime.now(timezone.utc)
|
||||
utc7 = timezone(timedelta(hours=7))
|
||||
now_utc7 = now_utc.astimezone(utc7)
|
||||
|
||||
print(f"\n{detail['title']}")
|
||||
print(f"URL: {detail['url']}")
|
||||
print(f"Livestream: {detail['livestream']}")
|
||||
|
||||
spread_str = format_spread(detail["best_bid"], detail["best_ask"]) if detail["best_bid"] and detail["best_ask"] else "N/A"
|
||||
time_str = get_match_time_str(e)
|
||||
print(f"\n{detail['time_status']}")
|
||||
print(f"ML: {detail['outcomes'][0]} {format_odds(float(detail['prices'][0]))} vs {detail['outcomes'][1]} {format_odds(float(detail['prices'][1]))}")
|
||||
print(f"ML Vol: ${detail['volume']:,.0f} | {spread_str}")
|
||||
@@ -648,7 +592,9 @@ def send_to_telegram(match_events, non_match_events, category, matches_only=Fals
|
||||
vol = get_ml_volume(e)
|
||||
title = e.get("title", "?")
|
||||
url = get_event_url(e)
|
||||
start_time_wib, rel_time = get_start_time_wib(e)
|
||||
td = _get_time_data(e)
|
||||
start_time_wib = td["abs_time"]
|
||||
rel_time = td["time_status"]
|
||||
team_a = outcomes[0] if len(outcomes) > 0 else "?"
|
||||
team_b = outcomes[1] if len(outcomes) > 1 else "?"
|
||||
odds_a = format_odds(float(prices[0])) if len(prices) > 0 else "?"
|
||||
@@ -673,7 +619,9 @@ def send_to_telegram(match_events, non_match_events, category, matches_only=Fals
|
||||
for i, e in enumerate(non_match_events, 1):
|
||||
title = e.get("title", "?")
|
||||
url = get_event_url(e)
|
||||
start_time_wib, rel_time = get_start_time_wib(e)
|
||||
td = _get_time_data(e)
|
||||
start_time_wib = td["abs_time"]
|
||||
rel_time = td["time_status"]
|
||||
total_vol = sum(float(m.get("volume", 0)) for m in e.get("markets", []))
|
||||
market_count = len(e.get("markets", []))
|
||||
lines.append(f"<b>{i}.</b> <a href=\"{url}\">{escape_html(title)}</a>")
|
||||
|
||||
@@ -8,6 +8,7 @@ import unittest
|
||||
from unittest.mock import patch, MagicMock
|
||||
import sys
|
||||
import os
|
||||
from datetime import datetime, timezone, timedelta
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'scripts'))
|
||||
from browse import send_telegram_message
|
||||
@@ -201,5 +202,123 @@ class TestHtmlInjection(unittest.TestCase):
|
||||
"Ampersand not escaped — title may NOT be escaped")
|
||||
|
||||
|
||||
class TestTimeFunctions(unittest.TestCase):
|
||||
"""Tests for _get_time_data() unified helper.
|
||||
|
||||
These tests verify the helper returns correct time_status, time_urgency,
|
||||
and abs_time for various event scenarios. Callers extract the fields they
|
||||
need from the returned dict.
|
||||
"""
|
||||
|
||||
def _make_event(self, start_time):
|
||||
"""Helper to create a minimal event with a startTime."""
|
||||
return {"startTime": start_time}
|
||||
|
||||
def _frozen_dt(self, year, month, day, hour, minute, second=0):
|
||||
return datetime(year, month, day, hour, minute, second,
|
||||
tzinfo=timezone.utc)
|
||||
|
||||
def _mock_datetime(self, frozen):
|
||||
"""Return a mock datetime class that freezes now() to the given datetime."""
|
||||
class MockDatetime:
|
||||
@staticmethod
|
||||
def now(tz=None):
|
||||
if tz is None:
|
||||
return frozen
|
||||
return frozen.astimezone(tz)
|
||||
fromisoformat = staticmethod(datetime.fromisoformat)
|
||||
def __call__(self, *a, **k):
|
||||
return datetime(*a, **k)
|
||||
return MockDatetime
|
||||
|
||||
# === _get_time_data core tests ===
|
||||
|
||||
def test_get_time_data_tbd(self):
|
||||
"""No startTime -> TBD/0urgency/abs TBD."""
|
||||
from browse import _get_time_data
|
||||
td = _get_time_data({})
|
||||
self.assertEqual(td["time_status"], "TBD")
|
||||
self.assertEqual(td["time_urgency"], 0)
|
||||
self.assertEqual(td["abs_time"], "TBD")
|
||||
|
||||
def test_get_time_data_in_30m(self):
|
||||
"""Starts in 30 minutes -> 'In 30m', urgency 3."""
|
||||
frozen = self._frozen_dt(2026, 3, 25, 12, 0, 0)
|
||||
with patch('browse.datetime', self._mock_datetime(frozen)):
|
||||
from browse import _get_time_data
|
||||
td = _get_time_data(self._make_event("2026-03-25T12:30:00Z"))
|
||||
self.assertEqual(td["time_status"], "In 30m")
|
||||
self.assertEqual(td["time_urgency"], 3)
|
||||
self.assertIn("WIB", td["abs_time"])
|
||||
|
||||
def test_get_time_data_in_6h(self):
|
||||
"""Starts in 6 hours -> 'In 6h', urgency 2."""
|
||||
frozen = self._frozen_dt(2026, 3, 25, 12, 0, 0)
|
||||
with patch('browse.datetime', self._mock_datetime(frozen)):
|
||||
from browse import _get_time_data
|
||||
td = _get_time_data(self._make_event("2026-03-25T18:00:00Z"))
|
||||
self.assertEqual(td["time_status"], "In 6h")
|
||||
self.assertEqual(td["time_urgency"], 2)
|
||||
self.assertIn("WIB", td["abs_time"])
|
||||
|
||||
def test_get_time_data_in_2d(self):
|
||||
"""Starts in 2 days -> 'In 2d', urgency 1."""
|
||||
frozen = self._frozen_dt(2026, 3, 25, 12, 0, 0)
|
||||
with patch('browse.datetime', self._mock_datetime(frozen)):
|
||||
from browse import _get_time_data
|
||||
td = _get_time_data(self._make_event("2026-03-27T12:00:00Z"))
|
||||
self.assertEqual(td["time_status"], "In 2d")
|
||||
self.assertEqual(td["time_urgency"], 1)
|
||||
|
||||
def test_get_time_data_live(self):
|
||||
"""Started 30 minutes ago -> 'LIVE', urgency 3."""
|
||||
frozen = self._frozen_dt(2026, 3, 25, 12, 30, 0)
|
||||
with patch('browse.datetime', self._mock_datetime(frozen)):
|
||||
from browse import _get_time_data
|
||||
td = _get_time_data(self._make_event("2026-03-25T12:00:00Z"))
|
||||
self.assertEqual(td["time_status"], "LIVE")
|
||||
self.assertEqual(td["time_urgency"], 3)
|
||||
self.assertIn("WIB", td["abs_time"])
|
||||
|
||||
def test_get_time_data_started_2h_ago(self):
|
||||
"""Started 2 hours ago -> 'LIVE 2h', urgency 3."""
|
||||
frozen = self._frozen_dt(2026, 3, 25, 14, 0, 0)
|
||||
with patch('browse.datetime', self._mock_datetime(frozen)):
|
||||
from browse import _get_time_data
|
||||
td = _get_time_data(self._make_event("2026-03-25T12:00:00Z"))
|
||||
self.assertEqual(td["time_status"], "LIVE 2h")
|
||||
self.assertEqual(td["time_urgency"], 3)
|
||||
|
||||
def test_get_time_data_started_12h_ago(self):
|
||||
"""Started 12 hours ago -> '12h ago', urgency 1."""
|
||||
frozen = self._frozen_dt(2026, 3, 26, 0, 0, 0)
|
||||
with patch('browse.datetime', self._mock_datetime(frozen)):
|
||||
from browse import _get_time_data
|
||||
td = _get_time_data(self._make_event("2026-03-25T12:00:00Z"))
|
||||
self.assertEqual(td["time_status"], "12h ago")
|
||||
self.assertEqual(td["time_urgency"], 1)
|
||||
|
||||
def test_get_time_data_started_2d_ago(self):
|
||||
"""Started 2 days ago -> '2d ago', urgency 0."""
|
||||
frozen = self._frozen_dt(2026, 3, 27, 12, 0, 0)
|
||||
with patch('browse.datetime', self._mock_datetime(frozen)):
|
||||
from browse import _get_time_data
|
||||
td = _get_time_data(self._make_event("2026-03-25T12:00:00Z"))
|
||||
self.assertEqual(td["time_status"], "2d ago")
|
||||
self.assertEqual(td["time_urgency"], 0)
|
||||
|
||||
def test_get_time_data_abs_time_format(self):
|
||||
"""abs_time is formatted correctly in WIB."""
|
||||
frozen = self._frozen_dt(2026, 3, 25, 12, 0, 0)
|
||||
with patch('browse.datetime', self._mock_datetime(frozen)):
|
||||
from browse import _get_time_data
|
||||
# 19:00 UTC = 02:00 WIB next day
|
||||
td = _get_time_data(self._make_event("2026-03-26T02:00:00Z"))
|
||||
self.assertIn("WIB", td["abs_time"])
|
||||
# UTC 12:00 -> WIB 19:00 same day
|
||||
td2 = _get_time_data(self._make_event("2026-03-25T12:00:00Z"))
|
||||
self.assertEqual(td2["abs_time"], "Mar 25, 19:00 WIB")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user