iCal/ICS feeds synchroniseren naar Discourse topics (simpele Python script, cron-vriendelijk)

Continuing the discussion from :date: 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 .ics feed.
  • 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.
  • 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.
4 likes

Hi everyone,

I’m not a coder, but I have previously managed to set up the official mail-receiver in Discourse by following the UI-based guide and configuring the API key directly via the admin panel—so I can handle step-by-step instructions and copying values when they’re spelled out clearly.

I came across this Python script for syncing .ICS feeds into Discourse topics, and it looks promising. But I’m stuck because I haven’t coded before and don’t know how to translate it into something I can manage.

Could someone please walk me through the simplest possible version of this—ideally with:

  1. Exactly what needs to go into each field or variable (e.g., how to locate or paste in the API key, base URL, category ID, etc.).
  2. The most basic instructions to schedule the script (like a cron line, but explained step by step).
  3. Any parts of the script I can simply copy and paste rather than having to understand Python.

Basically, I want to get this working the same way I managed the mail-receiver—via clear copy-and-paste instructions—even if someone has to point out exactly what parts to paste where.

Thanks in advance for helping a non-coder get this running!

3 likes

This looks really useful, thanks for sharing.

I can see how it would work for people who are comfortable running a Python script on a server with cron, but like Ethsim2 I’m not a coder. I’ve followed admin guides before (mail-receiver, API keys, etc.), so I can copy/paste values when told where they go, but I’d need a clearer breakdown of each step.

Would anyone be up for turning this into a “for admins” guide, similar to how the official mail-receiver instructions are written? That way more of us could test it and report back. It might also get wider adoption if it was packaged as a plugin or even just a template container that could be pulled in.

Happy to help test from the non-coder side if someone can show the way.

3 likes

Yes, exactly — that’s what I was hoping for too.

When I set up the official mail-receiver, I managed because the guide explained where to create the API key in the Discourse UI and then which values to paste into the config. I didn’t need to touch any code beyond copying what was shown.

If this ICS sync could be explained in the same way — “get your API key here, paste it here, run this exact command” — then I think people like me could get it working. A plugin or a template would make it even more accessible.

I’d definitely be willing to test and give feedback if someone can lay it out step-by-step for non-coders.

2 likes

Interesting approach.

I can see how a script like this is a good proof of concept, but I wonder whether it points toward a bigger gap: there isn’t currently an “official” way of consuming inbound ICS feeds in Discourse. Right now it’s up to custom scripts or manual import.

If this functionality were wrapped into a plugin (even just a thin layer over the API), it could cover a lot of real use cases:

  • University timetables or council meeting calendars flowing straight into topics.
  • Community events kept in sync without anyone having to re-enter details.
  • Easier testing and adoption by admins who aren’t comfortable running custom scripts.

So the question for me is: should this stay as an external script, or does it belong as a proper plugin that others can maintain and improve together? I’d be interested to hear what others think.

3 likes

Looks cool, thank you for sharing ! Is the first topic that gets created from the external sources becomes an event visible in the discourse calendar from the thread?

1 like

Thanks for the kind words, I really appreciate it.

From what I can see, the current script just brings the .ics entries into Discourse as topics — it doesn’t yet make them full calendar events that the #discourse-calendar plugin would recognise. But that’s still a useful step, and it shows the direction things can go.

If you do have a version of your own that you’re working on, please don’t hesitate to share it here. Even if it’s just a rough draft, it could give others a chance to try it out, compare approaches, and help us move toward a more complete solution together.

You’re welcome. I’m sorry but for the moment, my priority would be ical url exports from discourse events and not import, but I will share here if I ever come up with something. For now I just modify small stuffs like here to improve the create event button placement that is by default hidden in the plus menu.

Regarding your script and its content injection in the topics body, I remember that in the past you could add the syntax below to the composer and get an event created out of it :slight_smile:

As it doesn’t work anymore, I wonder if anyone knows if there is a new syntax as this could be integrated to your script in order to have the external event visible in the calendar of the category and in the global calendar as well.(well I don’t know if this would be as easy as that but who knows)

It used to be this syntax :

[event start="2025-11-16 19:00" status="public" name="585 test event" timezone="Europe/Paris" allowedGroups="trust_level_0" end="2025-11-16 21:00"] [/event]

yes, that syntax did render the “rich display of an event”, but i couldn’t then press and load the new topic on the PWA. Also there was no calendar category, and it wasn’t showing on the “Upcoming Events” month view, but it was showing with name “585 test event” on the agenda.

I don’t think the last error in my log is relevant, but it’s the last error there after this incident;


i’m running Discourse version v3.5.0.beta8 +305

