radicale birthday calendars

radicale webhooks

radicale/test.sh

@@ -0,0 +1,17 @@
+#!/bin/sh
+
+# USER="jintara"
+# PATH_="/data/collections/collection-root/jintara/bda0aa8d-8db8-1362-ee05-64d9cf7c8271/560b559b-130a-490e-acf8-1aa4e5ac4752.vcf" # дарья кухно
+# PATH_="/data/collections/collection-root/jintara/bda0aa8d-8db8-1362-ee05-64d9cf7c8271/8fb3b96b-0b08-4606-95e9-fec553f8dc5e.vcf" # ольга кузина
+
+USER="anton"
+#PATH_="/data/collections/collection-root/anton/c9cc103a-bb2f-cde2-0138-25126ce8ab3f/708ac282-53b5-4044-9352-55883fcc6013.vcf" # test birthday
+PATH_="/data/collections/collection-root/anton/c9cc103a-bb2f-cde2-0138-25126ce8ab3f/3f38ee57-74c8-4c65-99c7-ca6159e8bd7a.vcf" # Александр Кухно
+
+
+REQUEST="PUT"
+
+curl -s -X POST http://localhost:19000/hooks/on-change \
+  -H "Content-Type: application/json" \
+  -H "X-Secret: RUCVDCATrwtYlCT680FSW4IT1PrvkLB9" \
+  -d "{\"user\":\"$USER\",\"path\":\"$PATH_\",\"request\":\"$REQUEST\"}"

radicale/webhooks/hooks.yaml

@@ -0,0 +1,29 @@
+- id: on-change
+  execute-command: /usr/local/bin/python3
+  command-working-directory: /etc/webhook/scripts
+  include-command-output-in-response: true
+  pass-arguments-to-command:
+    - source: string
+      name: "/etc/webhook/scripts/on-change.py"
+    - source: payload
+      name: user
+    - source: payload
+      name: path
+    - source: payload
+      name: request
+  trigger-rule:
+    match:
+      type: value
+      value: RUCVDCATrwtYlCT680FSW4IT1PrvkLB9
+      parameter:
+        source: header
+        name: X-Secret
+- id: rescan
+  execute-command: /usr/local/bin/python3
+  command-working-directory: /etc/webhook/scripts
+  include-command-output-in-response: true
+  pass-arguments-to-command:
+    - source: string
+      name: "/etc/webhook/scripts/rescan.py"
+    - source: url
+      name: user

radicale/webhooks/radicale.sh

@@ -0,0 +1,15 @@
+#!/bin/sh
+USER="$1"
+PATH_="$2"
+REQUEST="$3"
+
+# Only fire on actual writes
+case "$REQUEST" in
+  PUT|DELETE) ;;
+  *) exit 0 ;;
+esac
+
+curl -s -X POST $WEBHOOKS_URI/on-change \
+  -H "Content-Type: application/json" \
+  -H "X-Secret: RUCVDCATrwtYlCT680FSW4IT1PrvkLB9" \
+  -d "{\"user\":\"$USER\",\"path\":\"$PATH_\",\"request\":\"$REQUEST\"}"

radicale/webhooks/scripts/config.yaml

@@ -0,0 +1,15 @@
+calendars:
+  anton:
+    base:
+      dir: /data/collections/collection-root/anton
+    dates_target: 7c3bf078-e287-ed95-f9f6-016fc84663df
+    rescan:
+      - path: c9cc103a-bb2f-cde2-0138-25126ce8ab3f        
+  jintara:
+    base:
+      dir: /data/collections/collection-root/jintara
+    dates_target: ebbbb557-5ce9-60cf-7a58-f25ace327913
+    rescan:
+      - path: bda0aa8d-8db8-1362-ee05-64d9cf7c8271
+
+    

