#!/usr/bin/env python3 """ IMAP to Nextcloud Calendar Importer Monitors an IMAP mailbox for emails with ICS attachments and imports them to Nextcloud calendar """ import imaplib import email from email.header import decode_header import caldav from caldav.elements import dav, cdav import logging import sys from datetime import datetime import os import re import ssl from icalendar import Calendar from bs4 import BeautifulSoup # Configuration from environment variables CONFIG = { # IMAP Settings 'IMAP_SERVER': os.getenv('IMAP_SERVER', 'imap.example.com'), 'IMAP_PORT': int(os.getenv('IMAP_PORT', '993')), 'IMAP_EMAIL': os.getenv('IMAP_EMAIL', ''), 'IMAP_PASSWORD': os.getenv('IMAP_PASSWORD', ''), 'IMAP_MAILBOX': os.getenv('IMAP_MAILBOX', 'INBOX'), 'IMAP_PROCESSED_FOLDER': os.getenv('IMAP_PROCESSED_FOLDER', 'Processed'), 'IMAP_USE_SSL': os.getenv('IMAP_USE_SSL', 'true').lower() == 'true', 'IMAP_USE_STARTTLS': os.getenv('IMAP_USE_STARTTLS', 'false').lower() == 'true', 'IMAP_VERIFY_SSL': os.getenv('IMAP_VERIFY_SSL', 'true').lower() == 'true', # Nextcloud CalDAV Settings 'CALDAV_URL': os.getenv('CALDAV_URL', ''), 'CALDAV_USERNAME': os.getenv('CALDAV_USERNAME', ''), 'CALDAV_PASSWORD': os.getenv('CALDAV_PASSWORD', ''), 'CALENDAR_NAME': os.getenv('CALENDAR_NAME', 'personal'), 'CALDAV_VERIFY_SSL': os.getenv('CALDAV_VERIFY_SSL', 'true').lower() == 'true', # Processing Settings 'MARK_AS_SEEN': os.getenv('MARK_AS_SEEN', 'true').lower() == 'true', 'MOVE_PROCESSED': os.getenv('MOVE_PROCESSED', 'false').lower() == 'true', 'CHECK_INTERVAL': int(os.getenv('CHECK_INTERVAL', '300')), # seconds 'RUN_ONCE': os.getenv('RUN_ONCE', 'false').lower() == 'true', # Filtering Settings 'SUMMARY_FILTER': os.getenv('SUMMARY_FILTER', ''), # Regex pattern or simple string 'SUMMARY_FILTER_MODE': os.getenv('SUMMARY_FILTER_MODE', 'contains'), # contains, regex, exact 'INVERT_FILTER': os.getenv('INVERT_FILTER', 'false').lower() == 'true', # If true, exclude matches # Deduplication Settings 'DEDUPE_BY_SUMMARY_PATTERN': os.getenv('DEDUPE_BY_SUMMARY_PATTERN', ''), # Regex to extract unique ID from summary (e.g., "Order (\d+)") 'DEDUPE_ENABLED': os.getenv('DEDUPE_ENABLED', 'false').lower() == 'true', # Enable deduplication by summary pattern # Logging 'LOG_LEVEL': os.getenv('LOG_LEVEL', 'INFO'), } # Setup logging log_level = getattr(logging, CONFIG['LOG_LEVEL'].upper(), logging.INFO) logging.basicConfig( level=log_level, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[logging.StreamHandler(sys.stdout)] ) logger = logging.getLogger(__name__) class IMAPCalendarImporter: def __init__(self, config): self.config = config self.imap = None self.caldav_client = None self.calendar = None def connect_imap(self): """Connect to IMAP server""" try: logger.info(f"Connecting to IMAP server: {self.config['IMAP_SERVER']}") # Create SSL context ssl_context = ssl.create_default_context() # Allow insecure connections if configured if not self.config['IMAP_VERIFY_SSL']: logger.warning("IMAP SSL verification disabled - accepting self-signed certificates") ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE # Choose connection method if self.config['IMAP_USE_STARTTLS']: # STARTTLS connection (port 143 typically) logger.info("Using STARTTLS for IMAP connection") self.imap = imaplib.IMAP4( self.config['IMAP_SERVER'], self.config['IMAP_PORT'] ) self.imap.starttls(ssl_context=ssl_context) elif self.config['IMAP_USE_SSL']: # Direct SSL/TLS connection (port 993 typically) logger.info("Using direct SSL/TLS for IMAP connection") self.imap = imaplib.IMAP4_SSL( self.config['IMAP_SERVER'], self.config['IMAP_PORT'], ssl_context=ssl_context ) else: # Plain connection (not recommended) logger.warning("Using unencrypted IMAP connection - not recommended!") self.imap = imaplib.IMAP4( self.config['IMAP_SERVER'], self.config['IMAP_PORT'] ) self.imap.login( self.config['IMAP_EMAIL'], self.config['IMAP_PASSWORD'] ) logger.info("Successfully connected to IMAP server") return True except Exception as e: logger.error(f"Failed to connect to IMAP: {e}") return False def connect_caldav(self): """Connect to Nextcloud CalDAV""" try: logger.info(f"Connecting to CalDAV: {self.config['CALDAV_URL']}") # Configure SSL verification if not self.config['CALDAV_VERIFY_SSL']: logger.warning("CalDAV SSL verification disabled - accepting self-signed certificates") import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) self.caldav_client = caldav.DAVClient( url=self.config['CALDAV_URL'], username=self.config['CALDAV_USERNAME'], password=self.config['CALDAV_PASSWORD'], ssl_verify_cert=self.config['CALDAV_VERIFY_SSL'] ) principal = self.caldav_client.principal() calendars = principal.calendars() # Find the target calendar for cal in calendars: cal_url = str(cal.url) if hasattr(cal.url, '__str__') else cal.url cal_name = cal.name if hasattr(cal, 'name') else '' logger.debug(f"Checking calendar: {cal_url} (name: {cal_name})") if self.config['CALENDAR_NAME'] in cal_url or self.config['CALENDAR_NAME'] == cal_name: self.calendar = cal logger.info(f"Found calendar: {cal_url}") break if not self.calendar: logger.error(f"Calendar '{self.config['CALENDAR_NAME']}' not found") available_cals = [] for cal in calendars: cal_url = str(cal.url) if hasattr(cal.url, '__str__') else cal.url cal_name = cal.name if hasattr(cal, 'name') else 'unknown' available_cals.append(f"{cal_name} ({cal_url})") logger.info(f"Available calendars: {available_cals}") return False logger.info("Successfully connected to CalDAV") return True except Exception as e: logger.error(f"Failed to connect to CalDAV: {e}") return False def decode_header_value(self, value): """Decode email header value""" if value is None: return "" decoded = decode_header(value) header_parts = [] for content, encoding in decoded: if isinstance(content, bytes): header_parts.append( content.decode(encoding or 'utf-8', errors='replace') ) else: header_parts.append(content) return ''.join(header_parts) def matches_filter(self, summary): """Check if event summary matches the filter""" if not self.config['SUMMARY_FILTER']: return True # No filter, accept all filter_pattern = self.config['SUMMARY_FILTER'] mode = self.config['SUMMARY_FILTER_MODE'] invert = self.config['INVERT_FILTER'] try: if mode == 'exact': matches = summary == filter_pattern elif mode == 'regex': matches = bool(re.search(filter_pattern, summary, re.IGNORECASE)) else: # contains matches = filter_pattern.lower() in summary.lower() # Invert if configured return not matches if invert else matches except Exception as e: logger.error(f"Error applying filter: {e}") return True # On error, accept the event def parse_ics_summary(self, ics_data): """Extract SUMMARY from ICS data""" try: cal = Calendar.from_ical(ics_data) for component in cal.walk(): if component.name == "VEVENT": summary = component.get('summary', '') if summary: return str(summary) return "" except Exception as e: logger.error(f"Error parsing ICS for summary: {e}") return "" def parse_ics_uid(self, ics_data): """Extract UID from ICS data""" try: cal = Calendar.from_ical(ics_data) for component in cal.walk(): if component.name == "VEVENT": uid = component.get('uid', '') if uid: return str(uid) return None except Exception as e: logger.error(f"Error parsing ICS for UID: {e}") return None def extract_dedupe_identifier(self, summary): """Extract deduplication identifier from summary using configured pattern""" if not self.config['DEDUPE_ENABLED'] or not self.config['DEDUPE_BY_SUMMARY_PATTERN']: return None try: match = re.search(self.config['DEDUPE_BY_SUMMARY_PATTERN'], summary) if match: # Return the first capture group if present, otherwise the whole match return match.group(1) if match.groups() else match.group(0) return None except Exception as e: logger.error(f"Error extracting dedupe identifier from '{summary}': {e}") return None def find_events_by_summary_pattern(self, dedupe_id): """Find existing calendar events that match the deduplication identifier""" matching_events = [] if not dedupe_id: return matching_events try: # Fetch all events from the calendar events = self.calendar.events() for event in events: try: cal = Calendar.from_ical(event.data) for component in cal.walk(): if component.name == "VEVENT": event_summary = str(component.get('summary', '')) event_dedupe_id = self.extract_dedupe_identifier(event_summary) if event_dedupe_id == dedupe_id: matching_events.append(event) logger.debug(f"Found matching event with summary: {event_summary}") except Exception as e: logger.debug(f"Error parsing event for deduplication: {e}") return matching_events except Exception as e: logger.error(f"Error searching for events by summary pattern: {e}") return matching_events def extract_product_table(self, email_message): """Extract and format product table from HTML email body""" try: for part in email_message.walk(): if part.get_content_type() == 'text/html': html_content = part.get_payload(decode=True) if html_content: html_text = html_content.decode('utf-8', errors='replace') soup = BeautifulSoup(html_text, 'html.parser') table = soup.find('table', id='productTable') if not table: continue lines = [] for row in table.find_all('tr'): # Skip header rows (containing th elements) if row.find('th'): continue cells = row.find_all('td') if len(cells) >= 2: # Extract cell text cell_texts = [cell.get_text(strip=True) for cell in cells] # Swap 1st and 2nd, add 'x' suffix to 2nd (now 1st) cell_texts[0], cell_texts[1] = f"{cell_texts[1]}x", cell_texts[0] lines.append(' '.join(cell_texts)) if lines: return '\n'.join(lines) return '' except Exception as e: logger.error(f"Error extracting product table: {e}") return '' def inject_description(self, ics_data, description): """Inject description into ICS data""" if not description: return ics_data try: cal = Calendar.from_ical(ics_data) for component in cal.walk(): if component.name == "VEVENT": component['description'] = description break return cal.to_ical().decode('utf-8') except Exception as e: logger.error(f"Error injecting description: {e}") return ics_data def extract_ics_attachments(self, email_message): """Extract ICS files from email message""" ics_files = [] for part in email_message.walk(): content_type = part.get_content_type() content_disposition = str(part.get('Content-Disposition', '')) # Look for ICS attachments if 'attachment' in content_disposition or content_type in [ 'text/calendar', 'application/ics', 'application/calendar' ]: filename = part.get_filename() if filename: filename = self.decode_header_value(filename) if filename and filename.lower().endswith('.ics'): try: ics_data = part.get_payload(decode=True) if ics_data: ics_text = ics_data.decode('utf-8', errors='replace') summary = self.parse_ics_summary(ics_text) uid = self.parse_ics_uid(ics_text) # Check if summary matches filter if self.matches_filter(summary): product_table = self.extract_product_table(email_message) ics_files.append({ 'filename': filename, 'data': ics_text, 'summary': summary, 'uid': uid, 'product_table': product_table }) logger.info(f"Found matching ICS attachment: {filename} (Summary: {summary}, UID: {uid})") if product_table: logger.info(f"Extracted product table with {len(product_table.splitlines())} items") else: logger.info(f"Skipping ICS (filtered out): {filename} (Summary: {summary})") except Exception as e: logger.error(f"Error extracting ICS from {filename}: {e}") return ics_files def import_ics_to_calendar(self, ics_data, filename, summary, uid, product_table=''): """Import ICS data to Nextcloud calendar, updating if UID exists or deleting duplicates by summary pattern""" try: # Inject product table as description if present ics_data = self.inject_description(ics_data, product_table) # Check for deduplication by summary pattern first dedupe_id = self.extract_dedupe_identifier(summary) if dedupe_id: logger.debug(f"Extracted deduplication identifier: {dedupe_id}") existing_events = self.find_events_by_summary_pattern(dedupe_id) if existing_events: logger.info(f"Found {len(existing_events)} existing event(s) matching dedupe ID '{dedupe_id}', deleting them") for event in existing_events: try: event.delete() logger.info(f"Deleted existing event with matching dedupe ID") except Exception as e: logger.error(f"Failed to delete existing event: {e}") # After deleting duplicates, create the new event event = self.calendar.save_event(ics_data) logger.info(f"Successfully imported event {filename} to calendar after deduplication (Summary: {summary}, UID: {uid})") return True # Fall back to UID-based update if no dedupe pattern matched existing_event = None if uid: try: logger.debug(f"Searching for existing event with UID: {uid}") # Search calendar for events with this UID events = self.calendar.search(uid=uid) if events: existing_event = events[0] logger.info(f"Found existing event with UID {uid}, will update it") except Exception as e: logger.debug(f"No existing event found with UID {uid}: {e}") if existing_event: # Update existing event existing_event.data = ics_data existing_event.save() logger.info(f"Successfully updated event {filename} in calendar (Summary: {summary}, UID: {uid})") else: # Create new event in calendar event = self.calendar.save_event(ics_data) logger.info(f"Successfully imported new event {filename} to calendar (Summary: {summary}, UID: {uid})") return True except Exception as e: logger.error(f"Failed to import {filename}: {e}") return False def process_emails(self): """Process unread emails with ICS attachments""" try: # Select mailbox self.imap.select(self.config['IMAP_MAILBOX']) # Search for unseen emails status, messages = self.imap.search(None, 'UNSEEN') if status != 'OK': logger.error("Failed to search emails") return email_ids = messages[0].split() logger.info(f"Found {len(email_ids)} unread emails") processed_count = 0 imported_count = 0 filtered_count = 0 for email_id in email_ids: try: # Fetch email status, msg_data = self.imap.fetch(email_id, '(RFC822)') if status != 'OK': logger.error(f"Failed to fetch email {email_id}") continue # Parse email email_body = msg_data[0][1] email_message = email.message_from_bytes(email_body) subject = self.decode_header_value(email_message['Subject']) from_addr = self.decode_header_value(email_message['From']) logger.info(f"Processing email: {subject} from {from_addr}") # Extract ICS attachments ics_files = self.extract_ics_attachments(email_message) if ics_files: for ics_file in ics_files: if self.import_ics_to_calendar( ics_file['data'], ics_file['filename'], ics_file['summary'], ics_file['uid'], ics_file.get('product_table', '') ): imported_count += 1 # Mark as processed if self.config['MARK_AS_SEEN']: self.imap.store(email_id, '+FLAGS', '\\Seen') if self.config['MOVE_PROCESSED']: self.imap.copy(email_id, self.config['IMAP_PROCESSED_FOLDER']) self.imap.store(email_id, '+FLAGS', '\\Deleted') processed_count += 1 except Exception as e: logger.error(f"Error processing email {email_id}: {e}") # Expunge deleted messages if we moved any if self.config['MOVE_PROCESSED'] and processed_count > 0: self.imap.expunge() logger.info(f"Processed {processed_count} emails, imported {imported_count} ICS files") except Exception as e: logger.error(f"Error in process_emails: {e}") def run(self): """Main execution""" logger.info("Starting IMAP to Nextcloud Calendar Importer") if not self.connect_imap(): logger.error("Failed to connect to IMAP, exiting") return False if not self.connect_caldav(): logger.error("Failed to connect to CalDAV, exiting") return False self.process_emails() # Cleanup try: self.imap.close() self.imap.logout() except: pass logger.info("Completed processing") return True def main(): # Validate required configuration required_vars = ['IMAP_EMAIL', 'IMAP_PASSWORD', 'CALDAV_URL', 'CALDAV_USERNAME', 'CALDAV_PASSWORD'] missing = [var for var in required_vars if not CONFIG.get(var)] if missing: logger.error(f"Missing required environment variables: {', '.join(missing)}") sys.exit(1) # Log filter configuration if CONFIG['SUMMARY_FILTER']: filter_desc = f"{CONFIG['SUMMARY_FILTER_MODE']}: '{CONFIG['SUMMARY_FILTER']}'" if CONFIG['INVERT_FILTER']: filter_desc = f"EXCLUDE {filter_desc}" logger.info(f"Summary filter active: {filter_desc}") # Log deduplication configuration if CONFIG['DEDUPE_ENABLED'] and CONFIG['DEDUPE_BY_SUMMARY_PATTERN']: logger.info(f"Deduplication enabled with pattern: '{CONFIG['DEDUPE_BY_SUMMARY_PATTERN']}'") importer = IMAPCalendarImporter(CONFIG) if CONFIG['RUN_ONCE']: importer.run() else: import time logger.info(f"Running in continuous mode (checking every {CONFIG['CHECK_INTERVAL']} seconds)") while True: try: importer.run() time.sleep(CONFIG['CHECK_INTERVAL']) except KeyboardInterrupt: logger.info("Shutting down...") break except Exception as e: logger.error(f"Unexpected error: {e}") time.sleep(CONFIG['CHECK_INTERVAL']) if __name__ == '__main__': main()