"""时态快照：把「现在此地」解读成完整的时态语义对象。 消费 lunar 与 duration 纯函数，组装 ``TemporalSnapshot``。零 IO——纪念日、 相遇锚点、最后活跃时间均由调用方以值传入（``SnapshotInput``）。 """ from __future__ import annotations from datetime import UTC, date, datetime, timedelta from zoneinfo import ZoneInfo from pydantic import BaseModel from app.modules.time_sense.duration import ( AcquaintanceDuration, OfflineDuration, OfflineThresholds, classify_offline_band, compute_acquaintance_duration, ) from app.modules.time_sense.lunar import ( Festival, LunarDate, SolarTermInfo, get_day_period, get_festivals, get_shichen, get_solar_term, solar_to_lunar, ) class LocationContext(BaseModel): """位置上下文（v1 来源为档案/会话手动配置）。""" timezone: str city: str | None = None locale: str = "zh-CN" class AnniversaryRef(BaseModel): """纪念日引用（从持久层取出后传入快照计算的精简形态）。""" id: str name: str calendar: str # solar / lunar date_iso: str # 公历 ISO；农历在集成层先转为公历再传入（或这里仅支持公历命中判定） recurs_yearly: bool kind: str class MilestonePreview(BaseModel): """下一个特殊日子预览。""" anniversary_id: str name: str date_iso: str days_until: int class TemporalSnapshot(BaseModel): """时态快照——辰正的核心只读对象。""" now_utc: str local_time: str timezone: str local_hour: int shichen: str day_period: str weekday: int is_weekend: bool solar_date: str lunar: LunarDate | None season: str solar_term: SolarTermInfo | None festivals: list[Festival] acquaintance: AcquaintanceDuration offline: OfflineDuration next_milestone: MilestonePreview | None active_milestones: list[AnniversaryRef] home_tz_diff_minutes: int location: LocationContext | None class SnapshotInput(BaseModel): """compute_snapshot 的入参，全部以值传入，确保纯函数。""" now: datetime timezone: str location: LocationContext | None = None anniversaries: list[AnniversaryRef] = [] acquaintance_anchor: datetime last_active_at: datetime home_timezone: str = "Asia/Shanghai" offline_thresholds: OfflineThresholds = OfflineThresholds() def _season_for(northern: bool, month: int) -> str: """按月份返回季节，南北半球相反。""" seasons_north = ["冬", "冬", "春", "春", "春", "夏", "夏", "夏", "秋", "秋", "秋", "冬"] season = seasons_north[month - 1] if northern: return season flip = {"春": "秋", "夏": "冬", "秋": "春", "冬": "夏"} return flip[season] def _tz_offset_minutes(tz_name: str, at: datetime) -> int: """返回 tz_name 在 at 时刻相对 UTC 的偏移（分钟）。""" try: localized = at.astimezone(ZoneInfo(tz_name)) except Exception: return 0 offset = localized.utcoffset() return int(offset.total_seconds() // 60) if offset is not None else 0 def _safe_replace_year(d: date, year: int) -> date: """replace(year=...) 的安全版：2/29 在非闰年回退到 2/28。""" try: return d.replace(year=year) except ValueError: return d.replace(year=year, day=28) def _is_northern(city: str | None) -> bool: """粗判城市是否北半球（v1 简化：默认北半球；南半球已知城市查表）。""" southern = {"悉尼", "墨尔本", "奥克兰", "惠灵顿", "约翰内斯堡", "布宜诺斯艾利斯", "圣地亚哥"} return not (city and city in southern) def _next_milestone( anniversaries: list[AnniversaryRef], today: date ) -> tuple[MilestonePreview | None, list[AnniversaryRef]]: """返回（下一个纪念日预览，今日命中列表）。""" active: list[AnniversaryRef] = [] upcoming: list[tuple[int, AnniversaryRef, date]] = [] for anni in anniversaries: try: anni_date = date.fromisoformat(anni.date_iso[:10]) except ValueError: continue # 今日命中：月日相同（公历） if anni.recurs_yearly: hit_this_year = _safe_replace_year(anni_date, today.year) if hit_this_year == today: active.append(anni) next_occurrence = hit_this_year + timedelta(days=365) upcoming.append(((next_occurrence - today).days, anni, next_occurrence)) elif hit_this_year > today: upcoming.append(((hit_this_year - today).days, anni, hit_this_year)) else: next_occurrence = _safe_replace_year(anni_date, today.year + 1) upcoming.append(((next_occurrence - today).days, anni, next_occurrence)) else: if anni_date == today: active.append(anni) elif anni_date > today: upcoming.append(((anni_date - today).days, anni, anni_date)) upcoming.sort() next_preview = None if upcoming: days_until, anni, d = upcoming[0] next_preview = MilestonePreview( anniversary_id=anni.id, name=anni.name, date_iso=d.isoformat(), days_until=days_until ) return next_preview, active def compute_snapshot(input: SnapshotInput) -> TemporalSnapshot: """组装时态快照。纯函数：给定输入确定地算出全部字段。""" # 本地时间 try: local_dt = input.now.astimezone(ZoneInfo(input.timezone)) except Exception: local_dt = input.now # 时区非法时降级为 UTC local_hour = local_dt.hour today = local_dt.date() # 农历/节气/节日 try: lunar = solar_to_lunar(today) except Exception: lunar = None try: solar_term = get_solar_term(today) except Exception: solar_term = None try: festivals = get_festivals( today, locale=input.location.locale if input.location else "zh-CN" ) except Exception: festivals = [] # 时辰与时段 shichen = get_shichen(local_hour).name day_period = get_day_period(local_hour) # 认识时长与离线 acquaintance = compute_acquaintance_duration(input.acquaintance_anchor, input.now) offline = classify_offline_band(input.last_active_at, input.now, input.offline_thresholds) # 纪念日 next_milestone, active_milestones = _next_milestone(input.anniversaries, today) # 季节 northern = _is_northern(input.location.city if input.location else None) season = _season_for(northern, today.month) # 时差 home_offset = _tz_offset_minutes(input.home_timezone, input.now) user_offset = _tz_offset_minutes(input.timezone, input.now) home_tz_diff = ( user_offset - home_offset ) # 用户时区减家时区，东向为正（如家=上海、用户=东京，diff=+60） return TemporalSnapshot( now_utc=input.now.astimezone(UTC).isoformat(), local_time=local_dt.isoformat(), timezone=input.timezone, local_hour=local_hour, shichen=shichen, day_period=day_period, weekday=today.weekday(), # 0=周一 is_weekend=today.weekday() >= 5, solar_date=today.isoformat(), lunar=lunar, season=season, solar_term=solar_term, festivals=festivals, acquaintance=acquaintance, offline=offline, next_milestone=next_milestone, active_milestones=active_milestones, home_tz_diff_minutes=home_tz_diff, location=input.location, )