|
import requests |
|
from icalendar import Calendar |
|
from google.oauth2 import service_account |
|
from googleapiclient.discovery import build |
|
from googleapiclient.errors import HttpError |
|
from datetime import datetime, timedelta |
|
from googleapiclient.errors import HttpError |
|
from datetime import datetime |
|
from pprint import pp |
|
|
|
# Configuration |
|
ICS_URL = 'https://example.com/calendar.ics' # Replace with your ICS URL |
|
ICS_FILENAME = 'calendar.ics' |
|
SERVICE_ACCOUNT_FILE = 'path/to/credentials.json' # Credentials from google dev console |
|
SCOPES = ['https://www.googleapis.com/auth/calendar'] |
|
CALENDAR_ID = 'primary' # Important to replace with calendar ID from google calendar settings |
|
|
|
|
|
# Download ICS file |
|
def download_ics(url, filename): |
|
print(f"Downloading ICS file from {url}...") |
|
try: |
|
response = requests.get(url) |
|
response.raise_for_status() |
|
with open(filename, 'wb') as file: |
|
file.write(response.content) |
|
print("ICS file downloaded successfully.") |
|
except requests.RequestException as e: |
|
print(f"Failed to download ICS file: {e}") |
|
|
|
# Parse ICS file |
|
def parse_ics(filename): |
|
print(f"Parsing ICS file: {filename}...") |
|
events = [] |
|
try: |
|
with open(filename, 'rb') as file: |
|
cal = Calendar.from_ical(file.read()) |
|
for component in cal.walk(): |
|
if component.name == "VEVENT": |
|
dtstart = component.get('dtstart').dt |
|
summary = component.get('summary') |
|
if isinstance(dtstart, datetime): |
|
dtstart = dtstart.date() |
|
events.append({'date': dtstart, 'summary': summary}) |
|
print(f"Parsed {len(events)} events from ICS file.") |
|
except Exception as e: |
|
print(f"Failed to parse ICS file: {e}") |
|
return events |
|
|
|
# Google Calendar API setup |
|
def get_calendar_service(): |
|
print("Setting up Google Calendar API service...") |
|
try: |
|
credentials = service_account.Credentials.from_service_account_file( |
|
SERVICE_ACCOUNT_FILE, scopes=SCOPES) |
|
service = build('calendar', 'v3', credentials=credentials) |
|
print("Google Calendar API service set up successfully.") |
|
return service |
|
except Exception as e: |
|
print(f"Failed to set up Google Calendar API service: {e}") |
|
raise |
|
|
|
from googleapiclient.errors import HttpError |
|
|
|
def check_event_exists(service, event_date, summary): |
|
print(f"Checking if event '{summary}' on {event_date} exists in Google Calendar...") |
|
|
|
try: |
|
# Define time range for the entire day |
|
time_min = datetime.combine(event_date, datetime.min.time()).isoformat() + 'Z' |
|
time_max = datetime.combine(event_date, datetime.max.time()).isoformat() + 'Z' |
|
print(f"Time range for query: {time_min} to {time_max}") |
|
|
|
# Perform the search for events |
|
events_result = service.events().list( |
|
calendarId=CALENDAR_ID, |
|
timeMin=time_min, |
|
timeMax=time_max, |
|
q=summary, |
|
singleEvents=True |
|
).execute() |
|
|
|
events = events_result.get('items', []) |
|
pp(events) |
|
if not events: |
|
print(f"No events found for '{summary}' on {event_date}.") |
|
|
|
exists = False |
|
for event in events: |
|
event_summary = event.get('summary', 'No summary') |
|
start_date = event.get('start', {}).get('date', 'No start date') |
|
|
|
# Check if event matches the summary and start date |
|
if event_summary and summary in event_summary and start_date == event_date.isoformat(): |
|
exists = True |
|
print(summary, 'in', event_summary) |
|
print(f"Match found for event '{summary}' on {event_date}.") |
|
break |
|
|
|
print(f"Event exists in Google Calendar: {exists}") |
|
return exists |
|
|
|
except HttpError as error: |
|
print(f"An error occurred while checking for events: {error}") |
|
return False |
|
|
|
|
|
# Create new event |
|
def create_event(service, event_date, summary): |
|
print(f"Creating new event '{summary}' on {event_date}...") |
|
event = { |
|
'summary': summary, |
|
'start': { |
|
'date': event_date.isoformat(), |
|
'timeZone': 'UTC', |
|
}, |
|
'end': { |
|
'date': (event_date + timedelta(days=1)).isoformat(), |
|
'timeZone': 'UTC', |
|
}, |
|
} |
|
try: |
|
response = service.events().insert(calendarId=CALENDAR_ID, body=event).execute() |
|
print(f"Event '{summary}' created successfully. Event ID: {response.get('id')}") |
|
except HttpError as error: |
|
print(f"An error occurred while creating the event: {error}") |
|
|
|
# Main function |
|
def main(request=None): |
|
print("Starting script...") |
|
download_ics(ICS_URL, ICS_FILENAME) |
|
events = parse_ics(ICS_FILENAME) |
|
service = get_calendar_service() |
|
|
|
for event in events: |
|
if not check_event_exists(service, event['date'], event['summary']): |
|
print("Event not found online. Creating...") |
|
create_event(service, event['date'], event['summary']) |
|
|
|
print("Script completed.") |
|
|
|
|
|
if __name__ == '__main__': |
|
main() |