Nexus Agent
e3efbd4552
Cron matching, audit/alert date filters, and schedule UI align on Asia/Shanghai while JWT/TOTP/heartbeat stay UTC; follow-ups add WS alert timestamps, fixed install timezone copy, and 422 on invalid date filters. Co-authored-by: Cursor <cursoragent@cursor.com>
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""User-visible timestamps — always Asia/Shanghai (北京时间)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
_CALENDAR_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
|
|
|
|
BEIJING_TZ = ZoneInfo("Asia/Shanghai")
|
|
OPERATOR_TZ = BEIJING_TZ
|
|
|
|
|
|
def utc_now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def ensure_utc(dt: datetime) -> datetime:
|
|
if dt.tzinfo is None:
|
|
return dt.replace(tzinfo=timezone.utc)
|
|
return dt.astimezone(timezone.utc)
|
|
|
|
|
|
def to_naive_utc(dt: datetime) -> datetime:
|
|
return ensure_utc(dt).replace(tzinfo=None)
|
|
|
|
|
|
def to_beijing(dt: datetime) -> datetime:
|
|
"""Convert aware or naive-UTC datetime to Beijing wall clock."""
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt.astimezone(BEIJING_TZ)
|
|
|
|
|
|
def format_beijing(dt: datetime | None = None, *, with_label: bool = True) -> str:
|
|
"""Format datetime for operators (default: now in 北京时间)."""
|
|
when = to_beijing(dt or datetime.now(timezone.utc))
|
|
text = when.strftime("%Y-%m-%d %H:%M:%S")
|
|
return f"{text} 北京时间" if with_label else text
|
|
|
|
|
|
def format_beijing_now() -> str:
|
|
return format_beijing(with_label=True)
|
|
|
|
|
|
def operator_wall(dt: datetime) -> tuple[int, int, int, int, int]:
|
|
"""Cron wall-clock components in operator TZ: minute, hour, day, month, dow (0=Sunday)."""
|
|
bj = to_beijing(ensure_utc(dt))
|
|
return (
|
|
bj.minute,
|
|
bj.hour,
|
|
bj.day,
|
|
bj.month,
|
|
(bj.weekday() + 1) % 7,
|
|
)
|
|
|
|
|
|
def parse_operator_calendar_date(raw: str) -> str:
|
|
"""Validate YYYY-MM-DD for operator (Beijing) calendar day filters."""
|
|
text = (raw or "").strip()[:10]
|
|
if not _CALENDAR_DATE_RE.match(text):
|
|
raise ValueError(f"日期格式须为 YYYY-MM-DD，收到: {raw!r}")
|
|
try:
|
|
datetime.fromisoformat(text)
|
|
except ValueError as e:
|
|
raise ValueError(f"日期格式须为 YYYY-MM-DD，收到: {raw!r}") from e
|
|
return text
|
|
|
|
|
|
def beijing_day_range_utc(date_yyyy_mm_dd: str) -> tuple[datetime, datetime]:
|
|
"""Inclusive Beijing calendar day → naive UTC (start, end) for MySQL filters."""
|
|
base = datetime.fromisoformat(parse_operator_calendar_date(date_yyyy_mm_dd))
|
|
start_bj = base.replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=BEIJING_TZ)
|
|
end_bj = base.replace(hour=23, minute=59, second=59, microsecond=999999, tzinfo=BEIJING_TZ)
|
|
return (
|
|
start_bj.astimezone(timezone.utc).replace(tzinfo=None),
|
|
end_bj.astimezone(timezone.utc).replace(tzinfo=None),
|
|
)
|