| 1 | from datetime import datetime, timezone as dt_timezone |
| 2 | import os |
| 3 | import time |
| 4 | import pytz # type: ignore |
| 5 | |
| 6 | from helpers.print_style import PrintStyle |
| 7 | from helpers.dotenv import get_dotenv_value, save_dotenv_value |
| 8 | |
| 9 | |
| 10 | |
| 11 | class Localization: |
| 12 | """ |
| 13 | Localization class for handling timezone conversions around the user's IANA |
| 14 | timezone. UTC is still used when an external protocol requires an absolute |
| 15 | instant, but user-facing timestamps are formatted in the configured timezone. |
| 16 | """ |
| 17 | |
| 18 | # singleton |
| 19 | _instance = None |
| 20 | |
| 21 | @classmethod |
| 22 | def get(cls, *args, **kwargs): |
| 23 | if cls._instance is None: |
| 24 | cls._instance = cls(*args, **kwargs) |
| 25 | return cls._instance |
| 26 | |
| 27 | def __init__(self, timezone: str | None = None): |
| 28 | self.timezone: str = "UTC" |
| 29 | self._offset_minutes: int = 0 |
| 30 | self._last_timezone_change: datetime | None = None |
| 31 | # Load persisted values if available. |
| 32 | persisted_tz = str(get_dotenv_value("DEFAULT_USER_TIMEZONE", os.environ.get("TZ") or "UTC")) |
| 33 | persisted_offset = get_dotenv_value("DEFAULT_USER_UTC_OFFSET_MINUTES", None) |
| 34 | if timezone is not None: |
| 35 | # Explicit override |
| 36 | self.set_timezone(timezone) |
| 37 | else: |
| 38 | # Initialize from persisted values |
| 39 | try: |
| 40 | pytz.timezone(persisted_tz) |
| 41 | self.timezone = persisted_tz |
| 42 | except pytz.exceptions.UnknownTimeZoneError: |
| 43 | self.timezone = "UTC" |
| 44 | current_offset = self._compute_offset_minutes(self.timezone) |
| 45 | try: |
| 46 | persisted_offset_minutes = int(str(persisted_offset)) if persisted_offset is not None else None |
| 47 | except Exception: |
| 48 | persisted_offset_minutes = None |
| 49 | self._offset_minutes = current_offset |
| 50 | if persisted_offset_minutes != current_offset: |
| 51 | save_dotenv_value("DEFAULT_USER_UTC_OFFSET_MINUTES", str(self._offset_minutes)) |
| 52 | self.apply_process_timezone() |
| 53 | |
| 54 | def get_timezone(self) -> str: |
| 55 | return self.timezone |
| 56 | |
| 57 | def get_tzinfo(self): |
| 58 | try: |
| 59 | return pytz.timezone(self.timezone) |
| 60 | except pytz.exceptions.UnknownTimeZoneError: |
| 61 | return pytz.timezone("UTC") |
| 62 | |
| 63 | def _compute_offset_minutes(self, timezone_name: str) -> int: |
| 64 | tzinfo = pytz.timezone(timezone_name) |
| 65 | now_in_tz = datetime.now(tzinfo) |
| 66 | offset = now_in_tz.utcoffset() |
| 67 | return int(offset.total_seconds() // 60) if offset else 0 |
| 68 | |
| 69 | def get_offset_minutes(self) -> int: |
| 70 | return self._offset_minutes |
| 71 | |
| 72 | def apply_process_timezone(self) -> None: |
| 73 | """Apply the configured timezone to this process and child processes.""" |
| 74 | os.environ["TZ"] = self.timezone |
| 75 | if hasattr(time, "tzset"): |
| 76 | try: |
| 77 | time.tzset() |
| 78 | except Exception as e: |
| 79 | PrintStyle.error(f"Error applying timezone {self.timezone}: {e}") |
| 80 | |
| 81 | def now(self) -> datetime: |
| 82 | """Return the current datetime in the user's configured timezone.""" |
| 83 | return datetime.now(self.get_tzinfo()) |
| 84 | |
| 85 | def now_iso(self, sep: str = "T", timespec: str = "auto") -> str: |
| 86 | return self.now().isoformat(sep=sep, timespec=timespec) |
| 87 | |
| 88 | def localize_naive_datetime(self, dt: datetime) -> datetime: |
| 89 | """Treat a naive datetime as user-local and make it timezone-aware.""" |
| 90 | if dt.tzinfo is not None: |
| 91 | return dt |
| 92 | tzinfo = self.get_tzinfo() |
| 93 | try: |
| 94 | return tzinfo.localize(dt, is_dst=None) |
| 95 | except pytz.exceptions.AmbiguousTimeError: |
| 96 | return tzinfo.localize(dt, is_dst=False) |
| 97 | except pytz.exceptions.NonExistentTimeError: |
| 98 | return tzinfo.localize(dt, is_dst=True) |
| 99 | |
| 100 | def set_timezone(self, timezone: str) -> None: |
| 101 | """Set the user's IANA timezone and propagate it to child processes.""" |
| 102 | try: |
| 103 | # Validate timezone and compute its current offset |
| 104 | _ = pytz.timezone(timezone) |
| 105 | new_offset = self._compute_offset_minutes(timezone) |
| 106 | if timezone == self.timezone and new_offset == self._offset_minutes: |
| 107 | self.apply_process_timezone() |
| 108 | return |
| 109 | |
| 110 | prev_tz = getattr(self, "timezone", "None") |
| 111 | prev_off = getattr(self, "_offset_minutes", None) |
| 112 | PrintStyle.debug( |
| 113 | f"Changing timezone from {prev_tz} (offset {prev_off}) to {timezone} (offset {new_offset})" |
| 114 | ) |
| 115 | self._offset_minutes = new_offset |
| 116 | self.timezone = timezone |
| 117 | save_dotenv_value("DEFAULT_USER_TIMEZONE", timezone) |
| 118 | save_dotenv_value("DEFAULT_USER_UTC_OFFSET_MINUTES", str(self._offset_minutes)) |
| 119 | self.apply_process_timezone() |
| 120 | self._last_timezone_change = datetime.now() |
| 121 | except pytz.exceptions.UnknownTimeZoneError: |
| 122 | fallback_timezone = self.timezone |
| 123 | try: |
| 124 | pytz.timezone(fallback_timezone) |
| 125 | except pytz.exceptions.UnknownTimeZoneError: |
| 126 | fallback_timezone = "UTC" |
| 127 | |
| 128 | PrintStyle.error(f"Unknown timezone: {timezone}, keeping {fallback_timezone}") |
| 129 | self.timezone = fallback_timezone |
| 130 | self._offset_minutes = self._compute_offset_minutes(fallback_timezone) |
| 131 | self.apply_process_timezone() |
| 132 | |
| 133 | def localtime_str_to_utc_dt(self, localtime_str: str | None) -> datetime | None: |
| 134 | """ |
| 135 | Convert a local time ISO string to a UTC datetime object. |
| 136 | Returns None if input is None or invalid. |
| 137 | When input lacks tzinfo, assume the configured user timezone. |
| 138 | """ |
| 139 | if not localtime_str: |
| 140 | return None |
| 141 | |
| 142 | try: |
| 143 | localtime_str = localtime_str.strip().replace("Z", "+00:00") |
| 144 | # Handle both with and without timezone info |
| 145 | try: |
| 146 | # Try parsing with timezone info first |
| 147 | local_datetime_obj = datetime.fromisoformat(localtime_str) |
| 148 | if local_datetime_obj.tzinfo is None: |
| 149 | # If no timezone info, assume the configured user timezone. |
| 150 | local_datetime_obj = self.localize_naive_datetime(local_datetime_obj) |
| 151 | except ValueError: |
| 152 | # If timezone parsing fails, try a few common local formats. |
| 153 | cleaned = localtime_str.replace("T", " ") |
| 154 | for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"): |
| 155 | try: |
| 156 | local_datetime_obj = datetime.strptime(cleaned, fmt) |
| 157 | local_datetime_obj = self.localize_naive_datetime(local_datetime_obj) |
| 158 | break |
| 159 | except ValueError: |
| 160 | continue |
| 161 | else: |
| 162 | raise |
| 163 | |
| 164 | # Convert to UTC |
| 165 | return local_datetime_obj.astimezone(dt_timezone.utc) |
| 166 | except Exception as e: |
| 167 | PrintStyle.error(f"Error converting localtime string to UTC: {e}") |
| 168 | return None |
| 169 | |
| 170 | def utc_dt_to_localtime_str(self, utc_dt: datetime | None, sep: str = "T", timespec: str = "auto") -> str | None: |
| 171 | """ |
| 172 | Convert a UTC datetime object to a local time ISO string using the user's timezone. |
| 173 | Returns None if input is None. |
| 174 | """ |
| 175 | if utc_dt is None: |
| 176 | return None |
| 177 | |
| 178 | # At this point, utc_dt is definitely not None |
| 179 | assert utc_dt is not None |
| 180 | |
| 181 | try: |
| 182 | # Ensure datetime is timezone aware in UTC |
| 183 | if utc_dt.tzinfo is None: |
| 184 | utc_dt = utc_dt.replace(tzinfo=dt_timezone.utc) |
| 185 | else: |
| 186 | utc_dt = utc_dt.astimezone(dt_timezone.utc) |
| 187 | |
| 188 | # Convert to local time using the user's timezone. |
| 189 | local_datetime_obj = utc_dt.astimezone(self.get_tzinfo()) |
| 190 | return local_datetime_obj.isoformat(sep=sep, timespec=timespec) |
| 191 | except Exception as e: |
| 192 | PrintStyle.error(f"Error converting UTC datetime to localtime string: {e}") |
| 193 | return None |
| 194 | |
| 195 | def serialize_datetime(self, dt: datetime | None) -> str | None: |
| 196 | """ |
| 197 | Serialize a datetime object to ISO format string using the user's timezone. |
| 198 | This ensures the frontend receives dates with the correct offset for display. |
| 199 | """ |
| 200 | if dt is None: |
| 201 | return None |
| 202 | |
| 203 | # At this point, dt is definitely not None |
| 204 | assert dt is not None |
| 205 | |
| 206 | try: |
| 207 | # Ensure datetime is timezone aware (if not, assume the user's timezone) |
| 208 | if dt.tzinfo is None: |
| 209 | dt = self.localize_naive_datetime(dt) |
| 210 | |
| 211 | local_dt = dt.astimezone(self.get_tzinfo()) |
| 212 | return local_dt.isoformat() |
| 213 | except Exception as e: |
| 214 | PrintStyle.error(f"Error serializing datetime: {e}") |
| 215 | return None |