"""Phase 4 foundations: the six profile elements a task can build toward, and a pure work-pattern aggregator. A "foundation" is one of the stable parts of a person's profile. The core time-tracker asks "which of these does this task build toward?" and posts the mapping back; the aggregator rolls those mappings up per foundation to feed the coaching reminder engine and the goal dashboard. """ from datetime import datetime from typing import Any, Dict, List, Optional # foundation key -> human label FOUNDATIONS = { "love": "What you love", "strength": "What you're good at", "mission": "What the world needs", "vocation": "What you can be paid for", "short_term": "Near-term goals (6–12mo)", "long_term": "Long-term goals (3–5yr)", } # foundation key -> the DiscoveryProfile prose field it reflects FOUNDATION_TO_FIELD = { "love": "love_summary", "strength": "strength_summary", "mission": "mission_summary", "vocation": "vocation_summary", "short_term": "short_term_goals", "long_term": "long_term_goals", } def rollup(entries: List[Dict[str, Any]]) -> Dict[str, Any]: """Aggregate task mappings into per-foundation work patterns. Args: entries: each ``{"foundation": str, "minutes": int, "occurred_at": datetime}``. Already filtered to the desired window by the caller. Returns: ``{"total_minutes", "by_foundation": [...], "neglected": [...]}`` where ``by_foundation`` covers all six foundations (zero included), sorted by minutes descending, and ``neglected`` lists foundations with no minutes. """ agg: Dict[str, Dict[str, Any]] = { key: {"minutes": 0, "task_count": 0, "last_at": None} for key in FOUNDATIONS } for e in entries: key = e.get("foundation") if key not in agg: continue # ignore unknown foundations defensively minutes = int(e.get("minutes") or 0) agg[key]["minutes"] += minutes agg[key]["task_count"] += 1 occurred = e.get("occurred_at") if occurred is not None: prev = agg[key]["last_at"] if prev is None or occurred > prev: agg[key]["last_at"] = occurred total = sum(v["minutes"] for v in agg.values()) by_foundation = [ { "foundation": key, "label": FOUNDATIONS[key], "minutes": v["minutes"], "task_count": v["task_count"], "last_at": v["last_at"], "share": (v["minutes"] / total) if total else 0.0, } for key, v in agg.items() ] by_foundation.sort(key=lambda r: r["minutes"], reverse=True) neglected = [r["foundation"] for r in by_foundation if r["minutes"] == 0] return { "total_minutes": total, "by_foundation": by_foundation, "neglected": neglected, } def work_pattern_text(summary: Dict[str, Any], window_days: int) -> Optional[str]: """A short plain-language summary of recent work patterns for the coach to reference. Returns None when there is no logged activity.""" if not summary or summary.get("total_minutes", 0) <= 0: return None active = [r for r in summary["by_foundation"] if r["minutes"] > 0] spent = "; ".join( f"{r['label']} {r['minutes']} min ({round(r['share'] * 100)}%)" for r in active ) parts = [f"In the last {window_days} days you logged time toward: {spent}."] neglected = summary.get("neglected") or [] if neglected: names = ", ".join(FOUNDATIONS[k] for k in neglected) parts.append(f"Nothing was logged toward: {names}.") return " ".join(parts)