discourse-cakeday is 1 commit behind


screen recording


Since, “Update All”, the same topic took multiple presses to open

i don’t think the syntax you have there has changed, has it?

thanks for showing me that this rich text still gets converted into an event. Not in my environment but it must because I’ve messed it up quiet well to do some settings for other things :slight_smile:

So, we may modify the body injection from the script with this syntax then maybe, I will try if I find time

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)
1 like

ICS → Discourse Sync (minimal-privilege setup + integrated script)


1) Create a service account

  • Admin → Users → New → create: CalendarBot (regular user, not staff).
  • Confirm the email so the account can exist normally.

2) Choose / prepare the import category

  • Pick a category (e.g. Events).
  • Security: grant CalendarBot (or a group it belongs to) Create / Reply / See.
  • You can later move topics elsewhere; sync still works because the script keys on a UID tag, not the category.

3) Create a User API Key for the bot (minimal scope)

Go to: /admin/api/keys+ NewUser API Key → assign to CalendarBot → copy key.

Granular permissions (what the key/account must be able to do & why)

Account (CalendarBot) needs:

  • Create / Reply / See on the import category
    (so it can create a topic if one doesn’t exist)
  • Edit own posts (default for any user)
    (so it can update the first post when an event changes)
  • Search permission (default)
    (so it can find existing topics by tag)

Endpoints the script uses (implicitly via Discourse UI API):

  • GET /search.json — to find existing topics by UID tag
  • GET /t/{topic_id}.json — to fetch the first post id to update
  • POST /posts — to create a new topic (first post)
  • PUT /posts/{post_id} — to update the first post’s cooked/raw

Tagging considerations:

  • The script generates a stable tag ics-<hash> from each iCal UID.
  • Make sure tagging is enabled and that the bot can use/create tags in that category
    (if you restrict tags via Tag Groups, allow CalendarBot access to those tags).

4) Install dependencies on your job runner

Example (Ubuntu/Debian):

  • sudo apt update && sudo apt install -y python3-pip
  • pip3 install requests python-dateutil icalendar

without discourse-calendar compatibility

5) Place the script (integrated build_body)

  • Path suggestion: /opt/ics-sync/ics_to_discourse.py
  • Make it executable: chmod +x /opt/ics-sync/ics_to_discourse.py

Script file contents:

# ----- 8< ----- ics_to_discourse.py ----- 8< -----
#!/usr/bin/env python3

"""
ICS → Discourse topic sync (cron-friendly, idempotent).

- Keys on iCal UID via a stable tag (ics-<sha1>), so moving topics across categories is fine.
- Creates a topic if missing; otherwise updates the first post.
- Includes an opinionated build_body() for clean event posts.

Reqs: requests, python-dateutil, icalendar
"""

import os, sys, hashlib, html, json, requests
from datetime import datetime, date, time, timedelta, timezone
from dateutil import tz
from icalendar import Calendar, Event

# --- Configuration via environment ---
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")

# --- HTTP session ---
S = requests.Session()
S.headers.update({
    "Api-Key": API_KEY,
    "Api-Username": API_USER,
    "Content-Type": "application/json"
})

# --- Helpers ---
def fetch_ics(url: str) -> Calendar:
    r = requests.get(url, timeout=60)
    r.raise_for_status()
    return Calendar.from_ical(r.content)

def to_site_tz(val):
    """Return aware datetime in site tz. Handles date-only all-day events."""
    dt = getattr(val, "dt", val)
    if isinstance(dt, datetime):
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
    elif isinstance(dt, date):
        # Treat date-only as all-day starting 00:00 UTC, then shift to site tz
        dt = datetime(dt.year, dt.month, dt.day, tzinfo=timezone.utc)
    else:
        raise ValueError("Unsupported DTSTART/DTEND type")
    site = tz.gettz(SITE_TZ)
    return dt.astimezone(site)

def fmt_when(start_local: datetime, end_local: datetime) -> str:
    """Pretty 'When' line. Handles same-day, cross-day, and all-day-ish cases."""
    same_day = start_local.date() == end_local.date()
    # Choose a readable format; adjust to your locale preference
    if same_day:
        return f"{start_local:%a %d %b %Y, %H:%M}–{end_local:%H:%M} ({SITE_TZ})"
    else:
        return f"{start_local:%a %d %b %Y, %H:%M} → {end_local:%a %d %b %Y, %H:%M} ({SITE_TZ})"

