Refactored script to make it more readable
This commit is contained in:
parent
99ba891adb
commit
b03c199c7a
GPG Key ID:
C7ECE9C6C36EC2E6
62
src/auth_google.py
Normal file
62
src/auth_google.py
Normal file
@ -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.")
|
||||
80
src/caldav_client.py
Normal file
80
src/caldav_client.py
Normal file
@ -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
|
||||
70
src/main.py
Normal file
70
src/main.py
Normal file
@ -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()
|
||||
113
src/sync_logic.py
Normal file
113
src/sync_logic.py
Normal file
@ -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}")
|
||||
@ -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 <https://www.gnu.org/licenses/>.
|
||||
"""
|
||||
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()
|
||||
Loading…
x
Reference in New Issue
Block a user