Changes in this version of the ICS → Discourse sync script
Overview: What This Script Does and Who Should Use It
- Purpose: Keep Discourse topics in sync with an external calendar (.ics feed). Each event in your calendar is either created (if new) or updated (if changed), based on its UID.
- Who should use it: Discourse site admins who want a lightweight, automatable way to integrate and synchronize calendar events from external sources (Google Calendar, Outlook, university/club calendars, etc.).
- Integration: Works well alongside the discourse-calendar plugin, but does not require it.
- Deployment: Designed to run via cron or other task scheduler on any machine with Python 3 available.
How to Upgrade or Migrate from Previous Script Versions
To upgrade:
- Overwrite your existing script file with this new version.
- If you previously used
DEFAULT_TAGS, switch to the new DISCOURSE_DEFAULT_TAGS variable (but the script remains backward compatible).
- Check if your
.env or deployment config uses an integer for DISCOURSE_CATEGORY_ID—the script now accepts both integer and string values.
- If you want to enforce a tag length, set the
TAG_MAX_LEN environment variable.
No breaking changes:
This version maintains backwards compatibility except for improved all-day event handling and tag length caps, which only enhance compliance.
Here’s what’s new in the latest revision of the script, compared with the earlier one posted above:
- Environment variable for default tags
Old code used DEFAULT_TAGS; new code prefers DISCOURSE_DEFAULT_TAGS but still falls back for compatibility.
# old
DEFAULT_TAGS = [t.strip() for t in os.environ.get("DEFAULT_TAGS", "").split(",") if t.strip()]
# new
_default_tags_env = os.environ.get("DISCOURSE_DEFAULT_TAGS", os.environ.get("DEFAULT_TAGS", ""))
DEFAULT_TAGS = [t.strip() for t in _default_tags_env.split(",") if t.strip()]
- Category ID handling
Old code forced int(), which could crash on unset/non-numeric input. New code leaves it as a string, which Discourse accepts.
# old
CATEGORY_ID = int(os.environ.get("DISCOURSE_CATEGORY_ID", "1"))
# new
CATEGORY_ID = os.environ.get("DISCOURSE_CATEGORY_ID", "1") # keep as string; Discourse accepts string
- All-day event detection & formatting
Previously, all-day events were rendered as 00:00. New code detects DATE-style events or midnight–midnight spans and renders them cleanly.
# old (snippet inside build_body / human_range)
start_dt = _as_dt(vevent.decoded("dtstart"), tz)
end_dt = _as_dt(vevent.decoded("dtend"), tz) if vevent.get("dtend") else None
when = human_range(start_dt, end_dt) if start_dt and end_dt else ...
# new
start_dt = _as_dt(vevent.get("dtstart"), tz)
end_dt = _as_dt(vevent.get("dtend"), tz) if vevent.get("dtend") else None
all_day = _is_all_day(vevent, start_dt, end_dt)
when = human_range(start_dt, end_dt, all_day=all_day)
Making Events Render with the Discourse Calendar Plugin
If your site uses the discourse-calendar plugin, the [event] ... [/event] BBCode block will render as an interactive event and also appear in category/global calendars.
Sample output:
- UID tag shortening with configurable max length
Old code always used the full namespace + 10-char hash, risking tag length overflow.
New code trims the namespace so the whole tag fits within the site’s maximum tag length, which is now configurable via an environment variable TAG_MAX_LEN (default 30).
# old
def make_uid_tag(namespace, uid):
h = hashlib.sha1(uid.encode("utf-8")).hexdigest()[:10]
base = f"{namespace}-uid-{h}"
return base.lower()
# new
TAG_MAX_LEN = int(os.environ.get("TAG_MAX_LEN", "30")) # typical Discourse default is 30
def make_uid_tag(namespace, uid, max_len=30):
h = hashlib.sha1(uid.encode("utf-8")).hexdigest()[:10]
ns = _slugify(namespace)
fixed = f"-uid-{h}"
room_for_ns = max(1, max_len - len(fixed))
ns = ns[:room_for_ns]
return f"{ns}{fixed}"
- User-Agent header for API requests
Old code did not identify itself to Discourse. New code adds a lightweight User-Agent string.
# old
s.headers.update({
"Api-Key": API_KEY,
"Api-Username": API_USER,
"Content-Type": "application/json"
})
# new
s.headers.update({
"Api-Key": API_KEY,
"Api-Username": API_USER,
"Content-Type": "application/json",
"User-Agent": "ics2disc/1.0 (+Discourse ICS sync)"
})
How the Script Knows Whether to Update or Create a Topic
- Each calendar event (
VEVENT) has a unique identifier (UID).
- The script creates a tag based on this UID and searches for existing topics with that tag.
- If it finds a topic, it updates the first post only (does not change category, title, or human edits).
- If no topic exists for that UID, it creates a new topic in your configured category and with your configured tags.
- The logic is safe to run repeatedly (idempotent)—event topics will always reflect the latest data from your .ics feed.
Best Practices: Security, API Key, and Permissions
- Create a dedicated user for syncing (e.g.,
CalendarBot). Give it only “Create/Reply/See” permissions for the target category.
- API Key: Go to your site’s
/admin/api/keys page and create a “User API Key” for your sync bot. Copy it into your .env.
- Ensure your bot user can use and create tags in the designated category (adjust tag groups if they are restricted).
- Never post your API key publicly or commit
.env files to public repositories.
Deployment Checklist
- Python 3.7+ installed.
- Run
pip install requests icalendar python-dateutil pytz.
-
.env file created and filled out as shown above.
- User API Key created for a bot user with suitable permissions.
- Test the script using
python3 ./ics2disc.py --ics-url ... and check for errors in the terminal.
- Set up a cron job only after a manual test run is successful.
Minimal Working Example: Environment and Cron
Sample .env contents:
DISCOURSE_BASE_URL=[https://forum.example.com](https://forum.example.com/)
DISCOURSE_API_KEY=your_api_key_goes_here
DISCOURSE_API_USERNAME=system
DISCOURSE_CATEGORY_ID=12
DISCOURSE_DEFAULT_TAGS=calendar,ics
SITE_TZ=Europe/London
TAG_MAX_LEN=30
Install dependencies:
pip install requests icalendar python-dateutil pytz
Cron job to run every 30 minutes:
*/30 * * * * cd /path/to/ics2disc && /usr/bin/env bash -c 'source .env && ./ics2disc.py'
#!/usr/bin/env python3
"""
ICS -> Discourse topics (create/update by UID)
- Creates a new topic per VEVENT (keyed by UID tag).
- Updates the first post only (preserves human-edited titles; never moves categories).
- Tags each topic with:
* your default tags (from env),
* a namespace tag (slugified/truncated),
* a per-event UID tag = "<namespace>-uid-<10hex>" trimmed to TAG_MAX_LEN.
- Renders an [event] ... [/event] marker with When/Where/Description.
- Handles all-day events cleanly (no fake "00:00" times).
Environment variables
---------------------
DISCOURSE_BASE_URL e.g. https://forum.example.com
DISCOURSE_API_KEY your API key
DISCOURSE_API_USERNAME api username (e.g. system)
DISCOURSE_CATEGORY_ID category id (string ok) used on CREATE only
DISCOURSE_DEFAULT_TAGS comma-separated list (preferred)
DEFAULT_TAGS fallback env name for legacy setups
SITE_TZ IANA tz (e.g. Europe/London) for display; default UTC
ICS_NAMESPACE namespace prefix for tags; default "ics"
TAG_MAX_LEN max tag length (site setting); default "30"
CLI
---
--ics-url URL Fetch .ics from URL
--ics-file PATH Read .ics from local file
--namespace STR Override ICS_NAMESPACE for this run
--skip-errors Continue on event errors instead of aborting
"""
import os, sys, argparse, re, logging, hashlib
from datetime import datetime, date, timedelta
from dateutil.tz import gettz
from icalendar import Calendar
import requests
log = logging.getLogger("ics2disc")
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
# --- Config from environment ---
BASE = os.environ.get("DISCOURSE_BASE_URL", "").rstrip("/")
API_KEY = os.environ.get("DISCOURSE_API_KEY")
API_USER = os.environ.get("DISCOURSE_API_USERNAME")
CATEGORY_ID = os.environ.get("DISCOURSE_CATEGORY_ID", "1") # keep as string; API accepts string
_default_tags_env = os.environ.get("DISCOURSE_DEFAULT_TAGS", os.environ.get("DEFAULT_TAGS", ""))
DEFAULT_TAGS = [t.strip() for t in _default_tags_env.split(",") if t.strip()]
SITE_TZ = gettz(os.environ.get("SITE_TZ", "UTC"))
ICS_NAMESPACE = os.environ.get("ICS_NAMESPACE", "ics")
TAG_MAX_LEN = int(os.environ.get("TAG_MAX_LEN", "30")) # typical site default is 30
if not BASE or not API_KEY or not API_USER:
missing = [k for k, v in [
("DISCOURSE_BASE_URL", BASE),
("DISCOURSE_API_KEY", API_KEY),
("DISCOURSE_API_USERNAME", API_USER)
] if not v]
sys.exit(f"ERROR: Missing env: {', '.join(missing)}")
# --- HTTP session ---
def _session():
s = requests.Session()
s.headers.update({
"Api-Key": API_KEY,
"Api-Username": API_USER,
"Content-Type": "application/json",
"User-Agent": "ics2disc/1.0 (+Discourse ICS sync)"
})
return s
# --- Time helpers ---
def _as_dt(v, tz):
"""Accepts datetime/date/ical * and returns tz-aware datetime in tz."""
if isinstance(v, datetime):
return v.astimezone(tz) if v.tzinfo else v.replace(tzinfo=tz)
if isinstance(v, date):
return datetime(v.year, v.month, v.day, tzinfo=tz)
try:
if hasattr(v, "dt"):
return _as_dt(v.dt, tz)
except Exception:
pass
return None
def _is_all_day(vevent, start_dt, end_dt):
"""Treat DTSTART as DATE or midnight-to-midnight(+1) span as all-day."""
try:
raw = vevent.get("dtstart")
if hasattr(raw, "dt") and isinstance(raw.dt, date) and not isinstance(raw.dt, datetime):
return True
except Exception:
pass
if not start_dt or not end_dt:
return False
dur = end_dt - start_dt
return (start_dt.hour, start_dt.minute, start_dt.second) == (0, 0, 0) and \
(end_dt.hour, end_dt.minute, end_dt.second) == (0, 0, 0) and \
dur >= timedelta(days=1)
def human_range(start_dt, end_dt, all_day=False):
if not start_dt:
return ""
tzname = start_dt.tzname() or ""
if all_day:
if end_dt:
last_day = (end_dt - timedelta(days=1)).date() # DTEND exclusive
first_day = start_dt.date()
if first_day == last_day:
return f"All day {start_dt.strftime('%a %d %b %Y')} ({tzname})"
return f"{first_day.strftime('%a %d %b %Y')} – {last_day.strftime('%a %d %b %Y')} (all day, {tzname})"
return f"All day {start_dt.strftime('%a %d %b %Y')} ({tzname})"
if end_dt:
if start_dt.date() == end_dt.date():
return f"{start_dt.strftime('%a %d %b %Y, %H:%M')} – {end_dt.strftime('%H:%M')} ({tzname})"
return f"{start_dt.strftime('%a %d %b %Y, %H:%M')} – {end_dt.strftime('%a %d %b %Y, %H:%M')} ({tzname})"
return f"{start_dt.strftime('%a %d %b %Y, %H:%M')} ({tzname})"
# --- Body builder ---
def extract_marker_title(raw):
m = re.search(r"\[event\]\s*(.+?)\s*\[\/event\]", raw or "", re.I | re.S)
return m.group(1).strip() if m else None
def build_body(vevent, tz):
summary = (vevent.get("summary") or "").strip()
desc = (vevent.get("description") or "").strip()
loc = (vevent.get("location") or "").strip()
start_dt = _as_dt(vevent.get("dtstart"), tz)
end_dt = _as_dt(vevent.get("dtend"), tz) if vevent.get("dtend") else None
all_day = _is_all_day(vevent, start_dt, end_dt)
when = human_range(start_dt, end_dt, all_day=all_day)
parts = [f"[event] {summary} [/event]"]
if when:
parts.append(f"**When:** {when}")
if loc:
parts.append(f"**Where:** {loc}")
if desc:
parts.append("")
parts.append(desc)
raw = "\n".join(parts).strip()
return raw, summary, start_dt
# --- Tag helpers ---
def _slugify(s):
s = s.lower()
s = re.sub(r"[^a-z0-9\-]+", "-", s)
s = re.sub(r"-{2,}", "-", s).strip("-")
return s
def make_uid_tag(namespace, uid, max_len=30):
"""Build a per-event tag 'ns-uid-<10hex>' trimmed to max_len by shortening namespace only."""
h = hashlib.sha1(uid.encode("utf-8")).hexdigest()[:10]
ns = _slugify(namespace)
fixed = f"-uid-{h}"
room_for_ns = max(1, max_len - len(fixed))
ns = ns[:room_for_ns]
return f"{ns}{fixed}"
# --- Discourse lookups & edits ---
def find_topic_by_uid_tag(s, uid_tag):
"""
Return topic_id (int) for an existing topic carrying uid_tag, else None.
Try /tag/{tag}.json first; fall back to /search.json.
"""
try:
r = s.get(f"{BASE}/tag/{uid_tag}.json", timeout=30)
if r.status_code == 404:
log.debug("Tag %s not found via /tag JSON (404).", uid_tag)
elif r.status_code == 403:
log.debug("Forbidden on /tag JSON for %s (403) — will try search.json.", uid_tag)
else:
r.raise_for_status()
data = r.json() or {}
topics = ((data.get("topic_list") or {}).get("topics")) or []
for t in topics:
if uid_tag in (t.get("tags") or []):
log.info("Found existing topic %s via /tag JSON for %s.", t.get("id"), uid_tag)
return t.get("id")
except Exception as e:
log.debug("Tag JSON lookup failed for %s: %s", uid_tag, e)
try:
r = s.get(f"{BASE}/search.json", params={"q": f"tag:{uid_tag}"}, timeout=30)
r.raise_for_status()
data = r.json() or {}
topics = data.get("topics") or []
for t in topics:
if uid_tag in (t.get("tags") or []):
log.info("Found existing topic %s via search.json for %s.", t.get("id"), uid_tag)
return t.get("id")
log.info("No existing topic found for %s.", uid_tag)
except Exception as e:
log.warning("Search API lookup failed for %s: %s", uid_tag, e)
return None
def get_first_post_raw(s, topic_id):
"""Return (first_post_id, raw) by fetching /t/{id}.json?include_raw=1; fallback to /posts/{id}.json."""
r = s.get(f"{BASE}/t/{topic_id}.json", params={"include_raw": 1}, timeout=30)
r.raise_for_status()
data = r.json() or {}
posts = ((data.get("post_stream") or {}).get("posts")) or []
if posts:
fp = posts[0]
fp_id = fp.get("id")
raw = fp.get("raw")
if raw is not None:
return fp_id, raw
if fp_id:
r2 = s.get(f"{BASE}/posts/{fp_id}.json", params={"include_raw": 1}, timeout=30)
r2.raise_for_status()
d2 = r2.json() or {}
if "raw" in d2:
return fp_id, d2["raw"]
return None, None
def update_first_post(s, post_id, new_raw, reason=None):
payload = {"raw": new_raw}
if reason:
payload["edit_reason"] = reason
r = s.put(f"{BASE}/posts/{post_id}.json", json=payload, timeout=60)
if r.status_code >= 400:
log.error("Update post %s failed %s: %s", post_id, r.status_code, r.text)
r.raise_for_status()
return r.json()
def make_safe_title(summary, dtstart_dt):
"""Sanitize titles to avoid 'unclear' validator; keep entropy and cap length."""
summary = (summary or "").strip()
summary = re.sub(r'(.)\1{2,}', r'\1\1', summary) # collapse very long repeats
when = dtstart_dt.strftime("%a %d %b %Y %H:%M") if dtstart_dt else ""
title = f"{summary} — {when}".strip(" —")
alnums = [c.lower() for c in title if c.isalnum()]
if len(set(alnums)) < 6:
title = (title + " — event").strip()
return title[:120]
def create_topic(s, title, raw, category_id, tags, dtstart_dt=None):
"""
Create a new topic. Pads body to satisfy site min post length.
Retries once with sanitized title if validator complains.
Returns (topic_id, first_post_id).
"""
MIN_BODY = 40
raw = raw or ""
if len(raw) < MIN_BODY:
raw = (raw + "\n\n(autogenerated by ics2disc)").ljust(MIN_BODY + 1, " ")
payload = {"title": title, "raw": raw, "category": category_id}
if tags:
payload["tags"] = tags
r = s.post(f"{BASE}/posts.json", json=payload, timeout=60)
if r.status_code == 422:
try:
j = r.json()
errs = " ".join(j.get("errors") or [])
except Exception:
errs = r.text
if "Title seems unclear" in errs or "title" in errs.lower():
safe_title = make_safe_title(title, dtstart_dt)
if safe_title != title:
log.warning("Title rejected; retrying with sanitized title: %r", safe_title)
payload["title"] = safe_title
r = s.post(f"{BASE}/posts.json", json=payload, timeout=60)
if r.status_code >= 400:
log.error("Create failed %s: %s", r.status_code, r.text)
r.raise_for_status()
data = r.json()
return data["topic_id"], data["id"]
# --- Main VEVENT handler ---
def process_vevent(s, vevent, args, namespace):
uid = str(vevent.get("uid") or "").strip()
if not uid:
log.warning("Skipping VEVENT with no UID")
return
fresh_body, summary, start_dt = build_body(vevent, SITE_TZ)
# Tags: defaults + namespace + per-event uid tag (both slugified/capped)
uid_tag = make_uid_tag(namespace, uid, max_len=TAG_MAX_LEN)
ns_tag = _slugify(namespace)[:TAG_MAX_LEN] if namespace else "ics"
tags = list(DEFAULT_TAGS)
if ns_tag and ns_tag not in tags:
tags.append(ns_tag)
if uid_tag not in tags:
tags.append(uid_tag)
topic_id = find_topic_by_uid_tag(s, uid_tag)
if topic_id:
log.info("Found existing topic %s via tag %s.", topic_id, uid_tag)
first_post_id, old_raw = get_first_post_raw(s, topic_id)
if not first_post_id:
log.warning("Could not fetch first post raw for topic %s; defaulting to empty.", topic_id)
old_raw = ""
if (old_raw or "").strip() == fresh_body.strip():
log.info("No content change for topic %s.", topic_id)
else:
log.info("Updating topic #%s for UID %s …", topic_id, uid)
update_first_post(s, first_post_id, fresh_body, reason="ICS sync update")
log.info("Updated topic #%s", topic_id)
else:
auto_title = (summary or "").strip() or f"Event — {uid[:8]}"
log.info("Creating new topic for UID %s …", uid)
created_topic_id, first_post_id = create_topic(
s, auto_title, fresh_body, CATEGORY_ID, tags, dtstart_dt=start_dt
)
log.info("Created topic #%s (post %s)", created_topic_id, first_post_id)
# --- CLI entrypoint ---
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--ics-url", help="URL to .ics file")
ap.add_argument("--ics-file", help="Path to local .ics file")
ap.add_argument("--namespace", help="Tag namespace (defaults to ICS_NAMESPACE env)")
ap.add_argument("--skip-errors", action="store_true", help="Continue on event errors")
args = ap.parse_args()
feed_namespace = (args.namespace or ICS_NAMESPACE or "ics").strip()
if not (args.ics_url or args.ics_file):
sys.exit("ERROR: Provide --ics-url or --ics-file")
# Fetch ICS
if args.ics_url:
url = args.ics_url
log.info("Fetching ICS: %s", url)
r = requests.get(url, timeout=60, headers={"User-Agent": "ics2disc/1.0 (+Discourse ICS sync)"})
r.raise_for_status()
data = r.content
else:
with open(args.ics_file, "rb") as f:
data = f.read()
log.info("Using namespace: %s", feed_namespace)
cal = Calendar.from_ical(data)
s = _session()
for comp in cal.walk("VEVENT"):
try:
process_vevent(s, comp, args, feed_namespace)
except Exception as e:
if args.skip_errors:
uid = str(comp.get("uid") or "").strip()
log.error("Error on event UID=%s: %s", uid, e)
continue
raise
if __name__ == "__main__":
main()
Gotchas and Compatibility Notes
- Tag Groups: If your site restricts which tags are usable in a category (tag groups), make sure the bot user is allowed to create/use these tags.
- Category Names vs IDs: Always use the category’s numeric ID (find this in the URL when viewing the category in Discourse’s admin).
- api_username Permissions: Script requires API user (“CalendarBot”) to have Create/Reply/See on the target category, and permission to use tags.
- Tag length: If your Discourse site’s maximum tag length is customized, set
TAG_MAX_LEN accordingly.
- All-day events: The script now outputs proper date-only ranges so “all-day” events look correct in the Discourse event UI.
Common Troubleshooting Steps
- No topics created: Check script output/logs for errors (permissions, missing variables).
- API error responses: Double-check
DISCOURSE_API_KEY, DISCOURSE_API_USERNAME, and URL.
- Category issues: Verify the ID matches your Discourse site. Remember: category slug is not the same as numeric ID.
- Tagging problems: Make sure your bot can create the tags; see category tag settings.
- Hosted customer support: If you’re on official hosting and stuck, email team@discourse.org.
FAQ and Support
Q: Can I run this on Discourse’s managed hosting?
A: Yes, you can use the API to create and update topics, but you cannot run the script on the server itself. You must use your own machine, cloud VM, etc.
Q: My topics aren’t updating—how do I debug?
A: See the troubleshooting section above, and check for errors in your log output.
Q: Who do I contact for help?
A: For script or API issues, consult or post in the Meta support topic. For Discourse-hosted support, email team@discourse.org.
Q: Are pull requests/contributions accepted?
A: Absolutely! Open a discussion or submit code feedback on Meta.
References and Further Reading
With these changes:
