crypto_monitor/lib/common/history_window_lib.py

"""列表/导出用 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_UTC_THIS_MONTH = "utc_this_month"
PRESET_UTC_LAST3M = "utc_last3m"
PRESET_UTC_LAST6M = "utc_last6m"
PRESET_ALL = "all"
PRESET_CUSTOM = "custom"
PRESET_DEFAULT = PRESET_UTC_THIS_MONTH


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_DEFAULT):
    """
    从 ?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_DEFAULT).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_UTC_THIS_MONTH:
        start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
        end = now
        label = f"本月 {start.strftime('%Y-%m')}"
    elif preset == PRESET_UTC_LAST3M:
        start = now - timedelta(days=90)
        end = now
        label = "近3月"
    elif preset == PRESET_UTC_LAST6M:
        start = now - timedelta(days=180)
        end = now
        label = "近6月"
    elif preset == PRESET_ALL:
        start = datetime(2000, 1, 1, tzinfo=timezone.utc)
        end = now
        label = "全部"
    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"
    elif preset == PRESET_UTC_TODAY:
        start, end = utc_today_bounds(now)
        label = f"UTC当日 {start.strftime('%Y-%m-%d')}"
    else:
        return resolve_window(
            {**(query_mapping or {}), "win_preset": default_preset},
            default_preset=default_preset,
        )

    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


def utc_window_to_utc_sql_strings(start_utc, end_utc):
    """SQLite CURRENT_TIMESTAMP 写入 UTC 时,用于 created_at 范围比较."""
    return (
        start_utc.strftime("%Y-%m-%d %H:%M:%S"),
        end_utc.strftime("%Y-%m-%d %H:%M:%S"),
    )


def normalize_bj_datetime_storage(raw):
    """表单 datetime-local(含 T)入库前统一为 YYYY-MM-DD HH:MM:SS(北京时间)."""
    s = (raw or "").strip().replace("T", " ").replace("Z", "").strip()
    if not s:
        return ""
    for fmt, n in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d %H:%M", 16), ("%Y-%m-%d", 10)):
        try:
            return datetime.strptime(s[:n], fmt).strftime("%Y-%m-%d %H:%M:%S")
        except ValueError:
            continue
    return s


def sql_list_time_field(*columns):
    """
    SQLite 列表时间窗比较表达式.
    journal_entries 的 open/close 可能含 'T',直接与 bounds(空格格式)比会误判为超出上界.
    单列时不用 COALESCE(SQLite 要求 COALESCE 至少 2 个参数).
    """
    cols = [c for c in columns if c]
    if not cols:
        raise ValueError("sql_list_time_field requires at least one column")
    if len(cols) == 1:
        return f"REPLACE({cols[0]}, 'T', ' ')"
    return f"REPLACE(COALESCE({', '.join(cols)}), 'T', ' ')"


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_DEFAULT):
    """
    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)