"""ICS export/import utilities — manual parser (no external icalendar dependency).""" from __future__ import annotations import uuid as uuid_lib from datetime import datetime from typing import Any from app.plugins.builtins.calendar.models import CalendarEntry def _fmt_ics_datetime(dt: datetime) -> str: """Format datetime as ICS UTC string: 20260101T120000Z.""" return dt.strftime("%Y%m%dT%H%M%SZ") def _fmt_ics_date(d: Any) -> str: """Format date as ICS date string: 20260101.""" if isinstance(d, datetime): return d.strftime("%Y%m%d") return str(d).replace("-", "") def export_entries_to_ics(entries: list[CalendarEntry]) -> str: """Export calendar entries to ICS format string.""" lines = [ "BEGIN:VCALENDAR", "VERSION:2.0", "PRODID:-//LeoCRM//Calendar//EN", ] for entry in entries: lines.append("BEGIN:VEVENT") lines.append(f"UID:{entry.id}@leocrm") if entry.start_at: lines.append(f"DTSTART:{_fmt_ics_datetime(entry.start_at)}") if entry.end_at: lines.append(f"DTEND:{_fmt_ics_datetime(entry.end_at)}") lines.append(f"SUMMARY:{_escape_ics(entry.title)}") if entry.description: lines.append(f"DESCRIPTION:{_escape_ics(entry.description)}") if entry.location: lines.append(f"LOCATION:{_escape_ics(entry.location)}") lines.append("END:VEVENT") lines.append("END:VCALENDAR") return "\r\n".join(lines) + "\r\n" def _escape_ics(text: str) -> str: """Escape special characters for ICS format.""" return text.replace("\\", "\\\\").replace(";", "\\;").replace(",", "\\,").replace("\n", "\\n") def parse_ics(content: str) -> list[dict[str, Any]]: """Parse ICS file content into a list of event dicts. Returns list of {uid, dtstart, dtend, summary, description, location}. """ events: list[dict[str, Any]] = [] current_event: dict[str, Any] | None = None current_key: str | None = None current_val: str = "" for raw_line in content.replace("\r\n", "\n").replace("\r", "\n").split("\n"): line = raw_line.strip() if not line: continue # Handle line folding (starts with space) if line.startswith(" ") and current_key: current_val += line[1:] continue # Save previous key if we were accumulating if current_key and current_event is not None: current_event[current_key] = current_val current_key = None current_val = "" if line == "BEGIN:VEVENT": current_event = {} elif line == "END:VEVENT": if current_key and current_event is not None: current_event[current_key] = current_val current_key = None current_val = "" if current_event is not None: events.append(current_event) current_event = None elif current_event is not None and ":" in line: # Parse property name:value (may have params like DTSTART;TZID=...) prop_part, value = line.split(":", 1) prop_name = prop_part.split(";")[0].upper() current_key = _map_ics_property(prop_name) current_val = value return events def _map_ics_property(name: str) -> str: """Map ICS property names to internal field names.""" mapping = { "UID": "uid", "DTSTART": "dtstart", "DTEND": "dtend", "SUMMARY": "summary", "DESCRIPTION": "description", "LOCATION": "location", } return mapping.get(name, name.lower()) def _parse_ics_datetime(val: str) -> datetime | None: """Parse ICS datetime string (various formats).""" val = val.strip() # Try UTC format: 20260101T120000Z try: return datetime.strptime(val, "%Y%m%dT%H%M%SZ") except ValueError: pass # Try local format: 20260101T120000 try: return datetime.strptime(val, "%Y%m%dT%H%M%S") except ValueError: pass # Try date only: 20260101 try: return datetime.strptime(val, "%Y%m%d") except ValueError: pass return None def ics_events_to_entry_data( events: list[dict[str, Any]], calendar_id: Any, tenant_id: Any, user_id: Any ) -> list[dict[str, Any]]: """Convert parsed ICS events into CalendarEntry-compatible dicts.""" entries = [] for event in events: title = event.get("summary", "Imported Event") start_at = _parse_ics_datetime(event.get("dtstart", "")) end_at = _parse_ics_datetime(event.get("dtend", "")) entry_data = { "id": uuid_lib.uuid4(), "tenant_id": tenant_id, "calendar_id": calendar_id, "entry_type": "appointment", "subtype": "normal", "title": title, "description": event.get("description"), "start_at": start_at, "end_at": end_at, "all_day": False, "location": event.get("location"), "due_date": None, "priority": "medium", "status": "open", "assigned_to": None, "reminder": None, "recurrence": None, "source_mail_id": None, "created_by": user_id, "deleted_at": None, } entries.append(entry_data) return entries