leocrm/app/plugins/builtins/calendar/ics_utils.py

"""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