Syncing ICS / iCal Feeds into Discourse Topics
This guide shows how to set up a Python script that automatically pulls events from an ICS (iCalendar) feed and posts them as [event] topics in Discourse. It’s tested on a fresh Ubuntu 24.04 droplet.
1. Prerequisites
On your server:
apt update
apt -y install python3 python3-venv python3-pip curl nano ca-certificates
- App directory + virtualenv
mkdir -p /opt/ics_sync
cd /opt/ics_sync
python3 -m venv venv
source venv/bin/activate
pip install --upgrade pip
pip install requests python-dateutil icalendar
3. Environment file
Create /opt/ics_sync/.env with your Discourse + ICS settings:
cat > /opt/ics_sync/.env <<'EOF'
export DISCOURSE_BASE_URL="https://yourforum.example"
export DISCOURSE_API_KEY="PASTE_YOUR_API_KEY_HERE"
export DISCOURSE_API_USERNAME="system"
# ICS feed (URL or local path)
export ICS_SOURCE="https://example.com/feed.ics"
# Category to post into
export DISCOURSE_CATEGORY_ID=4
# Optional defaults
export DEFAULT_TAGS="events,ics"
export SITE_TZ="Europe/London"
EOF
Load it:
set -a
source .env
set +a
4. Test API access
Check you can create an event manually:
curl -sS -X POST "$DISCOURSE_BASE_URL/posts.json" \
-H "Api-Key: $DISCOURSE_API_KEY" \
-H "Api-Username: $DISCOURSE_API_USERNAME" \
-F "title=API test $(date +%s)" \
-F 'raw=[event start="2025-10-10T10:00:00Z" end="2025-10-10T11:00:00Z" timezone="Europe/London"]Test[/event]' \
-F "category=$DISCOURSE_CATEGORY_ID"
If you see JSON back with a topic_id, your API key works.
5. The Python sync script
Save as /opt/ics_sync/ics_to_discourse.py (make it executable if you like).
Click to expand script
#!/usr/bin/env python3
import os
import argparse
import logging
from datetime import datetime, date, timedelta
from typing import Optional, Tuple
import requests
from icalendar import Calendar, Event
from dateutil.tz import gettz
# -------- Logging --------
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s"
)
log = logging.getLogger("ics_sync")
# -------- Helpers for dates --------
def _as_dt(value, site_tz):
tz = gettz(site_tz)
if isinstance(value, date) and not isinstance(value, datetime):
return datetime(value.year, value.month, value.day, 0, 0, 0, tzinfo=tz)
if isinstance(value, datetime):
return value if value.tzinfo else value.replace(tzinfo=tz)
raise TypeError(f"Unsupported dt value type: {type(value)}")
def _is_all_day(vevent):
dtstart_prop = vevent.get('dtstart')
if not dtstart_prop:
return False
try:
if getattr(dtstart_prop, 'params', {}).get('VALUE') == 'DATE':
return True
except Exception:
pass
val = vevent.decoded('dtstart', None)
return isinstance(val, date) and not isinstance(val, datetime)
def _fmt_iso_z(dt):
dt_utc = dt.astimezone(gettz('UTC'))
return dt_utc.replace(microsecond=0).isoformat().replace('+00:00', 'Z')
def _fmt_date(d):
return d.date().isoformat()
def _safe_attr(s: str) -> str:
return (s or '').replace('"', '\\"').strip()
# -------- Build Discourse body --------
def build_body(vevent, site_tz="Europe/London", rsvp=False) -> Tuple[str, str]:
title = str(vevent.get('summary', 'Untitled event')).strip() or "Event"
location = str(vevent.get('location', '') or '').strip()
description = str(vevent.get('description', '') or '').strip()
uid = str(vevent.get('uid', '') or '').strip()
url = str(vevent.get('url', '') or '').strip()
raw_start = vevent.decoded('dtstart', None)
raw_end = vevent.decoded('dtend', None)
all_day = _is_all_day(vevent)
start_dt = _as_dt(raw_start, site_tz) if raw_start else datetime.now(gettz(site_tz))
end_dt = _as_dt(raw_end, site_tz) if raw_end else (
start_dt + (timedelta(days=1) if all_day else timedelta(hours=1))
)
attrs = []
if all_day:
attrs.append(f'start="{_fmt_date(start_dt)}"')
attrs.append(f'end="{_fmt_date(end_dt)}"') # DTEND exclusive kept as provided/derived
else:
attrs.append(f'start="{_fmt_iso_z(start_dt)}"')
attrs.append(f'end="{_fmt_iso_z(end_dt)}"')
attrs.append(f'timezone="{_safe_attr(site_tz)}"')
if location:
attrs.append(f'location="{_safe_attr(location)}"')
if url:
attrs.append(f'url="{_safe_attr(url)}"')
if rsvp:
attrs.append('status="public"')
else:
attrs.append('status="standalone"')
attrs.append('minimal="true"')
bbcode = "[event " + " ".join(attrs) + "]\n" + title + "\n[/event]"
parts = [bbcode]
# Human-friendly details block (optional)
lines = []
if title:
lines.append(f"Event = {title}")
if location:
lines.append(f"Venue = {location}")
if url:
lines.append(f"\nURL = {url}")
if uid:
lines.append(f"\nUID: {uid}")
if lines:
parts.append("\n" + "\n".join(lines))
if description:
parts.append("\nEvent Description:\n" + description.strip())
body = "\n".join(parts)
return title, body
# -------- Validation & repair --------
def validate_and_repair_times(vevent, site_tz: str, repair: bool) -> Optional[str]:
raw_start = vevent.decoded('dtstart', None)
raw_end = vevent.decoded('dtend', None)
if raw_start is None:
return "Missing DTSTART"
all_day = _is_all_day(vevent)
start_dt = _as_dt(raw_start, site_tz)
if raw_end is None:
if not repair:
return "Missing DTEND"
vevent['dtend'] = start_dt + (timedelta(days=1) if all_day else timedelta(hours=1))
return None
end_dt = _as_dt(raw_end, site_tz)
if all_day:
if end_dt <= start_dt:
if not repair:
return "All-day DTEND <= DTSTART"
vevent['dtend'] = start_dt + timedelta(days=1)
return None
return None
if end_dt <= start_dt:
if not repair:
return "DTEND <= DTSTART"
vevent['dtend'] = start_dt + timedelta(hours=1)
return None
return None
# -------- Discourse posting --------
def post_to_discourse(raw_body: str, title: str, args, tags=None):
base = args.base_url or os.environ.get("DISCOURSE_BASE_URL")
api_key = args.api_key or os.environ.get("DISCOURSE_API_KEY")
api_user = args.api_user or os.environ.get("DISCOURSE_API_USERNAME", "system")
category_id = int(args.category or os.environ.get("DISCOURSE_CATEGORY_ID", "1"))
if not base or not api_key:
raise RuntimeError("DISCOURSE_BASE_URL and DISCOURSE_API_KEY are required")
url = f"{base.rstrip('/')}/posts.json"
headers = {
"Api-Key": api_key,
"Api-Username": api_user,
"Content-Type": "application/json",
}
payload = {
"title": title,
"raw": raw_body,
"category": category_id,
}
if tags:
payload["tags"] = [t.strip() for t in tags.split(",") if t.strip()]
resp = requests.post(url, json=payload, headers=headers, timeout=60)
if resp.status_code >= 400:
try:
log.error("==== Discourse API error ====")
log.error("Status: %s", resp.status_code)
log.error("%s", resp.json())
log.error("==== end error ====")
except Exception:
log.error("HTTP %s; Body: %s", resp.status_code, resp.text)
resp.raise_for_status()
return resp.json()
# -------- ICS loading --------
def load_calendar(ics_source: str) -> Calendar:
if ics_source.startswith("http://") or ics_source.startswith("https://"):
log.info("Fetching ICS from URL: %s", ics_source)
r = requests.get(ics_source, timeout=60)
r.raise_for_status()
data = r.content
else:
log.info("Reading ICS from file: %s", ics_source)
with open(ics_source, "rb") as f:
data = f.read()
return Calendar.from_ical(data)
# -------- Main --------
def main():
p = argparse.ArgumentParser(description="ICS → Discourse (events) sync")
p.add_argument("--ics", dest="ics", default=os.environ.get("ICS_SOURCE"), help="ICS URL or path (or ICS_SOURCE env)")
p.add_argument("--base-url", dest="base_url", default=os.environ.get("DISCOURSE_BASE_URL"))
p.add_argument("--api-key", dest="api_key", default=os.environ.get("DISCOURSE_API_KEY"))
p.add_argument("--api-user", dest="api_user", default=os.environ.get("DISCOURSE_API_USERNAME", "system"))
p.add_argument("--category", dest="category", default=os.environ.get("DISCOURSE_CATEGORY_ID", "1"))
p.add_argument("--tags", dest="tags", default=os.environ.get("DEFAULT_TAGS", ""))
p.add_argument("--site-tz", dest="site_tz", default=os.environ.get("SITE_TZ", "Europe/London"))
p.add_argument("--rsvp", action="store_true", help="Set status=public instead of standalone")
p.add_argument("--dry-run", action="store_true", help="Do not POST; print previews only")
p.add_argument("--future-only", action="store_true", help="Skip events that ended in the past (by site TZ)")
p.add_argument("--skip-errors", action="store_true", default=True, help="Skip events that error when posting")
p.add_argument("--repair-times", action="store_true", default=True, help="Auto-fix invalid/missing DTEND")
args = p.parse_args()
if not args.ics:
log.error("No ICS source provided. Use --ics or set ICS_SOURCE env var.")
return
mode = "DRY-RUN" if args.dry_run else "LIVE POST"
log.info("Starting ICS → Discourse sync")
log.info("Source: %s", args.ics)
log.info("Mode: %s", mode)
log.info("Site TZ: %s | RSVP: %s", args.site_tz, args.rsvp)
cal = load_calendar(args.ics)
now_tz = datetime.now(gettz(args.site_tz))
processed = 0
skipped = 0
failed = 0
for vevent in cal.walk('vevent'):
title = str(vevent.get('summary', 'Untitled event')).strip() or "Event"
# Future-only filter
raw_end = vevent.decoded('dtend', None) if 'dtend' in vevent else None
raw_start = vevent.decoded('dtstart', None) if 'dtstart' in vevent else None
if raw_end or raw_start:
try:
end_dt = _as_dt(raw_end, args.site_tz) if raw_end else None
ref_end = end_dt or (_as_dt(raw_start, args.site_tz) + timedelta(hours=1) if raw_start else None)
if args.future_only and ref_end and ref_end < now_tz:
skipped += 1
continue
except Exception:
if args.future_only:
skipped += 1
continue
# Validate/repair times
err = validate_and_repair_times(vevent, args.site_tz, repair=args.repair_times)
if err:
logging.warning("Skipping '%s': %s", title, err)
skipped += 1
continue
post_title, body = build_body(vevent, site_tz=args.site_tz, rsvp=args.rsvp)
if args.dry_run:
print("\n=== PREVIEW ===\n")
print(body)
processed += 1
continue
try:
resp = post_to_discourse(body, post_title, args, tags=args.tags)
topic_id = resp.get("topic_id")
if topic_id:
log.info("Posted event '%s' → Topic ID %s", post_title, topic_id)
processed += 1
except requests.HTTPError as e:
failed += 1
if args.skip_errors:
log.warning("Post failed for '%s' (%s). Skipping and continuing.", post_title, e)
continue
else:
raise
except Exception as e:
failed += 1
if args.skip_errors:
log.warning("Unexpected error for '%s': %s", post_title, e)
continue
else:
raise
log.info("Done. Processed %d event(s). Skipped %d. Failed %d.", processed, skipped, failed)
if __name__ == "__main__":
main()
- Running it
cd /opt/ics_sync
source venv/bin/activate
set -a; source .env; set +a
# Dry run (prints but does not post):
python3 ics_to_discourse.py --dry-run --future-only
# Live run:
python3 ics_to_discourse.py --future-only
7. Automating with systemd (optional)
Create a service:
cat > /etc/systemd/system/ics-sync.service <<'EOF'
[Unit]
Description=ICS to Discourse sync
[Service]
Type=oneshot
WorkingDirectory=/opt/ics_sync
EnvironmentFile=/opt/ics_sync/.env
ExecStart=/opt/ics_sync/venv/bin/python /opt/ics_sync/ics_to_discourse.py --future-only
EOF
Create a timer (e.g. hourly):
cat > /etc/systemd/system/ics-sync.timer <<'EOF'
[Unit]
Description=Run ICS sync hourly
[Timer]
OnCalendar=hourly
Persistent=true
[Install]
WantedBy=timers.target
EOF
systemctl daemon-reload
systemctl enable --now ics-sync.timer