Continuing the discussion from 
Support iCal Feed Sync in Discourse Calendar Plugin (Import from .ics URLs):
Many of us want iCal/ICS import/sync to complement the official Calendar/Event features. While that’s not built-in yet, here’s a small Python script you can run on a schedule to pull an .ics feed and create/update topics in a chosen category. It’s idempotent and uses each event’s UID to avoid duplicates.
What it does (quick)
- Fetches your
.icsfeed. - For each
VEVENT:- Derives a stable tag from the ICS
UID(e.g.ics-deadbeef12). - If a topic with that tag exists → updates the first post.
- If not → creates a new topic in your chosen category with your default tags.
- Derives a stable tag from the ICS
- Formats event details (summary, start/end, location, description, URL) into Markdown.
- Safe to rerun (e.g., every 15–30 minutes).
Requirements
-
Discourse API key (admin/global) and API username (e.g.
system). -
Tags enabled on your forum.
-
Python 3.10+ and:
pip install requests icalendar python-dateutil pytz
-
A scheduler (cron, systemd timer, GitHub Actions runner, etc.).
Configure via environment variables
export DISCOURSE_BASE_URL="https://forum.example.com"
export DISCOURSE_API_KEY="YOUR_API_KEY"
export DISCOURSE_API_USERNAME="system"
export ICS_URL="https://calendar.example.com/feed.ics"
export CATEGORY_ID="42" # numeric category id
export DEFAULT_TAGS="events,ics" # comma-separated
export SITE_TZ="Europe/London" # display timezone
The script
#!/usr/bin/env python3
"""
ICS -> Discourse topic sync
"""
import os, hashlib, requests, sys, html
from datetime import datetime
from dateutil import tz
from icalendar import Calendar
BASE = os.environ["DISCOURSE_BASE_URL"].rstrip("/")
API_KEY = os.environ["DISCOURSE_API_KEY"]
API_USER = os.environ.get("DISCOURSE_API_USERNAME", "system")
ICS_URL = os.environ["ICS_URL"]
CATEGORY_ID = int(os.environ["CATEGORY_ID"])
DEFAULT_TAGS = [t.strip() for t in os.environ.get("DEFAULT_TAGS", "events,ics").split(",") if t.strip()]
SITE_TZ = os.environ.get("SITE_TZ", "Europe/London")
S = requests.Session()
S.headers.update({
"Api-Key": API_KEY,
"Api-Username": API_USER,
"Content-Type": "application/json"
})
def fetch_ics(url: str) -> Calendar:
r = requests.get(url, timeout=60)
r.raise_for_status()
return Calendar.from_ical(r.content)
def to_aware_dt(v):
dt = getattr(v, "dt", v)
if isinstance(dt, datetime):
if dt.tzinfo is None:
dt = dt.replace(tzinfo=tz.UTC)
else:
dt = datetime(dt.year, dt.month, dt.day, tzinfo=tz.UTC)
site = tz.gettz(SITE_TZ)
return dt.astimezone(site)
def uid_tag(uid: str) -> str:
h = hashlib.sha1(uid.encode("utf-8")).hexdigest()[:10]
return f"ics-{h}"
def discourse_search_by_tag(tag: str):
r = S.get(f"{BASE}/search.json", params={"q": f"tag:{tag}"}, timeout=60)
r.raise_for_status()
js = r.json()
return js.get("topics") or js.get("results") or []
def get_topic_first_post_id(topic_id: int) -> int:
r = S.get(f"{BASE}/t/{topic_id}.json", timeout=60)
r.raise_for_status()
js = r.json()
return js["post_stream"]["posts"][0]["id"]
def create_topic(title: str, raw: str, tags):
payload = {"title": title,"raw": raw,"category": CATEGORY_ID,"tags[]": tags}
r = S.post(f"{BASE}/posts.json", data=payload, timeout=60)
r.raise_for_status()
return r.json()
def update_first_post(post_id: int, new_raw: str):
r = S.put(f"{BASE}/posts/{post_id}.json", json={"raw": new_raw}, timeout=60)
r.raise_for_status()
return r.json()
def fmt_when(start_local, end_local):
same_day = (start_local.date() == end_local.date())
if same_day:
return f"{start_local:%a %d %b %Y, %H:%M}–{end_local:%H:%M} ({SITE_TZ})"
return f"{start_local:%a %d %b %Y, %H:%M} → {end_local:%a %d %b %Y, %H:%M} ({SITE_TZ})"
def build_body(summary, start_local, end_local, loc, desc, src_url):
parts = []
parts.append(f"**When:** {fmt_when(start_local, end_local)}")
if loc: parts.append(f"**Where:** {loc}")
if src_url: parts.append(f"**Source calendar:** {src_url}")
parts.append("")
if desc: parts.append(desc); parts.append("")
parts.append("<small>Synced from ICS • Do not edit manually; changes will be overwritten on next sync.</small>")
return "\n".join(parts)
def event_title(summary, start_local):
return f"{summary} — {start_local:%Y-%m-%d %H:%M}"
def clean_text(v):
if not v: return ""
return html.unescape(str(v)).strip()
def main():
cal = fetch_ics(ICS_URL)
created, updated, skipped = 0, 0, 0
for comp in cal.walk():
if comp.name != "VEVENT": continue
uid = clean_text(comp.get("UID"))
if not uid: skipped += 1; continue
tag = uid_tag(uid)
summary = clean_text(comp.get("SUMMARY") or "Untitled event")
loc = clean_text(comp.get("LOCATION"))
desc = clean_text(comp.get("DESCRIPTION"))
src_url = clean_text(comp.get("URL") or ICS_URL)
start_local = to_aware_dt(comp.get("DTSTART"))
end_local = to_aware_dt(comp.get("DTEND") or comp.get("DTSTART"))
title = event_title(summary, start_local)
body = build_body(summary, start_local, end_local, loc, desc, src_url)
topics = discourse_search_by_tag(tag)
if topics:
tid = topics[0]["id"]
pid = get_topic_first_post_id(tid)
update_first_post(pid, body); updated += 1
else:
tags = DEFAULT_TAGS + [tag]
create_topic(title, body, tags); created += 1
print(f"Done. Created: {created}, Updated: {updated}, Skipped: {skipped}")
if __name__ == "__main__":
try: main()
except Exception as e: sys.stderr.write(f"Error: {e}\n"); sys.exit(1)
Cron example
*/30 * * * * /usr/bin/env -S bash -lc 'cd /opt/ics-sync && ./ics_to_discourse.py >> sync.log 2>&1'
Notes & caveats
• Uses stable UID-based tags (e.g. ics-deadbeef12) to find/update topics.
• Overwrites the first post each run; fence manual content if needed.
• Multiple calendars: run with different ICS_URL + category/tag sets.
• Meant as a stop-gap until ICS import/sync lands officially.
