"""Deterministic customer-scoped period reporting.""" from __future__ import annotations from dataclasses import dataclass from datetime import date, timedelta from pathlib import Path from collections import Counter from statistics import fmean from typing import Any from collections.abc import Mapping from checkin_cli.customer_coaching import CustomerProfile, PlanWeek, TwelveWeekPlan from checkin_cli.models import ContractStatus, Event, EventType from checkin_cli.weekly_operations_domain_reporting import ( CustomerWeeklyReviewSource as CustomerWeeklyReviewSource, WeeklyActionOutcome, WeeklyReviewRequest, WeeklySummary as WeeklySummary, build_customer_weekly_review_source as _build_customer_weekly_review_source, ) @dataclass(frozen=True, slots=True) class CustomerPeriodReport: starts_on: date ends_on: date sample_count: int average_weight_kg: float | None average_calories_kcal: int | None average_protein_g: int | None average_carbohydrate_g: int | None average_fat_g: int | None average_water_liters: float | None average_sleep_hours: float | None average_sleep_quality_1to5: float | None average_readiness_1to5: float | None weight_change_kg: float | None weight_trend: str calorie_target_adherence_percent: int | None protein_target_adherence_percent: int | None completion_percent: int missing_day_count: int missing_weight_count: int missing_calories_count: int latest_meal_summary: str | None latest_digestion_summary: str | None latest_appetite_stress_summary: str | None repeated_digestion_signals: tuple[str, ...] well_done: tuple[str, ...] improvement_priorities: tuple[str, ...] next_actions: tuple[str, ...] def build_customer_weekly_review_source( summary: WeeklySummary, *, customer_key: str, latest_action: WeeklyActionOutcome | None, next_review_date: date, ) -> CustomerWeeklyReviewSource: """Compatibility boundary for the established weekly reporting API.""" return _build_customer_weekly_review_source( WeeklyReviewRequest(summary, customer_key, latest_action, next_review_date) ) @dataclass(frozen=True, slots=True) class PilotKPIJudgement: """One deterministic conjunction projected from one customer's events.""" starts_on: date ends_on: date eligible_weekdays: tuple[date, ...] accepted_checkin_dates: tuple[date, ...] checkin_rate_percent: float checkin_pass: bool satisfaction_score: float | None satisfaction_pass: bool weekly_operator_minutes: tuple[int, ...] operator_time_pass: bool renewal_valid: bool renewal_event_id: str | None passed: bool failure_reasons: tuple[str, ...] @property def all_pass(self) -> bool: return self.passed @property def kpi_pass(self) -> bool: return self.passed @property def checkin_rate(self) -> float: return self.checkin_rate_percent @property def satisfaction(self) -> float | None: return self.satisfaction_score @property def satisfaction_passed(self) -> bool: return self.satisfaction_pass @property def operator_minutes_by_week(self) -> tuple[int, ...]: return self.weekly_operator_minutes @property def renewal_pass(self) -> bool: return self.renewal_valid # Names used by the pilot callers; they all refer to this one immutable result. CustomerWeeklySummary = WeeklySummary CustomerWeeklyReview = CustomerWeeklyReviewSource KPIJudgement = PilotKPIJudgement CustomerKPIReport = PilotKPIJudgement PilotKPIReport = PilotKPIJudgement CustomerKPIJudgement = PilotKPIJudgement def build_customer_period_report( events_path: Path, starts_on: date, ends_on: date, *, targets: PlanWeek | None = None, plan: TwelveWeekPlan | None = None, profile: CustomerProfile | None = None, ) -> CustomerPeriodReport: """Project one customer's immutable events over an inclusive date range.""" events = _read_events(events_path) superseded = {event.supersedes for event in events if event.supersedes is not None} eligible = tuple(sorted(( event for event in events if event.status is ContractStatus.ACCEPTED and event.event_type in (EventType.MORNING_CHECKIN, EventType.NUTRITION_CHECKIN, EventType.CORRECTION) and event.event_id not in superseded and event.check_in is not None and starts_on <= date.fromisoformat(event.occurred_at_kst[:10]) <= ends_on ), key=lambda event: event.occurred_at_kst)) weights = _values(eligible, "body_weight_kg") calories = _values(eligible, "calories_kcal") sleeps = _values(eligible, "sleep_hours") sleep_qualities = _values(eligible, "sleep_quality_1to5") readiness = _values(eligible, "readiness_1to5") proteins = _values(eligible, "protein_g") carbohydrates = _values(eligible, "carbohydrate_g") fats = _values(eligible, "fat_g") waters = _values(eligible, "water_liters") total_days = (ends_on - starts_on).days + 1 calorie_adherence = _period_adherence( eligible, "calories_kcal", targets, plan, symmetric=True, ) protein_adherence = _period_adherence( eligible, "protein_g", targets, plan, symmetric=False, ) digestion_values = _texts(eligible, "digestion_summary") repeated_digestion = tuple( value for value, count in Counter(digestion_values).items() if count >= 2 and value.casefold() not in {"normal", "none", "정상"} ) average_sleep = _average(sleeps, 2) average_ready = _average(readiness, 2) well_done = _well_done(len(eligible), total_days, calorie_adherence, protein_adherence, average_sleep, profile) improvements = _improvements( len(eligible), total_days, calorie_adherence, protein_adherence, average_sleep, average_ready, repeated_digestion, profile, ) weight_change = round(float(weights[-1]) - float(weights[0]), 2) if len(weights) >= 2 else None return CustomerPeriodReport( starts_on=starts_on, ends_on=ends_on, sample_count=len(eligible), average_weight_kg=_average(weights), average_calories_kcal=_average_int(calories), average_protein_g=_average_int(proteins), average_carbohydrate_g=_average_int(carbohydrates), average_fat_g=_average_int(fats), average_water_liters=_average(waters), average_sleep_hours=average_sleep, average_sleep_quality_1to5=_average(sleep_qualities, 2), average_readiness_1to5=average_ready, weight_change_kg=weight_change, weight_trend=_weight_trend(weight_change), calorie_target_adherence_percent=calorie_adherence, protein_target_adherence_percent=protein_adherence, completion_percent=round(len(eligible) / total_days * 100), missing_day_count=max(0, total_days - len(eligible)), missing_weight_count=len(eligible) - len(weights), missing_calories_count=len(eligible) - len(calories), latest_meal_summary=_latest_text(eligible, "meal_summary"), latest_digestion_summary=_latest_text(eligible, "digestion_summary"), latest_appetite_stress_summary=_latest_text(eligible, "appetite_stress_summary"), repeated_digestion_signals=repeated_digestion, well_done=well_done, improvement_priorities=improvements, next_actions=_next_actions(improvements), ) def _values(events: tuple[Event, ...], field: str) -> tuple[float | int, ...]: values = (getattr(event.check_in, field, None) for event in events if event.check_in is not None) return tuple(value for value in values if isinstance(value, (float, int))) def _texts(events: tuple[Event, ...], field: str) -> tuple[str, ...]: values = (getattr(event.check_in, field, None) for event in events if event.check_in is not None) return tuple(value for value in values if isinstance(value, str) and value.strip()) def _latest_text(events: tuple[Event, ...], field: str) -> str | None: values = _texts(events, field) return values[-1] if values else None def _average(values: tuple[float | int, ...], digits: int = 2) -> float | None: if not values: return None return round(fmean(values), digits) def _average_int(values: tuple[float | int, ...]) -> int | None: return round(fmean(values)) if values else None def _period_adherence( events: tuple[Event, ...], field: str, targets: PlanWeek | None, plan: TwelveWeekPlan | None, *, symmetric: bool, ) -> int | None: comparisons: list[bool] = [] for event in events: value = getattr(event.check_in, field, None) if event.check_in is not None else None day = date.fromisoformat(event.occurred_at_kst[:10]) elapsed = max(0, (day - plan.starts_on).days) if plan is not None else 0 target_week = plan.weeks[min(11, elapsed // 7)] if plan is not None else targets target = getattr(target_week, field, None) if target_week is not None else None if not isinstance(value, (float, int)) or not isinstance(target, int): continue comparisons.append(abs(value - target) <= target * 0.1 if symmetric else value >= target) if not comparisons: return None return round(sum(comparisons) / len(comparisons) * 100) def _weight_trend(change: float | None) -> str: if change is None: return "insufficient_data" if abs(change) < 0.3: return "stable" return "increasing" if change > 0 else "decreasing" def _well_done( samples: int, days: int, calories: int | None, protein: int | None, sleep: float | int | None, profile: CustomerProfile | None, ) -> tuple[str, ...]: items: list[str] = [] if samples / days >= 0.8: items.append("체크인 지속성 80% 이상") if calories is not None and calories >= 80: items.append("칼로리 목표 범위 80% 이상") if protein is not None and protein >= 80: items.append("단백질 목표 달성 80% 이상") if profile and profile.sleep_goal_hours is not None and sleep is not None and sleep >= profile.sleep_goal_hours: items.append("수면 목표 달성") return tuple(items) def _improvements( samples: int, days: int, calories: int | None, protein: int | None, sleep: float | int | None, readiness: float | int | None, digestion: tuple[str, ...], profile: CustomerProfile | None, ) -> tuple[str, ...]: items: list[str] = [] if samples / days < 0.8: items.append("체크인 누락 줄이기") if calories is not None and calories < 80: items.append("칼로리 목표 범위 일관성") if protein is not None and protein < 80: items.append("단백질 목표 달성률") if profile and profile.sleep_goal_hours is not None and sleep is not None and sleep < profile.sleep_goal_hours: items.append("수면 목표 시간 확보") if readiness is not None and readiness < 3: items.append("낮은 컨디션과 회복 원인 확인") if digestion: items.append("반복 소화 불편과 식사 구성 대조") return tuple(items) def _next_actions(improvements: tuple[str, ...]) -> tuple[str, ...]: actions = { "체크인 누락 줄이기": "다음 주에는 체크인 시간을 한 시각으로 고정합니다.", "칼로리 목표 범위 일관성": "외식 전후 식사를 계획해 목표 칼로리 ±10% 범위를 확인합니다.", "단백질 목표 달성률": "각 식사에 단백질 공급원을 먼저 배치합니다.", "수면 목표 시간 확보": "취침 시작 시각을 30분 앞당기고 다음 체크인에서 수면 시간을 확인합니다.", "낮은 컨디션과 회복 원인 확인": "훈련량·수면·식사 누락을 함께 기록해 회복 저하 원인을 좁힙니다.", "반복 소화 불편과 식사 구성 대조": "불편이 있었던 식사와 양을 기록해 반복 패턴을 확인합니다.", } return tuple(actions[item] for item in improvements) def build_weekly_summary( events_path: Path, starts_on: date, ends_on: date | None = None, *, targets: PlanWeek | None = None, plan: TwelveWeekPlan | None = None, profile: CustomerProfile | None = None, ) -> WeeklySummary: """Build the bounded owner-facing summary for one inclusive KST week.""" if ends_on is None: ends_on = starts_on + timedelta(days=6) if ends_on < starts_on: raise ValueError("weekly summary end must not precede start") events = _active_events(_read_events(events_path)) eligible_weekdays = _weekday_dates(starts_on, ends_on) checkin_dates = tuple( day for day in _accepted_checkin_dates(events, starts_on, ends_on) if day in set(eligible_weekdays) ) report = build_customer_period_report( events_path, starts_on, ends_on, targets=targets, plan=plan, profile=profile, ) rate = _rate_percent(len(checkin_dates), len(eligible_weekdays)) keep = _unique_strings( (*report.well_done, *(("평일 체크인 80% 이상",) if rate >= 80.0 else ())), ) change = _unique_strings(report.improvement_priorities) if not checkin_dates and "체크인 누락 줄이기" not in change: change = (*change, "체크인 누락 줄이기") trends = _summary_trends(report) next_decision = _summary_next_decision(change, report.next_actions) return WeeklySummary( starts_on=starts_on, ends_on=ends_on, eligible_weekdays=eligible_weekdays, checkin_dates=checkin_dates, checkin_rate_percent=rate, trends=trends, keep_behaviors=keep, change_behaviors=change, next_decision=next_decision, average_weight_kg=report.average_weight_kg, weight_change_kg=report.weight_change_kg, weight_trend=report.weight_trend, ) def build_customer_weekly_summary( events_path: Path, starts_on: date, ends_on: date | None = None, *, targets: PlanWeek | None = None, plan: TwelveWeekPlan | None = None, profile: CustomerProfile | None = None, ) -> WeeklySummary: """Named customer-scoped alias for the weekly summary projection.""" return build_weekly_summary( events_path, starts_on, ends_on, targets=targets, plan=plan, profile=profile, ) def judge_pilot_kpis( events_path: Path, starts_on: date | TwelveWeekPlan | Any | None = None, ends_on: date | None = None, *, plan: TwelveWeekPlan | None = None, ) -> PilotKPIJudgement: """Judge the four pilot KPIs as one exact conjunction. The check-in denominator is the weekday set in the fixed 28-day window. Operator minutes are projected into four individual seven-day buckets; superseded entries are excluded before totals are calculated. """ starts_on, ends_on = _pilot_window(starts_on, ends_on, plan) events = _active_events(_read_events(events_path)) eligible_weekdays = _weekday_dates(starts_on, ends_on) accepted_dates = tuple( day for day in _accepted_checkin_dates(events, starts_on, ends_on) if day in set(eligible_weekdays) ) checkin_rate = _rate_percent(len(accepted_dates), len(eligible_weekdays)) checkin_pass = bool(eligible_weekdays) and len(accepted_dates) * 100 >= len(eligible_weekdays) * 80 satisfaction_scores = tuple( score for event in events if _event_type_value(event) == "satisfaction_record" and _event_status_value(event) == "accepted" and (satisfaction_day := _event_date(event)) is not None and starts_on <= satisfaction_day <= ends_on and (score := _number(_payload_value(event, "score_1to10", "score"))) is not None and 1 <= score <= 10 ) satisfaction_score = round(fmean(satisfaction_scores), 2) if satisfaction_scores else None satisfaction_pass = satisfaction_score is not None and satisfaction_score >= 8.0 weekly_buckets = _operator_time_buckets(events, starts_on) weekly_minutes = tuple(minutes for minutes, _ in weekly_buckets) operator_time_pass = all( minutes <= 60 for minutes, _ in weekly_buckets ) renewal_event = _effective_exact_renewal(events, starts_on) renewal_valid = renewal_event is not None failures: list[str] = [] if not checkin_pass: failures.append("checkin_rate") if not satisfaction_pass: failures.append("satisfaction") if not operator_time_pass: failures.extend( f"operator_time_week_{index + 1}" for index, (minutes, _) in enumerate(weekly_buckets) if minutes > 60 ) if not renewal_valid: failures.append("renewal") return PilotKPIJudgement( starts_on=starts_on, ends_on=ends_on, eligible_weekdays=eligible_weekdays, accepted_checkin_dates=accepted_dates, checkin_rate_percent=checkin_rate, checkin_pass=checkin_pass, satisfaction_score=satisfaction_score, satisfaction_pass=satisfaction_pass, weekly_operator_minutes=weekly_minutes, operator_time_pass=operator_time_pass, renewal_valid=renewal_valid, renewal_event_id=getattr(renewal_event, "event_id", None), passed=not failures, failure_reasons=tuple(failures), ) def build_pilot_kpi_judgement( events_path: Path, starts_on: date | TwelveWeekPlan | Any | None = None, ends_on: date | None = None, *, plan: TwelveWeekPlan | None = None, ) -> PilotKPIJudgement: """Explicit alias for the single-source pilot KPI projection.""" return judge_pilot_kpis(events_path, starts_on, ends_on, plan=plan) def build_customer_kpi_report( events_path: Path, starts_on: date | TwelveWeekPlan | Any | None = None, ends_on: date | None = None, *, plan: TwelveWeekPlan | None = None, ) -> PilotKPIJudgement: """Customer-scoped alias retained for gateway/reporting callers.""" return judge_pilot_kpis(events_path, starts_on, ends_on, plan=plan) def evaluate_pilot_kpis( events_path: Path, starts_on: date | TwelveWeekPlan | Any | None = None, ends_on: date | None = None, *, plan: TwelveWeekPlan | None = None, ) -> PilotKPIJudgement: return judge_pilot_kpis(events_path, starts_on, ends_on, plan=plan) def _pilot_window( starts_on: date | TwelveWeekPlan | Any | None, ends_on: date | None, plan: TwelveWeekPlan | None, ) -> tuple[date, date]: if plan is None and starts_on is not None and not isinstance(starts_on, date): candidate = getattr(starts_on, "spec", starts_on) candidate_plan = getattr(candidate, "plan", candidate) if hasattr(candidate_plan, "starts_on") and hasattr(candidate_plan, "weeks"): plan = candidate_plan if plan is not None: canonical_start = _as_date(getattr(plan, "starts_on", None)) if canonical_start is None: raise ValueError("pilot plan must provide starts_on") canonical_end = canonical_start + timedelta(days=27) if ends_on is not None and ends_on != canonical_end: raise ValueError("pilot KPI window must be exactly 28 days") # The plan's first 28 days are authoritative; callers cannot extend them. return canonical_start, canonical_end if not isinstance(starts_on, date): raise ValueError("pilot KPI judgement requires a start date or plan") canonical_end = starts_on + timedelta(days=27) if ends_on is not None and ends_on != canonical_end: raise ValueError("pilot KPI window must be exactly 28 days") return starts_on, canonical_end def _weekday_dates(starts_on: date, ends_on: date) -> tuple[date, ...]: return tuple( starts_on + timedelta(days=offset) for offset in range((ends_on - starts_on).days + 1) if (starts_on + timedelta(days=offset)).weekday() < 5 ) def _rate_percent(numerator: int, denominator: int) -> float: if denominator <= 0: return 0.0 return round(numerator / denominator * 100, 2) def _event_type_value(event: Event) -> str: value = getattr(event, "event_type", "") return str(getattr(value, "value", value)) def _event_status_value(event: Event) -> str: value = getattr(event, "status", "") return str(getattr(value, "value", value)) def _event_payload(event: Event) -> Any: for name in ("payload", "event_payload", "typed_payload"): payload = getattr(event, name, None) if payload is not None: return payload event_type = _event_type_value(event) for name in ( event_type, event_type.removesuffix("_record"), f"{event_type}_payload", ): payload = getattr(event, name, None) if payload is not None: return payload return None def _payload_value(event: Event, *names: str) -> Any: payload = _event_payload(event) for name in names: if payload is not None: value = getattr(payload, name, None) if value is not None: return value if isinstance(payload, Mapping): value = payload.get(name) if value is not None: return value value = getattr(event, name, None) if value is not None: return value return None def _as_date(value: Any) -> date | None: if isinstance(value, date): return value if isinstance(value, str): try: return date.fromisoformat(value[:10]) except ValueError: return None return None def _event_date(event: Event) -> date | None: occurred = _as_date(getattr(event, "occurred_at_kst", None)) if occurred is not None: return occurred return _as_date(_payload_value(event, "kst_date", "checkin_date", "collection_date", "work_date")) def _number(value: Any) -> float | None: if isinstance(value, bool) or not isinstance(value, (int, float)): return None return float(value) def _unique_event_stream(events: tuple[Event, ...]) -> tuple[Event, ...]: """Collapse duplicate append attempts without collapsing distinct entries.""" seen_keys: set[str] = set() selected: list[Event] = [] for event in events: key = getattr(event, "dedupe_key", None) if isinstance(key, str) and key: if key in seen_keys: continue seen_keys.add(key) selected.append(event) return tuple(selected) def _active_events(events: tuple[Event, ...]) -> tuple[Event, ...]: events = _unique_event_stream(events) superseded_refs: set[str] = { str(value) for event in events if (value := getattr(event, "supersedes", None)) is not None } for event in events: value = _payload_value(event, "supersedes_entry_id", "supersedes_event_id") if value is not None: superseded_refs.add(str(value)) return tuple( event for event in events if str(getattr(event, "event_id", "")) not in superseded_refs and str(_payload_value(event, "entry_id")) not in superseded_refs ) def _accepted_checkin_dates( events: tuple[Event, ...], starts_on: date, ends_on: date, ) -> tuple[date, ...]: accepted_types = { "check_in_validated", "correction", "morning_checkin", "nutrition_checkin", "customer_checkin", } dates = { day for event in events if _event_status_value(event) == "accepted" and _event_type_value(event) in accepted_types and (day := _event_date(event)) is not None and starts_on <= day <= ends_on } return tuple(sorted(dates)) def _operator_time_buckets( events: tuple[Event, ...], starts_on: date, ) -> tuple[tuple[int, int], ...]: buckets = [[0, 0] for _ in range(4)] seen_entries: set[str] = set() end = starts_on + timedelta(days=27) for event in events: if _event_status_value(event) != "accepted" or _event_type_value(event) != "operator_time_record": continue entry_id = _payload_value(event, "entry_id") key = str(entry_id) if entry_id is not None else str(getattr(event, "event_id", "")) if key in seen_entries: continue seen_entries.add(key) work_day = _as_date(_payload_value(event, "work_date", "kst_date")) minutes = _number(_payload_value(event, "minutes")) if work_day is None or minutes is None or not starts_on <= work_day <= end: continue bucket = (work_day - starts_on).days // 7 if 0 <= bucket < 4: buckets[bucket][0] += int(minutes) buckets[bucket][1] += 1 return tuple((minutes, entries) for minutes, entries in buckets) def _operator_minutes_by_week(events: tuple[Event, ...], starts_on: date) -> tuple[int, ...]: return tuple(minutes for minutes, _ in _operator_time_buckets(events, starts_on)) def _effective_exact_renewal(events: tuple[Event, ...], starts_on: date) -> Event | None: expected_start = starts_on + timedelta(days=28) expected_end = starts_on + timedelta(days=55) renewals = tuple( event for event in events if _event_status_value(event) == "accepted" and _event_type_value(event) == "payment_record" and _payload_value(event, "kind") == "renewal" ) if len(renewals) != 1: return None event = renewals[0] amount = _number(_payload_value(event, "amount_krw")) method = _payload_value(event, "method") period_start = _as_date(_payload_value(event, "period_start_on", "period_start")) period_end = _as_date(_payload_value(event, "period_end_on", "period_end")) paid_on = _as_date(_payload_value(event, "paid_on", "paid_date", "paid_at")) if ( amount != 150000 or method != "bank_transfer" or period_start != expected_start or period_end != expected_end or paid_on is None or paid_on > expected_end ): return None return event def _summary_trends(report: CustomerPeriodReport) -> tuple[str, ...]: trends: list[str] = [] if report.weight_change_kg is not None: trends.append(f"체중 {report.weight_trend} ({report.weight_change_kg:+.2f}kg)") if report.average_sleep_hours is not None: trends.append(f"평균 수면 {report.average_sleep_hours:.2f}시간") if report.average_readiness_1to5 is not None: trends.append(f"평균 컨디션 {report.average_readiness_1to5:.2f}/5") return tuple(trends) or ("관찰 가능한 추세 없음",) def _summary_next_decision(change: tuple[str, ...], actions: tuple[str, ...]) -> str: if change and actions: return f"조정: {actions[0]}" if change: return f"조정: {change[0]}" return "유지: 현재 행동을 유지하고 다음 주 추세를 확인합니다." def _unique_strings(values: tuple[str, ...]) -> tuple[str, ...]: return tuple(dict.fromkeys(value for value in values if isinstance(value, str) and value.strip())) def _read_events(path: Path) -> tuple[Event, ...]: if not path.exists(): return () return tuple( Event.model_validate_json(line) for line in path.read_text(encoding="utf-8").splitlines() if line ) build_weekly_customer_summary = build_weekly_summary judge_customer_kpis = judge_pilot_kpis evaluate_customer_kpis = judge_pilot_kpis build_customer_kpi_judgement = judge_pilot_kpis build_pilot_kpi_report = judge_pilot_kpis