calendarbot

main.py

import yaml
import os
import random
from datetime import datetime, timedelta, time
import pytz
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from google.auth.transport.requests import Request

# SCOPES: Read others' free/busy, Write events to your calendar
SCOPES = [
    "https://www.googleapis.com/auth/calendar",
    "https://www.googleapis.com/auth/calendar.events",
]
YAML_FILE = "people_config.yaml"
MEETING_DURATION_MINS = 30
DAYS_TO_SCAN = 7


def authenticate():
    creds = None
    # Load token.json if it exists (saved from previous run)
    # If not, trigger the browser flow:
    if not creds:
        flow = InstalledAppFlow.from_client_secrets_file("credentials.json", SCOPES)
        creds = flow.run_local_server(port=0)
        # Save creds to token.json for next time
    return build("calendar", "v3", credentials=creds)


# --- AUTHENTICATION ---
def authenticate_google():
    creds = None
    if os.path.exists("token.json"):
        creds = Credentials.from_authorized_user_file("token.json", SCOPES)

    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file("credentials.json", SCOPES)
            creds = flow.run_local_server(port=0)
        # Save the credentials for the next run
        with open("token.json", "w") as token:
            token.write(creds.to_json())

    return build("calendar", "v3", credentials=creds)


def get_free_busy(service, emails, time_min, time_max):
    # This queries Google for the *actual* meetings on their calendar
    body = {
        "timeMin": time_min.isoformat(),
        "timeMax": time_max.isoformat(),
        "items": [{"id": email} for email in emails],
    }
    return service.freebusy().query(body=body).execute()


def parse_people_config(file_path):
    with open(file_path, "r") as f:
        return yaml.safe_load(f)


def is_within_working_hours(slot_start, slot_end, person_config):
    # 1. Convert slot to person's local timezone
    tz = pytz.timezone(person_config["timezone"])
    local_start = slot_start.astimezone(tz)
    local_end = slot_end.astimezone(tz)

    # 2. Check Work Day Bounds
    w_start = time.fromisoformat(person_config["work_start"])
    w_end = time.fromisoformat(person_config["work_end"])

    if local_start.time() < w_start or local_end.time() > w_end:
        return False

    # 3. Check Lunch (No Fly Zone)
    if "lunch_start" in person_config:
        l_start = time.fromisoformat(person_config["lunch_start"])
        l_end = time.fromisoformat(person_config["lunch_end"])

        # If the meeting overlaps at all with lunch
        if not (local_end.time() <= l_start or local_start.time() >= l_end):
            return False

    return True


# Main Orchestrator
# def find_slots(service, person_A, person_B):
#     # 1. Get Google Busy Data
#     # 2. Generate candidates (e.g. every 30 mins for next 5 days)
#     # 3. Filter candidates:
#     #    - Is it marked 'busy' by Google?
#     #    - AND is it within Person A's YAML working hours?
#     #    - AND is it within Person B's YAML working hours?
#     pass


def is_time_in_range(start_dt, end_dt, start_time_str, end_time_str, user_tz):
    """
    Checks if a specific slot (start_dt to end_dt in UTC) falls strictly
    within a user's local start/end times (e.g., 09:00 to 17:00).
    """
    # Convert the UTC slot to the user's local time
    local_start = start_dt.astimezone(user_tz)
    local_end = end_dt.astimezone(user_tz)

    # Parse the abstract time strings (e.g., "09:00")
    # We allow the end time to be '00:00' if it represents midnight (next day) logic,
    # but for simplicity, we assume standard work hours here.
    lower_bound = datetime.strptime(start_time_str, "%H:%M").time()
    upper_bound = datetime.strptime(end_time_str, "%H:%M").time()

    # Check if the slot is entirely within working hours
    # Logic: Start time must be >= lower_bound AND End time must be <= upper_bound
    if local_start.time() >= lower_bound and local_end.time() <= upper_bound:
        return True
    return False


def is_overlapping_lunch(start_dt, end_dt, lunch_start_str, lunch_end_str, user_tz):
    """
    Checks if a slot overlaps with lunch.
    Returns True if it clashes (bad), False if it's safe (good).
    """
    if not lunch_start_str or not lunch_end_str:
        return False  # No lunch defined, so no clash

    local_start = start_dt.astimezone(user_tz)
    local_end = end_dt.astimezone(user_tz)

    l_start = datetime.strptime(lunch_start_str, "%H:%M").time()
    l_end = datetime.strptime(lunch_end_str, "%H:%M").time()

    # Overlap logic: A overlaps B if (StartA < EndB) and (EndA > StartB)
    # Here A is the meeting, B is lunch
    if local_start.time() < l_end and local_end.time() > l_start:
        return True
    return False


def check_google_busy(slot_start_utc, slot_end_utc, busy_list):
    """
    Checks if the slot overlaps with any 'busy' block returned by Google API.
    busy_list format from Google: [{'start': '...', 'end': '...'}, ...]
    """
    for busy in busy_list:
        # Google returns ISO strings with timezone info
        b_start = datetime.fromisoformat(busy["start"])
        b_end = datetime.fromisoformat(busy["end"])

        # Check for overlap
        if slot_start_utc < b_end and slot_end_utc > b_start:
            return True  # It is busy
    return False


