"""认识时长、离线档位、自然语言时长表达。 全部纯函数。离线阈值以参数传入(集成层从 ``TimeConfig`` 提供),不在本模块 读取配置,保持零 IO。 """ from __future__ import annotations import math from datetime import datetime from pydantic import BaseModel, model_validator _MINUTES_PER_DAY = 1440 class OfflineThresholds(BaseModel): """离线档位阈值(分钟)。默认值见辰正设计 §7.1。""" active_minutes: float = 60 # < 1h -> active short_absence_minutes: float = 720 # 1-12h -> short_absence long_absence_minutes: float = 4320 # 12h-3d -> long_absence reunion_impact_minutes: float = 4320 # > 3d -> reunion_impact @model_validator(mode="after") def _check_monotonicity(self) -> OfflineThresholds: if not ( self.active_minutes < self.short_absence_minutes < self.long_absence_minutes <= self.reunion_impact_minutes ): raise ValueError( "OfflineThresholds 必须满足 " "active_minutes < short_absence_minutes " "< long_absence_minutes <= reunion_impact_minutes" ) return self class AcquaintanceDuration(BaseModel): """认识时长。""" total_days: int years: int months: int days: int natural_language: str class OfflineDuration(BaseModel): """离线时长与档位。""" minutes: float band: str # offline_band 枚举 reunion_intensity: float # [0,1],仅 reunion_impact 档显著 def format_duration_cn(days: int) -> str: """把天数格式化为中文自然语言。""" if days <= 0: return "今天" if days < 30: return f"{days} 天" years = days // 365 remaining_after_year = days % 365 months = remaining_after_year // 30 rest = remaining_after_year % 30 if years >= 1 and months >= 1 and rest >= 1: return f"{years} 年 {months} 个月 {rest} 天({days} 天)" if years >= 1 and months >= 1: return f"{years} 年 {months} 个月({days} 天)" if years >= 1: return f"{years} 年({days} 天)" return f"{months} 个月 {rest} 天({days} 天)" def compute_acquaintance_duration(anchor: datetime, now: datetime) -> AcquaintanceDuration: """计算从 anchor 到 now 的认识时长。 年/月/日按 365 天/年、30 天/月近似拆分(用于自然语言展示,非历法精确)。 """ delta = now - anchor total_seconds = max(delta.total_seconds(), 0) total_days = int(total_seconds // 86400) years = total_days // 365 remaining_after_year = total_days % 365 months = remaining_after_year // 30 days = remaining_after_year % 30 return AcquaintanceDuration( total_days=total_days, years=years, months=months, days=days, natural_language=format_duration_cn(total_days), ) def classify_offline_band( last_active: datetime, now: datetime, thresholds: OfflineThresholds ) -> OfflineDuration: """按离线时长判定档位与重逢强度。""" delta = now - last_active minutes = max(delta.total_seconds() / 60.0, 0.0) if minutes < thresholds.active_minutes: return OfflineDuration(minutes=minutes, band="active", reunion_intensity=0.0) if minutes < thresholds.short_absence_minutes: return OfflineDuration(minutes=minutes, band="short_absence", reunion_intensity=0.0) if minutes < thresholds.long_absence_minutes: return OfflineDuration(minutes=minutes, band="long_absence", reunion_intensity=0.0) # reunion_impact:超过阈值后强度随离线时长对数增长,钳位 [0.5, 1.0] excess = minutes - thresholds.reunion_impact_minutes # 1 天以上额外 -> 强度快速上升;用 log 压缩 intensity = max(0.5, min(1.0, 0.5 + 0.5 * math.log10(max(excess, 1) / _MINUTES_PER_DAY + 1))) return OfflineDuration( minutes=minutes, band="reunion_impact", reunion_intensity=round(intensity, 3) )