def clean_text(s) -> str:
    if not s:
        return ""
    # Keep simple; ICS descriptions often contain newlines; Discourse handles them
    return str(s).strip()

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):
    # Discourse /search.json returns different shapes depending on version & params
    r = S.get(f"{BASE}/search.json", params={"q": f"tag:{tag}"}, timeout=60)
    r.raise_for_status()
    js = r.json()
    topics = []
    if "topics" in js and isinstance(js["topics"], list):
        topics = js["topics"]
    elif "results" in js and isinstance(js["results"], list):
        # Some instances put topics under "results"
        topics = [x.get("topic") or x for x in js["results"] if isinstance(x, dict)]
    # Normalize to a list of dictionaries with at least 'id'
    norm = []
    for t in topics:
        if isinstance(t, dict) and "id" in t:
            norm.append(t)
    return norm

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, category_id: int, tags: list[str]) -> int:
    payload = {
        "title": title,
        "raw": raw,
        "category": category_id,
        "tags[]": tags
    }
    r = S.post(f"{BASE}/posts.json", data=json.dumps(payload), timeout=60)
    r.raise_for_status()
    js = r.json()
    return js["topic_id"]

def update_post(post_id: int, raw: str):
    payload = {"post[raw]": raw}
    r = S.put(f"{BASE}/posts/{post_id}.json", data=json.dumps(payload), timeout=60)
    r.raise_for_status()

# --- Body builder (integrated from step 9) ---
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)

# --- Main processing ---
def process_event(vevent: Event, default_tags: list[str], category_id: int, source_url: str):
    uid = clean_text(vevent.get("UID"))
    if not uid:
        # Skip items without UID; sync cannot be stable
        return

    summary = clean_text(vevent.get("SUMMARY"))
    location = clean_text(vevent.get("LOCATION"))
    description = clean_text(vevent.get("DESCRIPTION"))

    dtstart = vevent.get("DTSTART")
    dtend = vevent.get("DTEND")

    if not dtstart:
        return  # cannot schedule without start
    # RFC5545: if no DTEND, treat as instantaneous or derive a sensible default
    if not dtend:
        # assume 1 hour duration
        dtend = dtstart

    start_local = to_site_tz(dtstart)
    end_local = to_site_tz(dtend)

    body = build_body(summary or "(no title)", start_local, end_local, location, description, source_url)

    tag = uid_tag(uid)
    tags = list(dict.fromkeys([tag] + default_tags))  # keep order, dedupe

    # Search for existing topic by UID tag
    hits = discourse_search_by_tag(tag)

    if hits:
        topic_id = hits[0]["id"]
        post_id = get_topic_first_post_id(topic_id)
        update_post(post_id, body)
        print(f"UPDATED: {summary!r} (topic {topic_id}, tag {tag})")
    else:
        title = summary or "Untitled event"
        topic_id = create_topic(title, body, category_id, tags)
        print(f"CREATED: {summary!r} (topic {topic_id}, tag {tag})")

def main():
    try:
        cal = fetch_ics(ICS_URL)
    except Exception as e:
        print(f"Failed to fetch ICS: {e}", file=sys.stderr)
        sys.exit(1)

    count = 0
    for component in cal.walk():
        if component.name == "VEVENT":
            try:
                process_event(component, DEFAULT_TAGS, CATEGORY_ID, ICS_URL)
                count += 1
            except Exception as e:
                print(f"Failed to process VEVENT: {e}", file=sys.stderr)
                continue
    print(f"Done. Processed {count} VEVENTs.")

if __name__ == "__main__":
    main()
# ----- 8< ----- /ics_to_discourse.py ----- 8< -----
with discourse-calendar compatibility

# --- Step 5: Build first-post body (now with [event] BBCode) ---

from datetime import datetime, date, timedelta
from dateutil.tz import gettz

def _as_dt(value, site_tz):
    """
    Convert VEVENT dtstart/dtend value to an aware datetime.
    - If it's a date (all-day), return midnight localized to site_tz.
    - If it's a naive datetime, localize to site_tz.
    - If it's aware, leave as-is.
    """
    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):
        if value.tzinfo is None:
            return value.replace(tzinfo=tz)
        return value
    raise TypeError(f"Unsupported dt value type: {type(value)}")

def _is_all_day(vevent):
    """
    True if DTSTART is a pure DATE per ICS (VALUE=DATE) or presented as date only.
    """
    dtstart_prop = vevent.get('dtstart')
    if not dtstart_prop:
        return False
    # icalendar prop carries .params, check VALUE=DATE if present
    try:
        if getattr(dtstart_prop, 'params', {}).get('VALUE') == 'DATE':
            return True
    except Exception:
        pass
    # If decoded is a date (not datetime), also treat as all-day
    val = vevent.decoded('dtstart', None)
    return isinstance(val, date) and not isinstance(val, datetime)

