diff --git a/src/auth_google.py b/src/auth_google.py new file mode 100644 index 0000000..62fd04b --- /dev/null +++ b/src/auth_google.py @@ -0,0 +1,62 @@ +"""Module for authenticating with Google Calendar API.""" + +import os +import pickle +from typing import List + +from google.auth.credentials import Credentials +from google.auth.transport.requests import Request +from google_auth_oauthlib.flow import InstalledAppFlow +from googleapiclient.discovery import Resource, build + +SCOPES: List[str] = ["https://www.googleapis.com/auth/calendar"] + + +def authenticate_google() -> Resource: + """Authenticate with Google Calendar API and return a service object. + + Attempts to load credentials from a pickle file, refresh them if expired, + or create new ones through OAuth2 flow if necessary. + + Returns: + Resource: An authenticated Google Calendar API service object. + """ + creds: Credentials | None = None + if os.path.exists("token.pickle"): + with open("token.pickle", "rb") as token: + creds = pickle.load(token) + + 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) + + with open("token.pickle", "wb") as token: + pickle.dump(creds, token) + + return build("calendar", "v3", credentials=creds) + + +def search_calendar_id(service: Resource, calendar_name: str) -> str: + """List all available Google calendars and return the ID of the target calendar. + + Args: + service: Authenticated Google Calendar API service object. + calendar_name: Name of the target Google Calendar. + + Returns: + str: The calendar ID of the target Google Calendar. + + Raises: + ValueError: If the target calendar is not found. + """ + calendars_result = service.calendarList().list().execute() + calendars = calendars_result.get("items", []) + + for calendar in calendars: + if calendar["summary"].lower() == calendar_name.lower(): + return calendar["id"] + + raise ValueError(f"No calendar named '{calendar_name}' found.") diff --git a/src/caldav_client.py b/src/caldav_client.py new file mode 100644 index 0000000..6af39c6 --- /dev/null +++ b/src/caldav_client.py @@ -0,0 +1,80 @@ +"""Module to interact with a CalDAV server and fetch events from a calendar.""" + +from typing import Any, Dict + +from caldav import Calendar as CalDAVCalendar +from caldav import DAVClient, Principal +from icalendar import Calendar + +EventDict = Dict[str, Any] +EventsDict = Dict[str, EventDict] + + +def connect_to_caldav(url: str, username: str, password: str) -> Principal: + """Connect to the CalDAV server and return the principal object. + + Args: + url: URL of the CalDAV server. + username: Username for authentication. + password: Password for authentication. + + Returns: + Principal: The authenticated CalDAV principal object. + """ + client = DAVClient(url, username=username, password=password) + return client.principal() + + +def get_calendar(principal: Principal, calendar_name: str) -> CalDAVCalendar: + """Get a specific calendar from the CalDAV principal. + + Args: + principal: Authenticated CalDAV principal object. + calendar_name: Name of the calendar to fetch. + + Returns: + CalDAVCalendar: The selected calendar object. + + Raises: + ValueError: If no matching calendar is found. + """ + calendars = principal.calendars() + if not calendars: + raise ValueError("No calendars found on the server.") + + for cal in calendars: + if cal.name.lower() == calendar_name.lower(): + return cal + + raise ValueError(f"No calendar named '{calendar_name}' found.") + + +def fetch_events(calendar: CalDAVCalendar) -> EventsDict: + """Fetch all events from the CalDAV calendar. + + Args: + calendar: CalDAV calendar object to fetch events from. + + Returns: + EventsDict: Dictionary of events indexed by their UIDs. + """ + events: EventsDict = {} + for event in calendar.events(): + ical = Calendar.from_ical(event.data) + for component in ical.walk(): + if component.name == "VEVENT": + uid = str(component.get("UID")) + dtstart = component.get("DTSTART") + dtend = component.get("DTEND") + last_modified = component.get("LAST-MODIFIED") + + events[uid] = { + "uid": uid, + "summary": str(component.get("SUMMARY")), + "start": dtstart.dt.isoformat() if dtstart else None, + "end": dtend.dt.isoformat() if dtend else None, + "last_modified": last_modified.dt.isoformat() if last_modified else None, + "google_event_id": None, + } + + return events diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..fc43df1 --- /dev/null +++ b/src/main.py @@ -0,0 +1,70 @@ +"""Main module for the calendar synchronization process.""" + +import os + +from auth_google import authenticate_google, search_calendar_id +from caldav_client import connect_to_caldav, fetch_events, get_calendar +from dotenv import load_dotenv +from sync_logic import ( + add_event_to_google, + compare_events, + delete_event_from_google, + error_events, + load_local_sync, + save_local_sync, +) + +load_dotenv() + +LOCAL_SYNC_FILE = "calendar_sync.json" +CALDAV_URL = os.getenv("CALDAV_URL") +CALDAV_USERNAME = os.getenv("CALDAV_USERNAME") +CALDAV_PASSWORD = os.getenv("CALDAV_PASSWORD") +CALDAV_CALENDAR_NAME = os.getenv("CALDAV_CALENDAR_NAME") +GOOGLE_CALENDAR_NAME = os.getenv("GOOGLE_CALENDAR_NAME") + + +def main() -> None: + """Run the calendar synchronization process.""" + try: + print("Authenticating with Google Calendar...") + service = authenticate_google() + google_calendar_id = search_calendar_id(service, GOOGLE_CALENDAR_NAME) + + print("Connecting to CalDAV server...") + principal = connect_to_caldav(CALDAV_URL, CALDAV_USERNAME, CALDAV_PASSWORD) + caldav_calendar = get_calendar(principal, CALDAV_CALENDAR_NAME) + + print(f"Fetching events from CalDAV calendar: {caldav_calendar.name}") + server_events = fetch_events(caldav_calendar) + + print("Loading local sync data...") + local_events = load_local_sync(LOCAL_SYNC_FILE) + + print("Comparing events...") + new_events, updated_events, deleted_events = compare_events(local_events, server_events) + + print(f"Adding/Updating {len(new_events) + len(updated_events)} events to Google Calendar...") + for event in new_events + updated_events: + add_event_to_google(service, event, google_calendar_id) + + print(f"Deleting {len(deleted_events)} events from Google Calendar...") + for event in deleted_events: + delete_event_from_google(service, event, google_calendar_id) + + print("Saving updated sync data...") + save_local_sync(LOCAL_SYNC_FILE, server_events) + + print("Sync process completed successfully.") + + if error_events: + print("The following events encountered errors during sync:") + for event in error_events: + print(event) + + except Exception as e: + print(f"Error occurred during sync: {e}") + + +if __name__ == "__main__": + main() diff --git a/src/sync_logic.py b/src/sync_logic.py new file mode 100644 index 0000000..448fe61 --- /dev/null +++ b/src/sync_logic.py @@ -0,0 +1,113 @@ +"""Logic for syncing events between a local calendar and Google Calendar.""" + +import json +import os +import time +from typing import Any, Dict, List, Tuple + +from googleapiclient.discovery import Resource + +EventDict = Dict[str, Any] +EventsDict = Dict[str, EventDict] + +error_events = [] + + +def load_local_sync(file_path: str) -> EventsDict: + """Load the locally synced events from a JSON file. + + Args: + file_path: Path to the JSON file. + + Returns: + EventsDict: Dictionary of previously synced events. + """ + if not os.path.exists(file_path): + return {} + with open(file_path, "r") as file: + return json.load(file) + + +def save_local_sync(file_path: str, events: EventsDict) -> None: + """Save the events to the local sync JSON file. + + Args: + file_path: Path to the JSON file. + events: Dictionary of events to save. + """ + with open(file_path, "w") as file: + json.dump(events, file, indent=4) + + +def compare_events( + local_events: EventsDict, + server_events: EventsDict, +) -> Tuple[List[EventDict], List[EventDict], List[EventDict]]: + """Compare local and server events to determine changes. + + Args: + local_events: Dictionary of locally stored events. + server_events: Dictionary of events from the server. + + Returns: + Tuple containing lists of new, updated, and deleted events. + """ + new_events: List[EventDict] = [] + updated_events: List[EventDict] = [] + deleted_events: List[EventDict] = [] + + for uid, event in server_events.items(): + if uid not in local_events: + new_events.append(event) + elif event["last_modified"] != local_events[uid].get("last_modified"): + # Preserve the Google Calendar event ID when updating + event["google_event_id"] = local_events[uid].get("google_event_id") + updated_events.append(event) + + for uid in local_events: + if uid not in server_events: + deleted_events.append(local_events[uid]) + + return new_events, updated_events, deleted_events + + +def add_event_to_google(service: Resource, event: EventDict, calendar_id: str) -> None: + """Add a single event to Google Calendar. + + Args: + service: Authenticated Google Calendar API service object. + event: Dictionary containing event details. + calendar_id: ID of the target Google Calendar. + """ + try: + print(f"Adding event to Google Calendar: {event['summary']}") + google_event = { + "summary": event["summary"], + "start": {"dateTime": event["start"], "timeZone": "UTC"}, + "end": {"dateTime": event["end"], "timeZone": "UTC"}, + } + created_event = service.events().insert(calendarId=calendar_id, body=google_event).execute() + event["google_event_id"] = created_event["id"] + print(f"Event created: {created_event.get('htmlLink')}") + time.sleep(0.5) # Prevent rate-limiting + except Exception as e: + print(f"Failed to add event: {event['summary']} - {e}") + error_events.append(event) + + +def delete_event_from_google(service: Resource, event: EventDict, calendar_id: str) -> None: + """Delete a single event from Google Calendar. + + Args: + service: Authenticated Google Calendar API service object. + event: Dictionary containing event details. + calendar_id: ID of the target Google Calendar. + """ + google_event_id = event.get("google_event_id") + if not google_event_id: + raise ValueError(f"Cannot delete event {event['summary']}: missing Google Calendar ID") + try: + service.events().delete(calendarId=calendar_id, eventId=google_event_id).execute() + print(f"Deleted event: {event['summary']}") + except Exception as e: + print(f"Failed to delete event: {event['summary']} - {e}") diff --git a/sync_caldav_to_google.py b/sync_caldav_to_google.py deleted file mode 100644 index f055176..0000000 --- a/sync_caldav_to_google.py +++ /dev/null @@ -1,324 +0,0 @@ -"""Synchronize CalDAV calendar events with Google Calendar.""" - -""" -CalDAV2Google. -Copyright (C) 2024 Roger Gonzalez - -This program is free software: you can redistribute it and/or modify -it under the terms of the GNU General Public License as published by -the Free Software Foundation, either version 3 of the License, or -(at your option) any later version. - -This program is distributed in the hope that it will be useful, -but WITHOUT ANY WARRANTY; without even the implied warranty of -MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -GNU General Public License for more details. - -You should have received a copy of the GNU General Public License -along with this program. If not, see . -""" -import json -import os -import pickle -import time -from typing import Any, Dict, List, Tuple - -from caldav import DAVClient, Principal, Calendar as CalDAVCalendar -from google.auth.credentials import Credentials -from google.auth.transport.requests import Request -from google_auth_oauthlib.flow import InstalledAppFlow -from googleapiclient.discovery import Resource, build -from icalendar import Calendar -from dotenv import load_dotenv - -load_dotenv(".env") - -EventDict = Dict[str, Any] -EventsDict = Dict[str, EventDict] - -CALDAV_URL: str = os.getenv("CALDAV_URL") -CALDAV_USERNAME: str = os.getenv("CALDAV_USERNAME") -CALDAV_PASSWORD: str = os.getenv("CALDAV_PASSWORD") -CALDAV_CALENDAR_NAME: str = os.getenv("CALDAV_CALENDAR_NAME") -GOOGLE_CALENDAR_NAME: str = os.getenv("GOOGLE_CALENDAR_NAME") - -LOCAL_SYNC_FILE: str = "calendar_sync.json" -SCOPES: List[str] = ["https://www.googleapis.com/auth/calendar"] - -errored = [] - -def authenticate_google() -> Resource: - """Authenticate with Google Calendar API and return a service object. - - Attempts to load credentials from a pickle file, refresh them if expired, - or create new ones through OAuth2 flow if necessary. - - Returns: - Resource: An authenticated Google Calendar API service object. - - Raises: - FileNotFoundError: If credentials.json is not found. - pickle.PickleError: If token.pickle cannot be read or written. - """ - creds: Credentials | None = None - if os.path.exists("token.pickle"): - with open("token.pickle", "rb") as token: - creds = pickle.load(token) - - 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) - - with open("token.pickle", "wb") as token: - pickle.dump(creds, token) - - return build("calendar", "v3", credentials=creds) - -def search_calendar_id(service: Resource) -> str: - """List all available Google calendars and return the ID of the target calendar. - - Args: - service: Authenticated Google Calendar API service object. - - Returns: - str: The calendar ID of the target Google Calendar. - - Raises: - ValueError: If the target calendar is not found. - """ - calendars_result = service.calendarList().list().execute() - calendars: List[Dict[str, Any]] = calendars_result.get("items", []) - - print("Available calendars:") - for calendar in calendars: - print(f"{calendar['summary']} (ID: {calendar['id']})") - if calendar["summary"].lower() == GOOGLE_CALENDAR_NAME.lower(): - return calendar["id"] - - raise ValueError("No calendar named 'Personal' found.") - -def add_event_to_google(service: Resource, event: EventDict, calendar_id: str) -> None: - """Add a single event to Google Calendar. - - It includes a small delay (0.5s) to prevent rate-limiting. - - Args: - service: Authenticated Google Calendar API service object. - event: Dictionary containing event details. - calendar_id: ID of the target Google Calendar. - - Returns: - None: but modifies the event dict to include google_event_id if successful - """ - try: - print(f"Adding event to Google Calendar: {event['summary']}") - google_event = { - "summary": event["summary"], - "start": {"dateTime": event["start"], "timeZone": "UTC"}, - "end": {"dateTime": event["end"], "timeZone": "UTC"}, - } - print(f"Google event body: {google_event}") - created_event = service.events().insert(calendarId=calendar_id, body=google_event).execute() - print(f"Event created: {created_event.get('htmlLink')}") - # Store the Google Calendar event ID - event["google_event_id"] = created_event["id"] - time.sleep(0.5) - except Exception as e: - print(f"Failed to add event: {event['summary']}. Error: {str(e)}") - errored.append(event) - - -def delete_event_from_google(service: Resource, event: EventDict, calendar_id: str) -> None: - """Delete a single event from Google Calendar. - - Args: - service: Authenticated Google Calendar API service object. - event: Dictionary containing event details. - calendar_id: ID of the target Google Calendar. - """ - google_event_id = event.get("google_event_id") - if not google_event_id: - print(f"Skipping deletion of event {event['summary']}: no Google Calendar ID found") - return - - print(f"Deleting event from Google Calendar: {event['summary']}") - try: - service.events().delete(calendarId=calendar_id, eventId=google_event_id).execute() - print(f"Event deleted: {event['summary']}") - except Exception as e: - print(f"Failed to delete event: {event['summary']} - {str(e)}") - raise - -def connect_to_caldav() -> Principal: - """Connect to the CalDAV server and return the principal object. - - Returns: - Principal: The authenticated CalDAV principal object. - - Raises: - Exception: If connection to the CalDAV server fails. - """ - client = DAVClient(CALDAV_URL, username=CALDAV_USERNAME, password=CALDAV_PASSWORD) - return client.principal() - -def get_calendar(principal: Principal) -> CalDAVCalendar: - """Get a specific calendar from the CalDAV principal. - - Args: - principal: Authenticated CalDAV principal object. - - Returns: - CalDAVCalendar: The selected calendar object. - - Raises: - ValueError: If no calendars are found on the server. - """ - calendars = principal.calendars() - if not calendars: - raise ValueError("No calendars found on the server.") - - print("Available CalDAV calendars:") - calendar = None - for i, cal in enumerate(calendars): - print(f"{i}: {cal.name}") - if cal.name.lower() == CALDAV_CALENDAR_NAME.lower(): - calendar = cal - - if not calendar: - raise ValueError(f"No calendar named {CALDAV_CALENDAR_NAME} found.") - - return calendar - - -def fetch_events(calendar: CalDAVCalendar) -> EventsDict: - """Fetch all events from the CalDAV calendar. - - Args: - calendar: CalDAV calendar object to fetch events from. - - Returns: - Dict[str, Dict[str, Any]]: Dictionary of events indexed by their UIDs. - """ - events: EventsDict = {} - for event in calendar.events(): - ical = Calendar.from_ical(event.data) - for component in ical.walk(): - if component.name == "VEVENT": - uid = str(component.get("UID")) - dtstart = component.get("DTSTART") - dtend = component.get("DTEND") - last_modified = component.get("LAST-MODIFIED") - - events[uid] = { - "uid": uid, - "summary": str(component.get("SUMMARY")), - "start": dtstart.dt.isoformat() if dtstart else None, - "end": dtend.dt.isoformat() if dtend else None, - "last_modified": last_modified.dt.isoformat() if last_modified else None, - "google_event_id": None # Initialize the Google Calendar event ID - } - - print(f"Total fetched events: {len(events)}") - return events - - -def load_local_sync() -> EventsDict: - """Load the locally synced events from JSON file. - - Returns: - Dict[str, Dict[str, Any]]: Dictionary of previously synced events. - """ - if not os.path.exists(LOCAL_SYNC_FILE): - return {} - with open(LOCAL_SYNC_FILE, "r") as file: - return json.load(file) - -def save_local_sync(events: EventsDict) -> None: - """Save the events to the local sync JSON file. - - Args: - events: Dictionary of events to save. - """ - with open(LOCAL_SYNC_FILE, "w") as file: - json.dump(events, file, indent=4) - -def compare_events( - local_events: EventsDict, - server_events: EventsDict, -) -> Tuple[List[EventDict], List[EventDict], List[EventDict]]: - """Compare local and server events to determine changes. - - Args: - local_events: Dictionary of locally stored events. - server_events: Dictionary of events from the server. - - Returns: - Tuple containing lists of new, updated, and deleted events. - """ - new_events: List[EventDict] = [] - updated_events: List[EventDict] = [] - deleted_events: List[EventDict] = [] - - for uid, event in server_events.items(): - if uid not in local_events: - new_events.append(event) - elif event["last_modified"] != local_events[uid].get("last_modified"): - # Preserve the Google Calendar event ID when updating - event["google_event_id"] = local_events[uid].get("google_event_id") - updated_events.append(event) - - for uid in local_events: - if uid not in server_events: - deleted_events.append(local_events[uid]) - - print(f"New events: {len(new_events)}") - print(f"Updated events: {len(updated_events)}") - print(f"Deleted events: {len(deleted_events)}") - - return new_events, updated_events, deleted_events - -def main() -> None: - """Run the calendar synchronization process.""" - print("Authenticating with Google Calendar...") - service = authenticate_google() - personal_calendar_id = search_calendar_id(service) - - print("Connecting to CalDAV server...") - principal = connect_to_caldav() - calendar = get_calendar(principal) - - print(f"Syncing calendar: {calendar.name}") - server_events = fetch_events(calendar) - - print("Loading local sync...") - local_events = load_local_sync() - print(f"Local events: {len(local_events)}") - - print("Comparing events...") - new_events, updated_events, deleted_events = compare_events(local_events, server_events) - - if new_events or updated_events: - print(f"Adding {len(new_events) + len(updated_events)} new/updated event(s) to Google Calendar...") - for event in new_events + updated_events: - add_event_to_google(service, event, personal_calendar_id) - - if deleted_events: - print(f"Deleting {len(deleted_events)} event(s) from Google Calendar...") - for event in deleted_events: - delete_event_from_google(service, event, personal_calendar_id) - - print("Saving updated sync...") - save_local_sync(server_events) - print("Sync completed.") - - if errored: - print("The following events failed to sync:") - for event in errored: - print(event) - -if __name__ == "__main__": - main()