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