|
8 | 8 |
|
9 | 9 | import re |
10 | 10 | import time |
11 | | -from datetime import datetime, timedelta |
12 | | -from typing import Tuple |
| 11 | +from datetime import datetime, timedelta, timezone |
| 12 | +from typing import Optional, Tuple |
13 | 13 |
|
14 | 14 |
|
15 | 15 | def _extract_from_regex(duration: str) -> Tuple[int, str]: |
@@ -93,3 +93,253 @@ def duration_in_seconds(duration: str) -> int: |
93 | 93 |
|
94 | 94 | else: |
95 | 95 | raise ValueError(f"Unsupported duration unit, passed duration: {duration}") |
| 96 | + |
| 97 | + |
| 98 | +def get_next_standardized_reset_time( |
| 99 | + duration: str, current_time: datetime, timezone_str: str = "UTC" |
| 100 | +) -> datetime: |
| 101 | + """ |
| 102 | + Get the next standardized reset time based on the duration. |
| 103 | +
|
| 104 | + All durations will reset at predictable intervals, aligned from the current time: |
| 105 | + - Nd: If N=1, reset at next midnight; if N>1, reset every N days from now |
| 106 | + - Nh: Every N hours, aligned to hour boundaries (e.g., 1:00, 2:00) |
| 107 | + - Nm: Every N minutes, aligned to minute boundaries (e.g., 1:05, 1:10) |
| 108 | + - Ns: Every N seconds, aligned to second boundaries |
| 109 | +
|
| 110 | + Parameters: |
| 111 | + - duration: Duration string (e.g. "30s", "30m", "30h", "30d") |
| 112 | + - current_time: Current datetime |
| 113 | + - timezone_str: Timezone string (e.g. "UTC", "US/Eastern", "Asia/Kolkata") |
| 114 | +
|
| 115 | + Returns: |
| 116 | + - Next reset time at a standardized interval in the specified timezone |
| 117 | + """ |
| 118 | + # Set up timezone and normalize current time |
| 119 | + current_time, timezone = _setup_timezone(current_time, timezone_str) |
| 120 | + |
| 121 | + # Parse duration |
| 122 | + value, unit = _parse_duration(duration) |
| 123 | + if value is None: |
| 124 | + # Fall back to default if format is invalid |
| 125 | + return current_time.replace( |
| 126 | + hour=0, minute=0, second=0, microsecond=0 |
| 127 | + ) + timedelta(days=1) |
| 128 | + |
| 129 | + # Midnight of the current day in the specified timezone |
| 130 | + base_midnight = current_time.replace(hour=0, minute=0, second=0, microsecond=0) |
| 131 | + |
| 132 | + # Handle different time units |
| 133 | + if unit == "d": |
| 134 | + return _handle_day_reset(current_time, base_midnight, value, timezone) |
| 135 | + elif unit == "h": |
| 136 | + return _handle_hour_reset(current_time, base_midnight, value) |
| 137 | + elif unit == "m": |
| 138 | + return _handle_minute_reset(current_time, base_midnight, value) |
| 139 | + elif unit == "s": |
| 140 | + return _handle_second_reset(current_time, base_midnight, value) |
| 141 | + else: |
| 142 | + # Unrecognized unit, default to next midnight |
| 143 | + return base_midnight + timedelta(days=1) |
| 144 | + |
| 145 | + |
| 146 | +def _setup_timezone( |
| 147 | + current_time: datetime, timezone_str: str = "UTC" |
| 148 | +) -> Tuple[datetime, timezone]: |
| 149 | + """Set up timezone and normalize current time to that timezone.""" |
| 150 | + try: |
| 151 | + if timezone_str is None: |
| 152 | + tz = timezone.utc |
| 153 | + else: |
| 154 | + # Map common timezone strings to their UTC offsets |
| 155 | + timezone_map = { |
| 156 | + "US/Eastern": timezone(timedelta(hours=-4)), # EDT |
| 157 | + "US/Pacific": timezone(timedelta(hours=-7)), # PDT |
| 158 | + "Asia/Kolkata": timezone(timedelta(hours=5, minutes=30)), # IST |
| 159 | + "Europe/London": timezone(timedelta(hours=1)), # BST |
| 160 | + "UTC": timezone.utc, |
| 161 | + } |
| 162 | + tz = timezone_map.get(timezone_str, timezone.utc) |
| 163 | + except Exception: |
| 164 | + # If timezone is invalid, fall back to UTC |
| 165 | + tz = timezone.utc |
| 166 | + |
| 167 | + # Convert current_time to the target timezone |
| 168 | + if current_time.tzinfo is None: |
| 169 | + # Naive datetime - assume it's UTC |
| 170 | + utc_time = current_time.replace(tzinfo=timezone.utc) |
| 171 | + current_time = utc_time.astimezone(tz) |
| 172 | + else: |
| 173 | + # Already has timezone - convert to target timezone |
| 174 | + current_time = current_time.astimezone(tz) |
| 175 | + |
| 176 | + return current_time, tz |
| 177 | + |
| 178 | + |
| 179 | +def _parse_duration(duration: str) -> Tuple[Optional[int], Optional[str]]: |
| 180 | + """Parse the duration string into value and unit.""" |
| 181 | + match = re.match(r"(\d+)([a-z]+)", duration) |
| 182 | + if not match: |
| 183 | + return None, None |
| 184 | + |
| 185 | + value, unit = match.groups() |
| 186 | + return int(value), unit |
| 187 | + |
| 188 | + |
| 189 | +def _handle_day_reset( |
| 190 | + current_time: datetime, base_midnight: datetime, value: int, timezone: timezone |
| 191 | +) -> datetime: |
| 192 | + """Handle day-based reset times.""" |
| 193 | + if value == 1: # Daily reset at midnight |
| 194 | + return base_midnight + timedelta(days=1) |
| 195 | + elif value == 7: # Weekly reset on Monday at midnight |
| 196 | + days_until_monday = (7 - current_time.weekday()) % 7 |
| 197 | + if days_until_monday == 0: # If today is Monday |
| 198 | + days_until_monday = 7 |
| 199 | + return base_midnight + timedelta(days=days_until_monday) |
| 200 | + elif value == 30: # Monthly reset on 1st at midnight |
| 201 | + # Get 1st of next month at midnight |
| 202 | + if current_time.month == 12: |
| 203 | + next_reset = datetime( |
| 204 | + year=current_time.year + 1, |
| 205 | + month=1, |
| 206 | + day=1, |
| 207 | + hour=0, |
| 208 | + minute=0, |
| 209 | + second=0, |
| 210 | + microsecond=0, |
| 211 | + tzinfo=timezone, |
| 212 | + ) |
| 213 | + else: |
| 214 | + next_reset = datetime( |
| 215 | + year=current_time.year, |
| 216 | + month=current_time.month + 1, |
| 217 | + day=1, |
| 218 | + hour=0, |
| 219 | + minute=0, |
| 220 | + second=0, |
| 221 | + microsecond=0, |
| 222 | + tzinfo=timezone, |
| 223 | + ) |
| 224 | + return next_reset |
| 225 | + else: # Custom day value - next interval is value days from current |
| 226 | + return current_time.replace( |
| 227 | + hour=0, minute=0, second=0, microsecond=0 |
| 228 | + ) + timedelta(days=value) |
| 229 | + |
| 230 | + |
| 231 | +def _handle_hour_reset( |
| 232 | + current_time: datetime, base_midnight: datetime, value: int |
| 233 | +) -> datetime: |
| 234 | + """Handle hour-based reset times.""" |
| 235 | + current_hour = current_time.hour |
| 236 | + current_minute = current_time.minute |
| 237 | + current_second = current_time.second |
| 238 | + current_microsecond = current_time.microsecond |
| 239 | + |
| 240 | + # Calculate next hour aligned with the value |
| 241 | + if current_minute == 0 and current_second == 0 and current_microsecond == 0: |
| 242 | + next_hour = ( |
| 243 | + current_hour + value - (current_hour % value) |
| 244 | + if current_hour % value != 0 |
| 245 | + else current_hour + value |
| 246 | + ) |
| 247 | + else: |
| 248 | + next_hour = ( |
| 249 | + current_hour + value - (current_hour % value) |
| 250 | + if current_hour % value != 0 |
| 251 | + else current_hour + value |
| 252 | + ) |
| 253 | + |
| 254 | + # Handle overnight case |
| 255 | + if next_hour >= 24: |
| 256 | + next_hour = next_hour % 24 |
| 257 | + next_day = base_midnight + timedelta(days=1) |
| 258 | + return next_day.replace(hour=next_hour) |
| 259 | + |
| 260 | + return current_time.replace(hour=next_hour, minute=0, second=0, microsecond=0) |
| 261 | + |
| 262 | + |
| 263 | +def _handle_minute_reset( |
| 264 | + current_time: datetime, base_midnight: datetime, value: int |
| 265 | +) -> datetime: |
| 266 | + """Handle minute-based reset times.""" |
| 267 | + current_hour = current_time.hour |
| 268 | + current_minute = current_time.minute |
| 269 | + current_second = current_time.second |
| 270 | + current_microsecond = current_time.microsecond |
| 271 | + |
| 272 | + # Calculate next minute aligned with the value |
| 273 | + if current_second == 0 and current_microsecond == 0: |
| 274 | + next_minute = ( |
| 275 | + current_minute + value - (current_minute % value) |
| 276 | + if current_minute % value != 0 |
| 277 | + else current_minute + value |
| 278 | + ) |
| 279 | + else: |
| 280 | + next_minute = ( |
| 281 | + current_minute + value - (current_minute % value) |
| 282 | + if current_minute % value != 0 |
| 283 | + else current_minute + value |
| 284 | + ) |
| 285 | + |
| 286 | + # Handle hour rollover |
| 287 | + next_hour = current_hour + (next_minute // 60) |
| 288 | + next_minute = next_minute % 60 |
| 289 | + |
| 290 | + # Handle overnight case |
| 291 | + if next_hour >= 24: |
| 292 | + next_hour = next_hour % 24 |
| 293 | + next_day = base_midnight + timedelta(days=1) |
| 294 | + return next_day.replace( |
| 295 | + hour=next_hour, minute=next_minute, second=0, microsecond=0 |
| 296 | + ) |
| 297 | + |
| 298 | + return current_time.replace( |
| 299 | + hour=next_hour, minute=next_minute, second=0, microsecond=0 |
| 300 | + ) |
| 301 | + |
| 302 | + |
| 303 | +def _handle_second_reset( |
| 304 | + current_time: datetime, base_midnight: datetime, value: int |
| 305 | +) -> datetime: |
| 306 | + """Handle second-based reset times.""" |
| 307 | + current_hour = current_time.hour |
| 308 | + current_minute = current_time.minute |
| 309 | + current_second = current_time.second |
| 310 | + current_microsecond = current_time.microsecond |
| 311 | + |
| 312 | + # Calculate next second aligned with the value |
| 313 | + if current_microsecond == 0: |
| 314 | + next_second = ( |
| 315 | + current_second + value - (current_second % value) |
| 316 | + if current_second % value != 0 |
| 317 | + else current_second + value |
| 318 | + ) |
| 319 | + else: |
| 320 | + next_second = ( |
| 321 | + current_second + value - (current_second % value) |
| 322 | + if current_second % value != 0 |
| 323 | + else current_second + value |
| 324 | + ) |
| 325 | + |
| 326 | + # Handle minute rollover |
| 327 | + additional_minutes = next_second // 60 |
| 328 | + next_second = next_second % 60 |
| 329 | + next_minute = current_minute + additional_minutes |
| 330 | + |
| 331 | + # Handle hour rollover |
| 332 | + next_hour = current_hour + (next_minute // 60) |
| 333 | + next_minute = next_minute % 60 |
| 334 | + |
| 335 | + # Handle overnight case |
| 336 | + if next_hour >= 24: |
| 337 | + next_hour = next_hour % 24 |
| 338 | + next_day = base_midnight + timedelta(days=1) |
| 339 | + return next_day.replace( |
| 340 | + hour=next_hour, minute=next_minute, second=next_second, microsecond=0 |
| 341 | + ) |
| 342 | + |
| 343 | + return current_time.replace( |
| 344 | + hour=next_hour, minute=next_minute, second=next_second, microsecond=0 |
| 345 | + ) |
0 commit comments