def _fmt_iso_z(dt):
    """Format aware datetime as UTC ISO8601 with trailing Z."""
    dt_utc = dt.astimezone(gettz('UTC'))
    return dt_utc.replace(microsecond=0).isoformat().replace('+00:00', 'Z')

def _safe_attr(s: str) -> str:
    """Escape quotes for BBCode attribute values."""
    return (s or '').replace('"', '\\"').strip()

def build_body(vevent, site_tz="Europe/London"):
    """
    Return the first-post body for a topic, using [event] BBCode.
    Renders cleanly with the discourse-events plugin; degrades to raw text otherwise.
    """
    title = str(vevent.get('summary', 'Untitled event')).strip()
    location = str(vevent.get('location', '') or '').strip()
    description = str(vevent.get('description', '') or '').strip()
    uid = str(vevent.get('uid', '') or '').strip()

    # Pull start/end (handle missing DTEND by +1 hour default)
    raw_start = vevent.decoded('dtstart', None)
    raw_end = vevent.decoded('dtend', None)

    all_day = _is_all_day(vevent)

    if raw_start is None:
        # Shouldn't happen for valid ICS, but guard anyway
        start_dt = datetime.now(gettz(site_tz))
    else:
        start_dt = _as_dt(raw_start, site_tz)

    if raw_end is None:
        # RFC5545: if no DTEND and no DURATION, choose a sensible default
        if all_day:
            # all-day default: same day end at +1 day midnight
            end_dt = (start_dt + timedelta(days=1))
        else:
            end_dt = start_dt + timedelta(hours=1)
    else:
        end_dt = _as_dt(raw_end, site_tz)

    # Attributes for BBCode
    attrs = []
    attrs.append(f'start="{_fmt_iso_z(start_dt)}"')
    attrs.append(f'end="{_fmt_iso_z(end_dt)}"')
    attrs.append(f'all_day="{"true" if all_day else "false"}"')
    # include timezone to hint local rendering
    attrs.append(f'timezone="{_safe_attr(site_tz)}"')
    if location:
        attrs.append(f'location="{_safe_attr(location)}"')

    bbcode = "[event " + " ".join(attrs) + "]\n" + (title or "Event") + "\n[/event]"

    parts = [bbcode]
    if description:
        parts.append("")  # blank line
        parts.append(description.strip())
    if uid:
        parts.append("")
        parts.append(f"UID: {uid}")

    return "\n".join(parts)

6) Environment file

Create /opt/ics-sync/.env and edit values to match your site:

export DISCOURSE_BASE_URL="https://forum.example.com"
export DISCOURSE_API_KEY="PUT_THE_USER_API_KEY_HERE"
export DISCOURSE_API_USERNAME="CalendarBot"
export ICS_URL="https://example.com/path/to/calendar.ics"
export CATEGORY_ID=42
export DEFAULT_TAGS="events,ics"
export SITE_TZ="Europe/London"

Load it in your shell before testing:

  • cd /opt/ics-sync && source .env

7) First test run

  • cd /opt/ics-sync && ./ics_to_discourse.py

Expected:

  • New topics appear in the chosen category (first run).
  • Each topic carries a UID tag like ics-deadbeef12.
  • Re-runs update those topics, even if you move them to other categories.

8) Cron job (every 30 minutes example)

Edit crontab (crontab -e) and add:

*/30 * * * * /usr/bin/env -S bash -lc 'cd /opt/ics-sync && source .env && ./ics_to_discourse.py >> sync.log 2>&1'
about the cron job
*/30 * * * *    
    # Run every 30 minutes (on the hour and half past)

 /usr/bin/env -S bash -lc
    # Use /usr/bin/env to start bash
    # -S → split arguments (lets you pass multiple options)
    # bash -lc → run bash as a login shell (-l) and execute the command string (-c)

 'cd /opt/ics-sync && source .env && ./ics_to_discourse.py >> sync.log 2>&1'
    # Single-quoted command string passed to bash:
    #
    # cd /opt/ics-sync
    #     → change into the project directory
    #
    # source .env
    #     → load environment variables from the .env file
    #
    # ./ics_to_discourse.py
    #     → run the sync script
    #
    # >> sync.log 2>&1
    #     → append standard output (stdout) to sync.log
    #     → redirect standard error (stderr, "2") to the same place as stdout ("&1"),
    #       so errors also go into sync.log

4 likes

Just a non tested idea, but the build body function look to be built to append things easily, so with the right syntax that would be :

parts.append(
        f'[event start="{start_local}" end="{end_local}" '
        f'status="public" name="{summary}" timezone="Europe/Paris" '
        f'allowedGroups="trust_level_0"] [/event]'
    )
