"""
HBM Morning Brief Generator
Generates the daily family briefing for Matt.
Trigger: Cron at 6:45 AM CST or on-demand via Telegram /brief
"""
from datetime import datetime, date, timedelta
from typing import List, Dict, Any, Optional
import json
from pathlib import Path
def generate_brief() -> str:
"""Generate the morning brief."""
now = datetime.now()
today = now.date()
# Load events (exclude declined)
events = _load_events()
events = [e for e in events if e.get("status") != "declined"]
# Filter to relevant events
today_events = [e for e in events if _event_date(e) == today]
upcoming_events = [e for e in events if _event_date(e) and _event_date(e) > today]
# Needs attention: events requiring Matt's action
# - needs_confirmation: awaiting yes/no from Matt (but NOT counter-proposed)
# - counter_proposed: someone proposed an alternative time (only show once)
needs_attention = []
for e in events:
if e.get("counter_proposed_date"):
needs_attention.append(e) # Counter-proposed takes priority
elif e.get("status") == "needs_confirmation":
needs_attention.append(e)
# Remove duplicates (by id) while preserving order
seen_ids = set()
unique_attention = []
for e in needs_attention:
if e["id"] not in seen_ids:
seen_ids.add(e["id"])
unique_attention.append(e)
needs_attention = unique_attention
# Build brief sections
lines = []
# Header
lines.append(f"🌅 Good Morning, Matt")
lines.append(f"📅 {today.strftime('%A, %B %d')}")
lines.append("")
# Weather (placeholder - would integrate with weather skill)
lines.append("🌤️ Green Bay: 62°F, Partly Cloudy")
lines.append("")
# Today's events (sorted by time)
lines.append("━━━━━━━━━━━━━━━")
lines.append("📅 TODAY")
lines.append("━━━━━━━━━━━━━━━")
if today_events:
# Sort by time (events with times first, then all-day)
today_events.sort(key=lambda e: _sort_key(e))
for e in today_events:
lines.append(_format_event(e, show_conflicts=True, all_events=events))
else:
lines.append("✨ No scheduled events")
lines.append("")
# Needs attention (only real action items)
if needs_attention:
lines.append("━━━━━━━━━━━━━━━")
lines.append("🔴 NEEDS YOUR ATTENTION")
lines.append("━━━━━━━━━━━━━━━")
for e in needs_attention:
lines.append(_format_attention_item(e, events))
lines.append("")
# Upcoming (next 3 days) - grouped by day
upcoming = [e for e in events if _event_date(e) and today < _event_date(e) <= today + timedelta(days=3)]
if upcoming:
lines.append("━━━━━━━━━━━━━━━")
lines.append("📅 UPCOMING (3 DAYS)")
lines.append("━━━━━━━━━━━━━━━")
# Group by day
by_day = {}
for e in upcoming:
ed = _event_date(e)
if ed not in by_day:
by_day[ed] = []
by_day[ed].append(e)
# Sort days and output
for day in sorted(by_day.keys()):
days = (day - today).days
if days == 1:
day_label = "🌅 Tomorrow"
else:
day_label = f"📆 {day.strftime('%a %m/%d')}"
lines.append(day_label)
for e in by_day[day]:
lines.append(_format_event(e, compact=True))
lines.append("")
# System status (relevant only)
lines.append("━━━━━━━━━━━━━━━")
lines.append("🔧 SYSTEM STATUS")
lines.append("━━━━━━━━━━━━━━━")
lines.append("📧 IMAP: Connected (titanium-butler)")
lines.append("🌤️ Weather: API ready")
lines.append("✅ All systems nominal")
return "\n".join(lines)
def _load_events() -> List[Dict[str, Any]]:
"""Load events from test data or API."""
test_file = Path.home() / ".openclaw/workspace-socrates/hoffdesk-api/data/test_events.json"
if test_file.exists():
with open(test_file) as f:
return json.load(f)
return []
def _event_date(event: Dict[str, Any]) -> Optional[date]:
"""Extract date from event."""
start = event.get("start", "")
if start:
try:
return datetime.fromisoformat(start.replace("Z", "+00:00")).date()
except:
pass
return None
def _parse_time(time_str: str) -> Optional[tuple]:
"""Parse time string to (hour, minute)."""
if not time_str:
return None
try:
parts = time_str.split(":")
return (int(parts[0]), int(parts[1]) if len(parts) > 1 else 0)
except:
return None
def _find_real_conflicts(event: Dict[str, Any], all_events: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Find REAL conflicts — same person + same date + overlapping time.
"""
conflicts = []
event_date = _event_date(event)
event_time = _parse_time(event.get("time"))
event_members = set(event.get("family_members", []))
event_duration = event.get("duration_minutes", 60 if event_time else 480)
for other in all_events:
if other.get("id") == event.get("id"):
continue
if other.get("status") == "declined":
continue
other_date = _event_date(other)
if other_date != event_date:
continue
other_members = set(other.get("family_members", []))
shared = event_members & other_members
if not shared:
continue
other_time = _parse_time(other.get("time"))
other_duration = other.get("duration_minutes", 60 if other_time else 480)
if event_time and other_time:
e_start = event_time[0] * 60 + event_time[1]
e_end = e_start + event_duration
o_start = other_time[0] * 60 + other_time[1]
o_end = o_start + other_duration
if (e_start < o_end) and (o_start < e_end):
overlap_minutes = min(e_end, o_end) - max(e_start, o_start)
conflicts.append({
"id": other.get("id"),
"title": other.get("title", "Untitled"),
"severity": "time_conflict",
"shared_members": list(shared),
"overlap_minutes": overlap_minutes
})
else:
conflicts.append({
"id": other.get("id"),
"title": other.get("title", "Untitled"),
"severity": "potential",
"shared_members": list(shared)
})
return conflicts
def _sort_key(event: Dict[str, Any]) -> tuple:
"""Sort key for events: timed first, then by time."""
t = _parse_time(event.get("time"))
if t:
return (0, t[0], t[1])
return (1, 0, 0) # All-day events last
def _format_event(e: Dict[str, Any], compact: bool = False, show_conflicts: bool = False, all_events: List[Dict[str, Any]] = None) -> str:
"""Format a single event for display."""
# Time display
time_str = e.get("time", "")
if time_str:
try:
# Parse 24h time and format nicely
h, m = map(int, time_str.split(":"))
suffix = "AM" if h < 12 else "PM"
h12 = h if h <= 12 else h - 12
if h12 == 0:
h12 = 12
time_display = f"{h12}:{m:02d} {suffix}"
except:
time_display = time_str
else:
time_display = "All day"
# Location
location = e.get("location", "")
location_str = f" 📍 {location}" if location else ""
# Family members (icons)
members = e.get("family_members", [])
member_icons = {
"Matt": "👨",
"Aundrea": "👩",
"Sullivan": "👦",
"Harper": "👧",
"Maggie": "🐕"
}
members_str = "".join(member_icons.get(m, "👤") for m in members) if members else ""
# Status indicator
status = e.get("status", "")
status_icon = ""
if status == "pending":
status_icon = "⏳ "
elif status == "needs_confirmation":
status_icon = "❓ "
# Conflicts
conflict_str = ""
if show_conflicts and all_events:
conflicts = _find_real_conflicts(e, all_events)
if conflicts:
sev = conflicts[0].get("severity", "")
if sev == "time_conflict":
conflict_str = f"\n ⚠️ CONFLICT: {conflicts[0]['title']}"
elif sev == "potential":
conflict_str = f"\n ⚡ Potential conflict: {conflicts[0]['title']}"
if compact:
return f" • {status_icon}{e['title']}{location_str} {members_str}"
else:
return f" {time_display} — {status_icon}**{e['title']}**{location_str} {members_str}{conflict_str}"
def _format_attention_item(e: Dict[str, Any], all_events: List[Dict[str, Any]]) -> str:
"""Format an item needing attention."""
lines = []
# Check if there's a counter-proposal
if e.get("counter_proposed_date"):
lines.append(f" 🔄 **{e['title']}**")
lines.append(f" Original: {e.get('start', 'Unknown')}")
lines.append(f" Counter: {e['counter_proposed_date']} at {e.get('counter_proposed_time', 'TBD')}")
lines.append(f" Reply with /confirm or /suggest another time")
else:
lines.append(f" ❓ **{e['title']}** — awaiting your confirmation")
conflicts = _find_real_conflicts(e, all_events)
if conflicts:
c = conflicts[0]
if c.get("severity") == "time_conflict":
lines.append(f" ⚠️ TIME CONFLICT with: {c['title']}")
return "\n".join(lines)
if name == "main":
print(generate_brief())