radicale/webhooks/scripts/contactdates/addressbook.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+from pprint import pprint
+import vobject
+
+from .types import Event
+from .utils import parse_date
+
+def read_contact_date_field(card, target: list[Event], field_name: str, target_field_name: str):
+    if field_name in card.contents:
+        for idx, component in enumerate(card.contents[field_name]):  # datetime.date or string depending on format
+            field_id = f"{field_name}-{idx}"
+            omit_year = component.params.get("X-APPLE-OMIT-YEAR")
+            dt = parse_date(field_id, target_field_name, component.value, omit_year[0] if omit_year else False)
+            target.append(dt)
+
+def read_contact_abdate_field(card, target: list[Event]):
+    grouped_fields = {}
+
+    field_name = "x-abdate"
+    field_name_2 = "x-ablabel"
+    if field_name in card.contents:
+        for component in card.contents[field_name]:  # datetime.date or string depending on format
+            if component.group not in grouped_fields:
+                grouped_fields[component.group] = {}
+            grouped_fields[component.group]["value"] = component.value
+
+    if field_name_2 in card.contents:
+        for component in card.contents[field_name_2]:  # datetime.date or string depending on format
+            if component.group not in grouped_fields:
+                grouped_fields[component.group] = {}
+            grouped_fields[component.group]["name"] = component.value
+
+    # if len(grouped_fields):
+    #     pprint(card)
+
+    for group in grouped_fields:
+        f = grouped_fields[group]
+        # pprint(group)
+        # pprint(f)
+        if "name" in f and "value" in f:
+            target.append(parse_date(group, f["name"], f["value"]))
+
+def read_contact(path: str) -> tuple[vobject.vCard, list[Event]]:
+    
+    # print("---")
+
+    # with open(path, 'r') as file:
+    #     print(file.read())
+
+    # print("---")
+
+    with open(path) as f:
+        card = vobject.readOne(f)
+
+    result = []
+
+    read_contact_date_field(card, result, "bday", "birthday")
+    read_contact_date_field(card, result, "anniversary", "anniversary")
+    read_contact_date_field(card, result, "x-anniversary", "anniversary")
+    # result["x-abdate"] = read_contact_date_field(card, "x-abdate") # need specific parsing. also see X-ABDATE;X-ABLABEL
+    read_contact_date_field(card, result, "x-anniversary", "anniversary")
+    read_contact_date_field(card, result, "x-evolution-anniversary", "anniversary") # also see X-EVOLUTION-SPOUSE
+    read_contact_abdate_field(card, result) # also see X-EVOLUTION-SPOUSE
+
+    return card, result

radicale/webhooks/scripts/contactdates/addressbook_utils.py

@@ -0,0 +1,41 @@
+import vobject
+
+def get_contact_name(card: vobject.vCard) -> str:
+    """Extract best available display name from a vCard, never raises."""
+
+    # 1. FN — formatted name, most reliable
+    try:
+        fn = card.contents["fn"][0].value.strip()
+        if fn:
+            return fn
+    except (KeyError, IndexError, AttributeError):
+        pass
+
+    # 2. N — structured name: last, first, middle, prefix, suffix
+    try:
+        n = card.contents["n"][0].value
+        parts = [n.prefix, n.given, n.additional, n.family, n.suffix]
+        name = " ".join(p for p in parts if p and p.strip())
+        if name.strip():
+            return name.strip()
+    except (KeyError, IndexError, AttributeError):
+        pass
+
+    # 3. ORG — organization name as last resort
+    try:
+        org = card.contents["org"][0].value
+        # org can be a list ["Company", "Department"] or a string
+        if isinstance(org, list):
+            org = " / ".join(p for p in org if p and p.strip())
+        if org and org.strip():
+            return org.strip()
+    except (KeyError, IndexError, AttributeError):
+        pass
+
+    # 4. EMAIL
+    try:
+        return card.contents["email"][0].value.strip()
+    except (KeyError, IndexError, AttributeError):
+        pass
+
+    return "Unknown"

radicale/webhooks/scripts/contactdates/calendar.py

@@ -0,0 +1,27 @@
+import vobject
+from datetime import date, datetime
+from pathlib import Path
+
+from .types import Event
+
+def create_event(name: str, base_id: str, eventDate: Event) -> tuple[str, str]:
+
+    year = eventDate.year or datetime.now().year
+    dt = date(year, eventDate.month, eventDate.day)
+
+    id = base_id
+
+    cal = vobject.iCalendar()
+    event = cal.add('vevent')
+    event.add('uid').value = id
+    event.add('summary').value = f"{name}, {eventDate.name}"
+    event.add('description').value = (f"{eventDate.year}-" if eventDate.year is not None else "") + f"{eventDate.month:02}-{eventDate.day:02}"
+    event.add('dtstart').value = dt
+    event.add('rrule').value = 'FREQ=YEARLY'
+
+    return id, cal.serialize()
+
+def write_calendar_event(prefix: str, event: Event, dir: Path):
+    evtid, ics = create_event(prefix, f"{id}-{event.id}", event)
+    with open(dir / f"{evtid}.ics", 'w') as f:
+        f.write(ics)

radicale/webhooks/scripts/contactdates/types.py

@@ -0,0 +1,8 @@
+from typing import NamedTuple, Optional
+
+class Event(NamedTuple):
+    id: str
+    name: str
+    day: int
+    month: int
+    year: Optional[int] = None

radicale/webhooks/scripts/contactdates/utils.py

@@ -0,0 +1,24 @@
+import re
+from datetime import date, datetime
+
+from .types import Event
+
+def parse_date(id: str, name: str, input: str, apple_omit_year: str | None = None) -> Event:
+    if isinstance(input, date):
+        return Event(id, name, input.day, input.month, input.year)
+    
+    match = re.match(r"^(-|NaN)-(\d{2})-?(\d{2})$", input) # no-year format: --MMDD or --MM-DD. Also NaN-MM-DD spotted
+    if match:
+        return Event(id, name, int(match.group(3)), int(match.group(2)))
+    
+    for fmt in ("%Y-%m-%d", "%Y%m%d"):
+        try:
+            dt = datetime.strptime(input, fmt).date()
+            if apple_omit_year == f"{dt.year}":
+                return Event(id, name, dt.day, dt.month)
+            else:
+                return Event(id, name, dt.day, dt.month, dt.year)
+        except ValueError:
+            continue
+    
+    return None # cannot parse