2 likes

That’s a good catch :+1: - wrapping in [event]...[/event] is definitely the way to get things showing nicely in the calendar UI.

Just to note: the variant under the details pane “with discourse-calendar compatibility” in my earlier post already does exactly this, so anyone who wants native event rendering can copy that block as-is.

Still, it’s helpful seeing it written out again here - makes it clearer what part of the body builder to tweak. If you’ve got any further refinements (extra attributes, or handling for all-day vs timed events), do share them - I think others following along would benefit.

2 likes

It’s cool to see all this interest & dev time investment here for building the missing pieces of discourse to expand communities. Events after all are one of the most important feature a community system can provide so nice to see it you see it too as a top priority :wink:

But I love already the event plugin in itself, discourse team really did a great job with the reminders, invitations, attendees management, and just having 1 thread per event with a global calendar for the category is is a luxury that you don’t find everywhere.

They provide us already so much for free that I can only be thankful no matter what feature is missing !

3 likes

Just to add a detail for anyone testing the script, that’s supposed to be compatible with discourse-events:

  • The [event] BBCode doesn’t recognize all_day="true".
    To make something render as all-day, the discourse-calendar plugin expects date-only values for start/end:
[event start=“2025-09-01” end=“2025-09-02” status=“standalone” minimal=“true”]
Welcome Week Fair
[/event]
  • Timed events need ISO datetimes with Z plus a timezone hint:
[event start=“2025-09-15T09:00:00Z” end=“2025-09-15T11:00:00Z” timezone=“Europe/London” status=“standalone” minimal=“true”]
Lecture: Linear Algebra
[/event]
  • One other note: right now the script parses LOCATION as a plain line outside of the BBCode.
    If you want the plugin to render it in the event header, it should go in the [event ...] tag itself:
[event start=“2025-09-15T09:00:00Z” end=“2025-09-15T11:00:00Z” timezone=“Europe/London” location=“Physics A2” status=“standalone” minimal=“true”]

Lecture: Linear Algebra

[/event]

That way LOCATION shows up properly in the event box, instead of only appearing further down in the post body.


Perhaps this revised code should be preferred for those whom are going to copy-paste and expect it to just work,


from datetime import datetime, date, timedelta
from dateutil.tz import gettz

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.replace(tzinfo=tz) if value.tzinfo is None else value
    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):
    # YYYY-MM-DD for all-day events
    return d.date().isoformat()

def _safe_attr(s: str) -> str:
    return (s or '').replace('"', '\\"').strip()

def build_body(vevent, site_tz="Europe/London", rsvp=False):
    """
    Return the first-post body for a topic, using [event] BBCode.
    - All-day events emit date-only start/end (no time).
    - Timed events emit UTC ISO8601 with trailing Z.
    - Adds optional status/minimal/url for nicer defaults.
    """
    title = str(vevent.get('summary', 'Untitled event')).strip()
    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()  # ICS URL field if present

    raw_start = vevent.decoded('dtstart', None)
    raw_end = vevent.decoded('dtend', None)
    all_day = _is_all_day(vevent)

    if raw_start is None:
        start_dt = datetime.now(gettz(site_tz))
    else:
        start_dt = _as_dt(raw_start, site_tz)

    if raw_end is None:
        end_dt = (start_dt + timedelta(days=1)) if all_day else (start_dt + timedelta(hours=1))
    else:
        end_dt = _as_dt(raw_end, site_tz)

    # Build attributes
    attrs = []
    if all_day:
        attrs.append(f'start="{_fmt_date(start_dt)}"')
        # RFC5545 all-day DTEND is exclusive; keep next day to render the span correctly
        attrs.append(f'end="{_fmt_date(end_dt)}"')
    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)}"')  # hint for UI

    if location:
        attrs.append(f'location="{_safe_attr(location)}"')
    if url:
        attrs.append(f'url="{_safe_attr(url)}"')

    # RSVP vs standalone
    if rsvp:
        attrs.append('status="public"')
    else:
        attrs.append('status="standalone"')
        attrs.append('minimal="true"')  # hide RSVP UI for info-only feeds

    bbcode = "[event " + " ".join(attrs) + "]\n" + (title or "Event") + "\n[/event]"

    parts = [bbcode]
    if description:
        parts.append("")
        parts.append(description.strip())
    if uid:
        parts.append("")
        parts.append(f"UID: {uid}")

    return "\n".join(parts)
