修复一些bug
This commit is contained in:
dekun
+127
-75
@@ -1,75 +1,127 @@
|
||||
"""列表/导出用 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
|
||||
"""列表/导出用 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)
|
||||
|
||||
Reference in New Issue
Block a user