radicale/webhooks/scripts/on-change.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+from pprint import pprint
+import sys
+import yaml
+from box import Box
+from pathlib import Path
+
+from contactdates import calendar, addressbook, addressbook_utils
+
+def load_config():
+    CONFIG_PATH = Path(__file__).parent / "config.yaml"
+
+    with open(CONFIG_PATH) as f:
+        return Box(yaml.safe_load(f))
+  
+def handle_contact_change(action: str, path: str, user: str):
+    config = load_config()
+  
+    base_path = Path(config.calendars[user].base.dir)
+    target_calendar = config.calendars[user].dates_target
+
+    dir = base_path / target_calendar
+
+    print(f"Reading contact {path}")
+
+    card, events = addressbook.read_contact(path)
+    name = addressbook_utils.get_contact_name(card)
+    id = card.contents["uid"][0].value.strip()
+
+
+    print(f"Writing contact \"{name}\" {len(events)} events to {dir}")
+
+    for event in events:
+        evtid, evt = calendar.create_event(name, f"{id}-{event.id}", event)
+        print(f"Writing contact \"{name}\" event {evtid}")
+        with open(dir / f"{evtid}.ics", 'w') as f:
+            f.write(evt)
+
+def handle_change():
+    user    = sys.argv[1] if len(sys.argv) > 1 else "?"
+    path    = sys.argv[2] if len(sys.argv) > 2 else "?"
+    request = sys.argv[3] if len(sys.argv) > 3 else "?"
+
+    # Distinguish contacts (CardDAV) vs calendar (CalDAV) by path
+    if ".vcf" in path:
+        kind = "CONTACT"
+    elif ".ics" in path:
+        kind = "EVENT"
+    else:
+        kind = "COLLECTION"
+
+    print(f"{request} {user} {kind} {path}")
+
+    if kind == "CONTACT":
+        handle_contact_change(request, path, user)
+
+def main():
+    handle_change()
+
+if __name__ == "__main__":
+    main()

radicale/webhooks/scripts/rescan.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+from pprint import pprint
+import sys
+import yaml
+import glob
+from box import Box
+from pathlib import Path
+
+from contactdates import calendar, addressbook, addressbook_utils
+
+def load_config():
+    CONFIG_PATH = Path(__file__).parent / "config.yaml"
+
+    with open(CONFIG_PATH) as f:
+        return Box(yaml.safe_load(f))
+    
+def process_contact(file: Path, target_dir: Path, user: str):
+    # print(f"Reading {user} contact {file.name}")
+    
+    card, events = addressbook.read_contact(str(file))
+    name = addressbook_utils.get_contact_name(card)
+    id = card.contents["uid"][0].value.strip()
+
+    if len(events):
+        print(f"Reading {user} contact {file.name}")
+        # pprint(card)
+        print(f"Writing contact \"{name}\" {len(events)} events to {target_dir}")
+
+        for file_to_delete in target_dir.glob(f"{id}-*.ics"):
+            print(f"Deleting old contact event {file_to_delete.name}")
+            file_to_delete.unlink()
+
+        for event in events:
+            evtid, evt = calendar.create_event(name, f"{id}-{event.id}", event)
+            print(f"Writing contact \"{name}\" event {evtid}")
+            with open(target_dir / f"{evtid}.ics", 'w') as f:
+                f.write(evt)
+
+def handle_rescan(user: str):
+    print(f"Rescanning {user} addressbooks")
+
+    config = load_config()
+
+    user_config = config.calendars[user]
+
+    pprint(user_config)
+    
+    base_path = Path(config.calendars[user].base.dir)
+    target_calendar = base_path / config.calendars[user].dates_target
+    addressbooks = user_config.rescan
+
+    for ab in addressbooks:
+        dir = base_path / ab.path
+        print(f"Rescanning {user} addressbook {dir}")
+
+        files = dir.glob("*.vcf")
+        for file in files:
+            process_contact(file, target_calendar, user)
+
+def main():    
+    user    = sys.argv[1] if len(sys.argv) > 1 else "?"
+    handle_rescan(user)
+
+if __name__ == "__main__":
+    main()

radicale/webhooks_image/Dockerfile

@@ -0,0 +1,6 @@
+FROM python:3.13-alpine
+COPY --from=almir/webhook /usr/local/bin/webhook /usr/local/bin/webhook
+USER root
+RUN pip install --no-cache-dir pyyaml python-box vobject
+EXPOSE 9000
+ENTRYPOINT ["webhook"]