to describe this script
Commentary on helper functions
  • _as_dt(value, site_tz)
    Converts an ICS DATE or DATETIME into a timezone-aware Python datetime.

    • If given a date-only (all-day) value, it becomes midnight in the site’s timezone.
    • If given a datetime with no timezone, it gets site_tz.
    • If already timezone-aware, it is returned unchanged.
  • _is_all_day(vevent)
    Checks if the event is defined as all-day in ICS (VALUE=DATE).
    If yes, the renderer should output date-only attributes instead of datetime stamps.

  • _fmt_iso_z(dt)
    Converts a datetime to UTC (Z), formatted in ISO8601.
    Example: 2025-08-20T18:30:00Z.

  • _fmt_date(d)
    Returns the date in YYYY-MM-DD format, used for all-day events.

  • _safe_attr(s)
    Escapes double quotes inside strings so they don’t break the [event] tag attributes.


Commentary on build_body() process
  • Extracts ICS fields (summary, location, description, uid, url, dtstart, dtend).

  • Handles missing start/end times:

    • If no dtstart, defaults to now.
    • If no dtend, assumes +1 day (all-day) or +1 hour (timed).
  • Determines whether the event is all-day via _is_all_day.

  • Builds the [event] BBCode attributes:

    • For all-day:
      [event start="2025-08-20" end="2025-08-21"]
      
      (end is exclusive per RFC5545, so it’s set to the following day).
    • For timed:
      [event start="2025-08-20T18:30:00Z" end="2025-08-20T20:00:00Z" timezone="Europe/London"]
      
  • Adds optional attributes:

    • location="..." if provided.
    • url="..." if provided.
  • Chooses RSVP style:

    • If rsvp=True[event ... status="public"]
    • Else → [event ... status="standalone" minimal="true"]
  • Constructs the body:

    1. The [event] block with the event’s title inside.
    2. Appends description if present.
    3. Appends UID if present.

Commentary on example output

Example ICS input:

  • Summary: Freshers Fair
  • Start: 2025-09-15 10:00 BST
  • End: 2025-09-15 16:00 BST
  • Location: University Park
  • Description: Clubs and societies fair.
  • UID: 123@uon.ac.uk

Generated post body:

[event start=“2025-09-15T09:00:00Z” end=“2025-09-15T15:00:00Z” timezone=“Europe/London” location=“University Park” status=“standalone” minimal=“true”]
Freshers Fair
[/event]

Clubs and societies fair.

UID: 123@uon.ac.uk

This can be pasted directly into Discourse. The [event] tag is parsed by the discourse-calendar plugin to show a nice event card.


Commentary on all-day vs timed events
  • All-day (e.g., 25th December 2025):
[event start=“2025-12-25” end=“2025-12-26” status=“standalone” minimal=“true”]
Christmas Day
[/event]
  • Timed (e.g., 20 Aug 2025, 18:30–20:00 London time):
[event start=“2025-08-20T17:30:00Z” end=“2025-08-20T19:00:00Z” timezone=“Europe/London” status=“standalone” minimal=“true”]
Evening Lecture
[/event]
3 likes

thanks, this makes a lot more sense now! the way you explained all-day vs timed events, and where the location goes, is super clear even for someone like me who isn’t from a coding background. i’ll try it with one of my uni timetable feeds and see how it shows up in discourse.

i think a short step-by-step for “admins who aren’t coders” could be really helpful alongside the script. happy to help test if that would be useful!

2 likes

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
  1. 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()
  1. 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
1 like

wow, i’ll test this out, but i’m not sure whether the previous scripts weren’t working because of post events and calendar features not being enabled in site settings.

Edit: i just need to go to that screen recording to find out what i did to that destroyed droplet

Thanks for sharing your script refinements — I’ve adapted it a bit further so that it can both create new topics and update existing ones by UID.

Key changes I made:

  • Preserves any manual title edits (script only updates if the title hasn’t been changed by a human).
  • Never touches the category on updates, so you can move topics freely without sync breaking.
  • Still posts events with [event] BBCode and a stable uid-... tag, so no duplicates.

Here’s the updated version:


#!/usr/bin/env python3
# ICS -> Discourse topics (create OR update by UID tag)
# Preserves human-edited titles and never changes category on update.
# Requirements: requests, python-dateutil, icalendar
import os, sys, argparse, re, json, logging
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", "system")
CATEGORY_ID = os.environ.get("DISCOURSE_CATEGORY_ID")  # numeric (string ok) - used on CREATE only
DEFAULT_TAGS = [t for t in os.environ.get("DISCOURSE_DEFAULT_TAGS", "").split(",") if t]
SITE_TZ = os.environ.get("SITE_TZ", "Europe/London")

# --- HTTP helpers ---
def _session():
    s = requests.Session()
    s.headers.update({
        "Api-Key": API_KEY,
        "Api-Username": API_USER,
        "Content-Type": "application/json"
    })
    return s

