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), )