def find_common_slots(
    person_a, person_b, google_busy_data, days_to_scan=5, duration_mins=30
):
    """
    person_a/b: Dicts from your YAML (email, timezone, work_start, etc.)
    google_busy_data: The specific 'calendars' dict from the freebusy query response
    """

    # 1. Setup Timezones
    tz_a = pytz.timezone(person_a["timezone"])
    tz_b = pytz.timezone(person_b["timezone"])

    # 2. Define Search Window (Start from next hour, scan X days)
    # Round up to next hour to avoid messy :13 start times
    now_utc = datetime.now(pytz.utc)
    start_search = now_utc.replace(minute=0, second=0, microsecond=0) + timedelta(
        hours=1
    )
    end_search = start_search + timedelta(days=days_to_scan)

    valid_slots = []

    # 3. Iterate through time in 30-minute increments
    current_slot = start_search
    while current_slot < end_search:
        slot_end = current_slot + timedelta(minutes=duration_mins)

        # Skip weekends (5=Saturday, 6=Sunday)
        if current_slot.weekday() >= 5:
            current_slot += timedelta(minutes=30)
            continue

        # --- CHECK PERSON A ---
        # Is it within work hours?
        if not is_time_in_range(
            current_slot, slot_end, person_a["work_start"], person_a["work_end"], tz_a
        ):
            current_slot += timedelta(minutes=30)
            continue

        # Is it during lunch?
        if is_overlapping_lunch(
            current_slot,
            slot_end,
            person_a.get("lunch_start"),
            person_a.get("lunch_end"),
            tz_a,
        ):
            current_slot += timedelta(minutes=30)
            continue

        # Is Google Calendar busy?
        busy_a = google_busy_data.get(person_a["email"], {}).get("busy", [])
        if check_google_busy(current_slot, slot_end, busy_a):
            current_slot += timedelta(minutes=30)
            continue

        # --- CHECK PERSON B ---
        # Is it within work hours?
        if not is_time_in_range(
            current_slot, slot_end, person_b["work_start"], person_b["work_end"], tz_b
        ):
            current_slot += timedelta(minutes=30)
            continue

        # Is it during lunch?
        if is_overlapping_lunch(
            current_slot,
            slot_end,
            person_b.get("lunch_start"),
            person_b.get("lunch_end"),
            tz_b,
        ):
            current_slot += timedelta(minutes=30)
            continue

        # Is Google Calendar busy?
        busy_b = google_busy_data.get(person_b["email"], {}).get("busy", [])
        if check_google_busy(current_slot, slot_end, busy_b):
            current_slot += timedelta(minutes=30)
            continue

        # --- SUCCESS ---
        # If we reached here, the slot works for both!
        valid_slots.append({"start": current_slot, "end": slot_end})

        current_slot += timedelta(minutes=30)

    return valid_slots


# --- MAIN EXECUTION ---
def main():
    print("1. Loading configuration...")
    with open(YAML_FILE, "r") as f:
        people = yaml.safe_load(f)

    if len(people) < 2:
        print("Error: Need at least 2 people in people_config.yaml")
        return

    # Pick 2 random people
    print("2. Selecting random partner...")
    partner = random.sample(people[1:], 1)
    p1, p2 = people[0], partner[0]
    print(
        f"   Selected: {p1['name']} ({p1['timezone']}) and {p2['name']} ({p2['timezone']})"
    )

    # Authenticate
    print("3. Authenticating with Google...")
    service = authenticate_google()

    # Query FreeBusy
    print("4. Fetching calendar availability...")
    time_min = datetime.now(pytz.utc)
    time_max = time_min + timedelta(days=DAYS_TO_SCAN)

    body = {
        "timeMin": time_min.isoformat(),
        "timeMax": time_max.isoformat(),
        "items": [{"id": p1["email"]}, {"id": p2["email"]}],
    }
    busy_response = service.freebusy().query(body=body).execute()

    # Find Slots
    print("5. Calculating intersections...")
    slots = find_common_slots(p1, p2, busy_response["calendars"])

    if not slots:
        print("   No overlapping time slots found in the next 7 days.")
        return

    # Pick a random slot from the valid ones
    selected_slot = random.choice(slots)
    final_start = selected_slot["start"]
    final_end = selected_slot["end"]

    print(
        f"   Found {len(slots)} valid slots. Selected: {final_start.strftime('%Y-%m-%d %H:%M UTC')}"
    )

    # Create Event
    print("6. Scheduling event...")
    event_body = {
        "summary": f"Random 1:1: {p1['name']} & {p2['name']}",
        "description": "This is an automated random 1:1 meeting generated by the bot.",
        "start": {"dateTime": final_start.isoformat()},
        "end": {"dateTime": final_end.isoformat()},
        "attendees": [
            {"email": p1["email"]},
            {"email": p2["email"]},
        ],
        "conferenceData": {
            "createRequest": {"requestId": f"random-{int(datetime.now().timestamp())}"}
        },
    }

    try:
        # event = (
        #     service.events()
        #     .insert(calendarId="primary", body=event_body, conferenceDataVersion=1)
        #     .execute()
        # )
        # print(f"   Success! Meeting scheduled. Link: {event.get('htmlLink')}")
        print(
            "   (Event scheduling code is currently commented out, but would schedule the meeting if uncommented)"
        )
    except Exception as e:
        print(f"   Error scheduling event: {e}")


if __name__ == "__main__":
    main()

people_config.yaml

- name: "Jeff Bryner"
  email: "0x7eff@jeffbryner.com"
  timezone: "America/New_York"
  work_start: "09:00"
  work_end: "17:00"
  lunch_start: "12:00"
  lunch_end: "13:00"

- name: "Jeffx2"
  email: "jeff@jeffbryner.com"
  timezone: "America/Los_Angeles"
  work_start: "08:30"
  work_end: "16:30"
  # Bob doesn't block lunch, so we omit those fields

requirements.txt

google-api-python-client
google-auth-httplib2 
google-auth-oauthlib 
pytz 
pyyaml