def _sanitize_tag(s):
    s = s.strip().lower()
    s = re.sub(r"[^a-z0-9\-]+", "-", s)
    s = re.sub(r"-{2,}", "-", s).strip("-")
    return s or "event"

# --- Time helpers ---
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 is not None 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):
    return dt.astimezone(gettz('UTC')).strftime("%Y-%m-%dT%H:%M:%SZ")

# --- Body builder (includes [event] BBCode) ---
def build_body(vevent, site_tz, rsvp=False):
    title = str(vevent.get('summary', 'Untitled')).strip() or "Untitled"
    desc = str(vevent.get('description', '')).strip()
    url = str(vevent.get('url', '')).strip()
    location = str(vevent.get('location', '')).strip()

    allday = _is_all_day(vevent)

    dtstart_raw = vevent.decoded('dtstart')
    dtend_raw = vevent.decoded('dtend', None)

    start_dt = _as_dt(dtstart_raw, site_tz)
    if dtend_raw is None:
        dtend_raw = (start_dt + (timedelta(days=1) if allday else timedelta(hours=1)))
    end_dt = _as_dt(dtend_raw, site_tz)

    if allday:
        start_attr = start_dt.strftime("%Y-%m-%d")
        if (end_dt - start_dt) >= timedelta(days=1):
            end_attr = (end_dt - timedelta(days=1)).strftime("%Y-%m-%d")
        else:
            end_attr = start_attr
        event_open = f'[event status="{"public" if rsvp else "standalone"}" timezone="{site_tz}" start="{start_attr}" end="{end_attr}"'
    else:
        event_open = f'[event status="{"public" if rsvp else "standalone"}" timezone="{site_tz}" start="{_fmt_iso_z(start_dt)}" end="{_fmt_iso_z(end_dt)}"'

    if location:
        event_open += f' location="{location}"'
    if url:
        event_open += f' url="{url}"'
    event_open += ' minimal="true"]'

    lines = [event_open, title, '[/event]']
    if desc:
        lines += ["", "---", "", desc]

    body = "\n".join(lines).strip()
    return title, body

# --- Marker utilities (track last auto-title inside the post) ---
MARKER_RE = re.compile(r'<!--\s*ics-sync:title="(.*?)"\s*-->')

def add_marker(body, auto_title):
    marker = f'\n\n<!-- ics-sync:title="{auto_title}" -->'
    return (body + marker).strip()

def strip_marker(text):
    return MARKER_RE.sub("", text or "").strip()

def extract_marker_title(text):
    m = MARKER_RE.search(text or "")
    return m.group(1) if m else None

# --- Discourse lookups & mutations ---
def find_topic_by_uid_tag(s, uid_tag):
    r = s.get(f"{BASE}/tags/{uid_tag}.json")
    if r.status_code == 404:
        return None
    r.raise_for_status()
    data = r.json()
    topics = data.get("topic_list", {}).get("topics", [])
    if not topics:
        return None
    return topics[0]["id"]  # newest topic with that tag

def read_topic(s, topic_id):
    r = s.get(f"{BASE}/t/{topic_id}.json")
    r.raise_for_status()
    return r.json()

def create_topic(s, title, raw, category_id, tags):
    payload = {
        "title": title,
        "raw": raw,
        "category": int(category_id) if category_id else None,
        "tags[]": tags or []
    }
    r = s.post(f"{BASE}/posts.json", json=payload)
    r.raise_for_status()
    data = r.json()
    return data["topic_id"], data["id"]  # topic_id, post_id

def update_topic_title_tags(s, topic_id, title=None, tags=None):
    payload = {}
    if title is not None:
        payload["title"] = title
    if tags is not None:
        payload["tags"] = tags
    if not payload:
        return
    r = s.put(f"{BASE}/t/-/{topic_id}.json", json=payload)
    r.raise_for_status()

def update_first_post(s, post_id, new_raw, reason="ICS sync update"):
    r = s.put(f"{BASE}/posts/{post_id}.json", json={"raw": new_raw, "edit_reason": reason})
    r.raise_for_status()

