flowdeck/app/services/importers/calendar.py

"""FlowDeck — iCalendar (.ics) importer (v5.6.0, Phase 4).

Parses VEVENT blocks (RFC 5545, best-effort) into a FlowDeck calendar
collection (start/end, all-day, location, description).
"""
from __future__ import annotations

import re
from typing import Any

from app.services.importers.base import (
    Importer,
    ImportResult,
    decode_text,
    make_collection,
    register_importer,
)

_SCHEMA = [
    {"name": "Title", "type": "title"},
    {"name": "Start", "type": "date"},
    {"name": "End", "type": "date"},
    {"name": "All day", "type": "checkbox"},
    {"name": "Location", "type": "text"},
    {"name": "Description", "type": "text"},
    {"name": "Calendar", "type": "text"},
]
_UNESCAPE = [("\\n", "\n"), ("\\N", "\n"), ("\\,", ","), ("\\;", ";"), ("\\\\", "\\")]


def _unfold(text: str) -> list[str]:
    lines: list[str] = []
    for raw in text.replace("\r\n", "\n").replace("\r", "\n").split("\n"):
        if raw[:1] in (" ", "\t") and lines:
            lines[-1] += raw[1:]
        else:
            lines.append(raw)
    return lines


def _unescape(value: str) -> str:
    for src, dst in _UNESCAPE:
        value = value.replace(src, dst)
    return value.strip()


def _parse_prop(line: str) -> tuple[str, dict[str, str], str]:
    if ":" not in line:
        return "", {}, ""
    head, _, value = line.partition(":")
    parts = head.split(";")
    name = parts[0].upper()
    params: dict[str, str] = {}
    for p in parts[1:]:
        if "=" in p:
            k, _, v = p.partition("=")
            params[k.upper()] = v
    return name, params, value


def _iso_datetime(value: str, params: dict[str, str]) -> tuple[str, bool]:
    """Return (iso, is_all_day)."""
    v = value.strip()
    if params.get("VALUE") == "DATE" or re.fullmatch(r"\d{8}", v):
        m = re.fullmatch(r"(\d{4})(\d{2})(\d{2})", v)
        return (f"{m.group(1)}-{m.group(2)}-{m.group(3)}", True) if m else ("", True)
    m = re.fullmatch(r"(\d{4})(\d{2})(\d{2})T(\d{2})(\d{2})(\d{2})(Z?)", v)
    if not m:
        return v, False
    date = f"{m.group(1)}-{m.group(2)}-{m.group(3)}T{m.group(4)}:{m.group(5)}:{m.group(6)}"
    return (date + "+00:00" if m.group(7) else date), False


@register_importer
class IcsImporter(Importer):
    source_id = "ics"
    label = "Calendrier (.ics)"
    description = "Export iCalendar (Google/Outlook/Apple) → collection d'événements."
    extensions = (".ics", ".ical")
    order = 33

    def detect(self, filename: str, data: bytes) -> bool:
        if filename.lower().endswith((".ics", ".ical")):
            return True
        return "BEGIN:VCALENDAR" in decode_text(data)[:2000]

    def parse(self, filename: str, data: bytes) -> ImportResult:
        result = ImportResult(source=self.source_id)
        lines = _unfold(decode_text(data))
        calendar = ""
        rows: list[dict] = []
        event: dict[str, Any] | None = None
        for line in lines:
            upper = line.strip().upper()
            if upper == "BEGIN:VEVENT":
                event = {}
                continue
            if upper == "END:VEVENT":
                if event is not None:
                    rows.append(_event_row(event, calendar))
                event = None
                continue
            name, params, value = _parse_prop(line.strip())
            if name == "X-WR-CALNAME" and not event:
                calendar = _unescape(value)
            if event is None:
                continue
            if name == "SUMMARY":
                event["title"] = _unescape(value)
            elif name == "DTSTART":
                event["start"], event["all_day"] = _iso_datetime(value, params)
            elif name == "DTEND":
                event["end"], _ = _iso_datetime(value, params)
            elif name == "LOCATION":
                event["location"] = _unescape(value)
            elif name == "DESCRIPTION":
                event["description"] = _unescape(value)
            elif name == "UID":
                event["uid"] = value
        result.pages.append(make_collection("Calendar", _SCHEMA, rows, source_path=filename))
        result.stats["rows"] = len(rows)
        return result.finalize()


def _event_row(event: dict[str, Any], calendar: str) -> dict:
    props: dict[str, Any] = {}
    if event.get("start"):
        props["Start"] = event["start"]
    if event.get("end"):
        props["End"] = event["end"]
    props["All day"] = bool(event.get("all_day"))
    if event.get("location"):
        props["Location"] = event["location"]
    if event.get("description"):
        props["Description"] = event["description"]
    if calendar:
        props["Calendar"] = calendar
    return {"title": (event.get("title") or "Event").strip()[:200], "properties": props}