77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
import hashlib
|
|
import logging
|
|
from icalendar import Calendar
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _stable_event_key(component) -> str:
|
|
event_bytes = component.to_ical()
|
|
text = event_bytes.decode("utf-8", errors="replace")
|
|
stable_lines = []
|
|
for line in text.split("\r\n"):
|
|
key = line.split(":")[0].upper() if ":" in line else ""
|
|
if key in ("DTSTAMP", "LAST-MODIFIED"):
|
|
continue
|
|
stable_lines.append(line)
|
|
stable_text = "\r\n".join(stable_lines)
|
|
return hashlib.sha256(stable_text.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def parse_ics_events(ics_text: str) -> dict[str, str]:
|
|
cal = Calendar.from_ical(ics_text.encode() if isinstance(ics_text, str) else ics_text)
|
|
result = {}
|
|
for component in cal.walk():
|
|
if component.name != "VEVENT":
|
|
continue
|
|
uid = component.get("UID")
|
|
if uid is None:
|
|
logger.warning("Skipping event with no UID")
|
|
continue
|
|
uid_str = str(uid)
|
|
try:
|
|
result[uid_str] = _stable_event_key(component)
|
|
except Exception as e:
|
|
logger.warning("Failed to process event %s: %s", uid_str, e)
|
|
return result
|
|
|
|
|
|
def compute_diff(ics_uids: dict[str, str], known_uids: dict[str, str]) -> dict:
|
|
to_add = []
|
|
to_update = []
|
|
to_delete = []
|
|
|
|
for uid, file_hash in ics_uids.items():
|
|
if uid not in known_uids:
|
|
to_add.append((uid, file_hash))
|
|
elif known_uids[uid] != file_hash:
|
|
to_update.append((uid, file_hash))
|
|
|
|
for uid in known_uids:
|
|
if uid not in ics_uids:
|
|
to_delete.append(uid)
|
|
|
|
return {
|
|
"to_add": to_add,
|
|
"to_update": to_update,
|
|
"to_delete": to_delete,
|
|
}
|
|
|
|
|
|
def parse_ics_events_with_data(ics_text: str) -> dict[str, bytes]:
|
|
cal = Calendar.from_ical(ics_text.encode() if isinstance(ics_text, str) else ics_text)
|
|
result = {}
|
|
for component in cal.walk():
|
|
if component.name != "VEVENT":
|
|
continue
|
|
uid = component.get("UID")
|
|
if uid is None:
|
|
logger.warning("Skipping event with no UID")
|
|
continue
|
|
uid_str = str(uid)
|
|
try:
|
|
result[uid_str] = component.to_ical()
|
|
except Exception as e:
|
|
logger.warning("Failed to process event %s: %s", uid_str, e)
|
|
return result
|