# --- Main processing ---
def process_vevent(s, vevent, args):
    uid = str(vevent.get('uid', '')).strip()
    if not uid:
        log.warning("Skipping event without UID")
        return

    uid_tag = _sanitize_tag(f"uid-{uid}")
    extra_tags = [t for t in (args.tags or []) if t]
    tags = list(dict.fromkeys(DEFAULT_TAGS + extra_tags + [uid_tag]))  # unique, stable order

    if args.future_only:
        now = datetime.now(gettz(SITE_TZ))
        dtstart = _as_dt(vevent.decoded('dtstart'), SITE_TZ)
        if dtstart < now - timedelta(hours=1):
            return

    auto_title, fresh_body_no_marker = build_body(vevent, SITE_TZ, rsvp=args.rsvp)
    fresh_body = add_marker(fresh_body_no_marker, auto_title)

    topic_id = find_topic_by_uid_tag(s, uid_tag)

    if topic_id is None:
        if args.dry_run:
            log.info(f"[DRY] CREATE: {auto_title}  tags={tags}")
            return
        log.info(f"Creating new topic for UID {uid} …")
        created_topic_id, first_post_id = create_topic(s, auto_title, fresh_body, CATEGORY_ID, tags)
        log.info(f"Created topic #{created_topic_id}")
        return

    # Update path
    topic = read_topic(s, topic_id)
    first_post = topic["post_stream"]["posts"][0]
    first_post_id = first_post["id"]
    old_raw = first_post["raw"]
    old_title_visible = topic["title"]
    old_marker_title = extract_marker_title(old_raw)  # last auto-title we wrote

    # Compare bodies while ignoring the marker so it doesn't cause churn
    old_raw_stripped = strip_marker(old_raw)
    need_post_update = (old_raw_stripped.strip() != fresh_body_no_marker.strip())

    # Title update policy:
    # - If we previously set the title (visible title == stored marker title),
    #   then it's safe to update it to the new auto_title.
    # - Otherwise, PRESERVE the current title (assume a human changed it).
    can_update_title = (old_marker_title is not None and old_title_visible.strip() == old_marker_title.strip())
    need_title_update = (can_update_title and old_title_visible.strip() != auto_title.strip())

    # Tags can always be normalized/updated; category is NEVER changed here.
    old_tags = topic.get("tags", [])
    need_tags_update = (sorted(old_tags) != sorted(tags))

    if not (need_post_update or need_title_update or need_tags_update):
        log.info(f"No changes for UID {uid} (topic #{topic_id})")
        return

    if args.dry_run:
        what = []
        if need_post_update: what.append("post")
        if need_title_update: what.append("title")
        if need_tags_update: what.append("tags")
        log.info(f"[DRY] UPDATE ({', '.join(what)}): topic #{topic_id} -> {auto_title} tags={tags}")
        return

    log.info(f"Updating topic #{topic_id} for UID {uid} …")
    if need_post_update:
        update_first_post(s, first_post_id, fresh_body, reason="ICS sync update")
    if need_title_update or need_tags_update:
        update_topic_title_tags(s, topic_id,
                                title=(auto_title if need_title_update else None),
                                tags=(tags if need_tags_update else None))
    log.info(f"Updated topic #{topic_id}")

def main():
    ap = argparse.ArgumentParser(description="Sync ICS feed into Discourse topics (create/update by UID). Preserves human titles; never moves categories.")
    ap.add_argument("--ics-url", help="URL to ICS feed")
    ap.add_argument("--ics-file", help="Path to local .ics")
    ap.add_argument("--future-only", action="store_true", help="Only import future events")
    ap.add_argument("--rsvp", action="store_true", help="Use status=\"public\" (RSVP) instead of standalone")
    ap.add_argument("--dry-run", action="store_true", help="Print actions without calling the API")
    ap.add_argument("--skip-errors", action="store_true", help="Continue on event errors")
    ap.add_argument("--tags", help="Comma-separated extra tags to add", default="")
    args = ap.parse_args()
    args.tags = [t.strip() for t in (args.tags.split(",") if args.tags else []) if t.strip()]

    for var in ("DISCOURSE_BASE_URL", "DISCOURSE_API_KEY", "DISCOURSE_API_USERNAME"):
        if not os.environ.get(var):
            log.error(f"Missing env: {var}")
            sys.exit(1)

    if not args.ics_url and not args.ics_file:
        log.error("Provide --ics-url or --ics-file")
        sys.exit(1)

    if args.ics_url:
        import urllib.request
        log.info(f"Fetching ICS: {args.ics_url}")
        with urllib.request.urlopen(args.ics_url) as resp:
            data = resp.read()
    else:
        with open(args.ics_file, "rb") as f:
            data = f.read()

    cal = Calendar.from_ical(data)
    s = _session()

    for comp in cal.walk("VEVENT"):
        try:
            process_vevent(s, comp, args)
        except Exception as e:
            if args.skip_errors:
                log.error(f"Error on event UID={comp.get('uid')}: {e}")
                continue
            raise

if __name__ == "__main__":
    main()

Happy to adjust further if you think there’s a cleaner approach — my aim is to keep it reliable for different Discourse setups.

1 like