"""列表/导出用 UTC 时间窗(Gate / Binance 主站共用)。""" from datetime import datetime, timedelta, timezone PRESET_UTC_TODAY = "utc_today" PRESET_UTC_LAST24H = "utc_last24h" PRESET_UTC_LAST7D = "utc_last7d" PRESET_CUSTOM = "custom" def utc_now(): return datetime.now(timezone.utc) def utc_today_bounds(now=None): now = now or utc_now() start = now.replace(hour=0, minute=0, second=0, microsecond=0) return start, now def resolve_window(query_mapping, default_preset=PRESET_UTC_TODAY): """ 从 ?win_preset= & from_utc= & to_utc= 解析窗口。 返回 dict: preset, start_utc, end_utc, label, start_ms, end_ms """ preset = (query_mapping.get("win_preset") or default_preset or PRESET_UTC_TODAY).strip().lower() now = utc_now() if preset == PRESET_UTC_LAST24H: start = now - timedelta(hours=24) end = now label = "近24小时(UTC)" elif preset == PRESET_UTC_LAST7D: start = now - timedelta(days=7) end = now label = "近7天(UTC)" elif preset == PRESET_CUSTOM: start = _parse_utc_input(query_mapping.get("from_utc")) or utc_today_bounds(now)[0] end = _parse_utc_input(query_mapping.get("to_utc")) or now if end < start: start, end = end, start label = f"{start.strftime('%Y-%m-%d %H:%M')} ~ {end.strftime('%Y-%m-%d %H:%M')} UTC" else: start, end = utc_today_bounds(now) preset = PRESET_UTC_TODAY label = f"UTC当日 {start.strftime('%Y-%m-%d')}" return { "preset": preset, "start_utc": start, "end_utc": end, "label": label, "start_ms": int(start.timestamp() * 1000), "end_ms": int(end.timestamp() * 1000), } def _parse_utc_input(raw): s = (raw or "").strip().replace("T", " ").replace("Z", "").strip() if not s: return None for fmt, n in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d %H:%M", 16), ("%Y-%m-%d", 10)): try: dt = datetime.strptime(s[:n], fmt) return dt.replace(tzinfo=timezone.utc) except Exception: continue return None def utc_window_to_bj_sql_strings(start_utc, end_utc, app_tz): """DB 存北京时间字符串时,用于 SQLite 字符串范围比较。""" start_bj = start_utc.astimezone(app_tz).strftime("%Y-%m-%d %H:%M:%S") end_bj = end_utc.astimezone(app_tz).strftime("%Y-%m-%d %H:%M:%S") return start_bj, end_bj SESSION_KEY_LIST_WIN = "list_win_filter" def query_mapping_from_session(session_store): """从 Flask session 恢复 win_preset / from_utc / to_utc。""" if not session_store: return {} block = session_store.get(SESSION_KEY_LIST_WIN) if not isinstance(block, dict): return {} preset = (block.get("preset") or "").strip() if not preset: return {} return { "win_preset": preset, "from_utc": (block.get("from_utc") or "").strip(), "to_utc": (block.get("to_utc") or "").strip(), } def resolve_list_window(query_mapping, session_store=None, default_preset=PRESET_UTC_TODAY): """ URL 带 win_preset 时解析并写入 session;无参数时用 session 中上次「应用」的预设。 """ qm = query_mapping or {} preset_in_q = (qm.get("win_preset") or "").strip() if preset_in_q: win = resolve_window(qm, default_preset=default_preset) if session_store is not None: session_store[SESSION_KEY_LIST_WIN] = { "preset": win["preset"], "from_utc": (qm.get("from_utc") or "").strip(), "to_utc": (qm.get("to_utc") or "").strip(), } return win stored = query_mapping_from_session(session_store) if stored.get("win_preset"): return resolve_window(stored, default_preset=default_preset) return resolve_window(qm, default_preset=default_preset) def list_window_redirect_query(session_store): """复盘/表单 POST 后重定向时附带列表筛选 query。""" from urllib.parse import urlencode stored = query_mapping_from_session(session_store) if not stored.get("win_preset"): return "" params = {k: v for k, v in stored.items() if v} return urlencode(params)