tools/weekly_event_summary_tool.py

from agentpro.tools import Tool
from typing import Any, List, Tuple, Optional
import datetime
import dateutil.parser
import re


class WeeklyEventSummaryTool(Tool):
    """
    Tool to provide a narrative summary of Google Calendar events for an entire week.
    Users can ask for “this week,” “next week,” “last week,” or specify a date (e.g., “week of June 10, 2025”).
    The tool will gather events from Monday through Sunday of the chosen week and return a day-by-day narrative.
    """

    # ── 1) Class attributes ─────────────────────────────────────────────────────────────────
    name: str = "Weekly Event Summary"
    description: str = (
        "Provide a narrative summary of Google Calendar events for any requested week. "
        "You can ask for:\n"
        "  • “My events this week.”\n"
        "  • “Show my schedule next week.”\n"
        "  • “Weekly summary for last week.”\n"
        "  • “What do I have the week of June 10, 2025?”\n"
        "  • “Events for week of 2025-06-10.”\n"
        "If no explicit week is mentioned, defaults to this week (Monday→Sunday)."
    )
    action_type: str = "weekly_event_summary"
    input_format: str = (
        "Natural-language calendar query specifying a week. Examples:\n"
        "  • “What’s on my calendar this week?”\n"
        "  • “Show me my schedule next week.”\n"
        "  • “Weekly summary for last week.”\n"
        "  • “Events for the week of June 10, 2025.”\n"
        "  • “Week of 2025-06-10.”"
    )

    # ── 2) We expect a Google Calendar “service” to be passed in at instantiation ───────────
    service: Any

    # ── 3) Main entry point ─────────────────────────────────────────────────────────────────
    def run(self, input_text: Any) -> str:
        """
        Parse a natural-language weekly summary command, identify target week, and return a narrative summary.
        """
        text = str(input_text).strip().lower()

        # 1) Determine which week the user means
        week_start = self._resolve_week_start(text)
        if week_start is None:
            return (
                "Sorry, I couldn't determine which week you meant. "
                "Please ask for 'this week', 'next week', 'last week', or 'week of <date>'."
            )

        # 2) Build a list of dates from Monday through Sunday
        dates = [week_start + datetime.timedelta(days=i) for i in range(7)]
        week_end = dates[-1]

        # 3) For each day in the week, fetch events and build narrative parts
        narrative_parts: List[str] = []
        for day in dates:
            events = self._fetch_events_on_date(day)
            day_str = day.strftime("%A, %B %d, %Y")
            if not events:
                narrative_parts.append(f"On {day_str}, you have no events scheduled.")
            else:
                # Build a single sentence listing each event’s start time, end time, and title
                sentences: List[str] = []
                for idx, ev in enumerate(events):
                    start_raw = ev["start"].get("dateTime")
                    end_raw   = ev["end"].get("dateTime")
                    summary = ev.get("summary", "(no title)")

                    if start_raw and end_raw:
                        start_dt = datetime.datetime.fromisoformat(start_raw.replace("Z", "+00:00"))
                        end_dt   = datetime.datetime.fromisoformat(end_raw.replace("Z", "+00:00"))
                        start_str = start_dt.strftime("%I:%M %p").lstrip("0")
                        end_str   = end_dt.strftime("%I:%M %p").lstrip("0")
                        sentences.append(f"“{summary}” from {start_str} to {end_str}")
                    else:
                        # Should not happen for non-all-day events since we filter them
                        sentences.append(f"“{summary}” (all-day)")

                # Join individual event descriptions with “; ”
                day_events_str = "; ".join(sentences)
                narrative_parts.append(f"On {day_str}, you have: {day_events_str}.")

        # 4) Combine into one multiline narrative
        week_range_str = f"{week_start.strftime('%B %d, %Y')} to {week_end.strftime('%B %d, %Y')}"
        header = f"Weekly summary for {week_range_str}:"
        body = " ".join(narrative_parts)
        return f"{header}\n\n{body}"

    # ────────────────────────────────────────────────────────────────────────────────────────────

    def _resolve_week_start(self, text: str) -> Optional[datetime.date]:
        """
        Determine the Monday (week_start) of the requested week.
        Supports 'this week', 'next week', 'last week', or 'week of <date>'.
        If no keyword found, defaults to this week.
        """
        today = datetime.date.today()
        weekday = today.weekday()  # Monday=0 ... Sunday=6
        monday_this_week = today - datetime.timedelta(days=weekday)

        # 1) Check for 'this week'
        if "this week" in text:
            return monday_this_week

        # 2) 'next week'
        if "next week" in text:
            return monday_this_week + datetime.timedelta(days=7)

        # 3) 'last week'
        if "last week" in text:
            return monday_this_week - datetime.timedelta(days=7)

        # 4) 'week of <date>'
        # Look for a date substring to parse
        # e.g., "week of june 10, 2025" or "week of 2025-06-10"
        match = re.search(
            r"week\s+of\s+(.+)", text
        )
        if match:
            date_part = match.group(1).strip()
            try:
                parsed = dateutil.parser.parse(date_part, fuzzy=True)
                target_date = parsed.date()
                # Find Monday of that week
                wd = target_date.weekday()
                return target_date - datetime.timedelta(days=wd)
            except (ValueError, OverflowError):
                return None

        # 5) If no explicit keyword, default to this week
        return monday_this_week

    def _fetch_events_on_date(self, date_obj: datetime.date) -> List[dict]:
        """
        Fetch all non-all-day events on the provided date (UTC midnight → next midnight).
        Returns a list of event dicts (as returned by Google Calendar API), sorted by start time.
        """
        start_of_day = datetime.datetime.combine(date_obj, datetime.time.min).isoformat() + "Z"
        end_of_day = (datetime.datetime.combine(date_obj, datetime.time.min)
                      + datetime.timedelta(days=1)).isoformat() + "Z"

        events_res = (
            self.service.events()
                        .list(
                            calendarId="primary",
                            timeMin=start_of_day,
                            timeMax=end_of_day,
                            singleEvents=True,
                            orderBy="startTime"
                        )
                        .execute()
        )
        items = events_res.get("items", [])
        # Filter out all-day events (they have 'start.date' instead of 'start.dateTime')
        non_all_day = [ev for ev in items if ev.get("start", {}).get("dateTime")]
        return sorted(
            non_all_day,
            key=lambda ev: datetime.datetime.fromisoformat(ev["start"]["dateTime"].replace("Z", "+00:00"))
        )