demand_server/app/strategies/odps/_date_utils.py

import re
from datetime import date, datetime, timedelta
from zoneinfo import ZoneInfo

SHANGHAI_TZ = ZoneInfo("Asia/Shanghai")
_DATE_PARTITION_RE = re.compile(r"^\d{8}$")


def today_shanghai_date() -> date:
    return datetime.now(SHANGHAI_TZ).date()


def same_day_last_year(value: date) -> date:
    try:
        return value.replace(year=value.year - 1)
    except ValueError:
        return date(value.year - 1, 2, 28)


def partition_yyyymmdd(value: date) -> str:
    return value.strftime("%Y%m%d")


def parse_partition_date(value: str) -> date:
    text = value.strip()
    if not _DATE_PARTITION_RE.match(text):
        raise ValueError(f"日期须为 YYYYMMDD 格式，当前为 {value!r}")
    return datetime.strptime(text, "%Y%m%d").date()


def resolve_solar_same_period_window(
    *,
    bizdate: str,
    period_days: int = 7,
) -> tuple[str, str, str]:
    """
    返回 (bizdate, start_date, end_date)。

    - bizdate：当天 YYYYMMDD
    - start_date：bizdate 去年同期（闰年 2/29 → 去年 2/28）
    - end_date：start_date + (period_days - 1) 天；period_days=7 → +6 天
    """
    if period_days < 1:
        raise ValueError("period_days 须 >= 1")

    biz_dt = parse_partition_date(bizdate)
    start_dt = same_day_last_year(biz_dt)
    end_dt = start_dt + timedelta(days=period_days - 1)
    return (
        partition_yyyymmdd(biz_dt),
        partition_yyyymmdd(start_dt),
        partition_yyyymmdd(end_dt),
    )