shanxia/bx/bx-backend/app/utils/date_range.py

"""按自然日（YYYY-MM-DD）解析为 [day_start, day_end) 的 create_time 筛选条件。"""

from datetime import datetime, timedelta


def current_month_bounds(now: datetime | None = None) -> tuple[datetime, datetime]:
    """当前自然月 [month_start, next_month_start)，与服务器本地时间一致（项目约定 Asia/Shanghai）。"""
    if now is None:
        now = datetime.now()
    month_start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
    if now.month == 12:
        month_end = datetime(now.year + 1, 1, 1)
    else:
        month_end = datetime(now.year, now.month + 1, 1)
    return month_start, month_end


def parse_create_time_day_bounds(apply_date: str | None) -> tuple[datetime, datetime] | None:
    if not apply_date or not str(apply_date).strip():
        return None
    try:
        day_start = datetime.strptime(str(apply_date).strip(), "%Y-%m-%d")
        return day_start, day_start + timedelta(days=1)
    except ValueError:
        return None


def _parse_day_start(value: str | None) -> datetime | None:
    if not value or not str(value).strip():
        return None
    try:
        return datetime.strptime(str(value).strip(), "%Y-%m-%d")
    except ValueError:
        return None


def parse_create_time_range_bounds(
    apply_date: str | None = None,
    apply_date_start: str | None = None,
    apply_date_end: str | None = None,
) -> tuple[datetime | None, datetime | None]:
    """解析申请时间筛选为 [inclusive_start, exclusive_end)，任一端可为 None 表示不限制。"""
    if apply_date_start or apply_date_end:
        start = _parse_day_start(apply_date_start)
        end_day = _parse_day_start(apply_date_end)
        end = end_day + timedelta(days=1) if end_day else None
        if start is None and end is None:
            return None, None
        return start, end

    single = parse_create_time_day_bounds(apply_date)
    if single:
        return single[0], single[1]
    return None, None


def build_create_time_filters(
    column,
    apply_date: str | None = None,
    apply_date_start: str | None = None,
    apply_date_end: str | None = None,
) -> list:
    """生成 create_time 区间筛选条件列表。"""
    start, end = parse_create_time_range_bounds(apply_date, apply_date_start, apply_date_end)
    filters: list = []
    if start is not None:
        filters.append(column >= start)
    if end is not None:
        filters.append(column < end)
    return filters


def resolve_apply_create_time(apply_date: str | None, now: datetime | None = None) -> datetime:
    """申请日 YYYY-MM-DD → create_time；当天用当前时刻，其它日期用当日 0 点。"""
    if now is None:
        now = datetime.now()
    bounds = parse_create_time_day_bounds(apply_date)
    if not bounds:
        return now
    day_start, _ = bounds
    if day_start.date() == now.date():
        return now
    return day_start