HDValentin/OSMcal_Mobilizon_Import
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial OSMCal to Mobilizon importer

Browse Source
This commit is contained in:
Valentin Bachem 2026-05-01 22:52:56 +02:00
commit cbadfc4c7c
No known key found for this signature in database
4 changed files with 1101 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

664
import-osmcal-to-mobilizon.py Executable file
View File

@ -0,0 +1,664 @@
#!/usr/bin/env python3
"""
Importiert OSMCal-Veranstaltungen im angegebenen Umkreis um Heidelberg nach Mobilizon.
Datenquellen:
- OSMCal API:
https://osmcal.org/api/v2/events/?in=de
Liefert Titel, Datum, Ort, Koordinaten und Event-URL.
- OSMCal Einzel-ICS:
https://osmcal.org/event/<ID>.ics
Liefert die Originalbeschreibung über DESCRIPTION.
Ziel:
- Mobilizon GraphQL API:
https://rheinneckar.events/api
Hinweise:
- Das Script importiert nur neue Events.
- Bereits importierte Events werden in imported-osmcal-events.json gespeichert.
- Bestehende Mobilizon-Entwürfe werden NICHT aktualisiert oder veröffentlicht.
- Veranstaltungsende wird immer auf Startzeit + 3 Stunden gesetzt.
Voraussetzungen:
vor Ausführung Api-Schlüssel in die .env Datei speichern
Ausführen:
source .venv/bin/activate
pip install requests python-dotenv icalendar
"""
import os
import re
import json
import html
import hashlib
from datetime import datetime, timedelta
from math import radians, sin, cos, sqrt, atan2
from pathlib import Path
import requests
from dotenv import load_dotenv
from icalendar import Calendar
# ------------------------------------------------------------
# .env laden
# ------------------------------------------------------------
SCRIPT_DIR = Path(__file__).resolve().parent
load_dotenv(SCRIPT_DIR / ".env")
# ------------------------------------------------------------
# Konfiguration
# ------------------------------------------------------------
# Zentrum: Heidelberg
CENTER_LAT = 49.4094
CENTER_LON = 8.6947
# Radius in Kilometern
RADIUS_KM = 18.0
# True = als Entwurf importieren
# False = direkt veröffentlichen
DRAFT = False
# Feste Dauer je Veranstaltung
EVENT_DURATION_HOURS = 3
# OSMCal API für kommende Veranstaltungen in Deutschland
OSMCAL_API = "https://osmcal.org/api/v2/events/?in=de"
# Mobilizon-Instanz aus .env oder Standardwert
MOBILIZON_INSTANCE = os.environ.get("MOBILIZON_INSTANCE", "https://rheinneckar.events")
MOBILIZON_API = f"{MOBILIZON_INSTANCE.rstrip('/')}/api"
# Diese Werte müssen in .env stehen
MOBILIZON_TOKEN = os.environ["MOBILIZON_TOKEN"]
ORGANIZER_ACTOR_ID = os.environ["MOBILIZON_ORGANIZER_ACTOR_ID"]
ATTRIBUTED_TO_ID = os.environ["MOBILIZON_ATTRIBUTED_TO_ID"]
# Datei zum Vermeiden von Duplikaten, hier werden bearbeitete Veranstaltungen gespeichert.
# vorher leeren wenn alle Veranstaltungen neu importiert werden sollen, mit
# echo '{}' > imported-osmcal-events.json
STATE_FILE = SCRIPT_DIR / "imported-osmcal-events.json"
# ------------------------------------------------------------
# Mobilizon GraphQL Mutation
# ------------------------------------------------------------
CREATE_EVENT_MUTATION = """
mutation createEvent(
$organizerActorId: ID!
$title: String!
$attributedToId: ID
$description: String!
$beginsOn: DateTime!
$endsOn: DateTime
$status: EventStatus
$visibility: EventVisibility
$joinOptions: EventJoinOptions
$draft: Boolean
$tags: [String]
$onlineAddress: String
$physicalAddress: AddressInput
$options: EventOptionsInput
) {
createEvent(
organizerActorId: $organizerActorId
attributedToId: $attributedToId
title: $title
description: $description
beginsOn: $beginsOn
endsOn: $endsOn
status: $status
visibility: $visibility
joinOptions: $joinOptions
draft: $draft
tags: $tags
onlineAddress: $onlineAddress
physicalAddress: $physicalAddress
options: $options
) {
id
uuid
title
url
}
}
"""
# ------------------------------------------------------------
# Allgemeine Hilfsfunktionen
# ------------------------------------------------------------
def clean_text(value):
"""
Wandelt Werte in saubere Strings um.
Gibt None zurück, wenn der Wert leer ist.
"""
if value is None:
return None
value = str(value).strip()
if not value:
return None
return value
def add_hours_to_iso_datetime(iso_datetime, hours):
"""
Addiert Stunden auf eine ISO-Datetime-Zeichenkette.
Beispiel:
2026-05-18T19:00:00+02:00
-> 2026-05-18T22:00:00+02:00
"""
dt = datetime.fromisoformat(iso_datetime)
return (dt + timedelta(hours=hours)).isoformat()
def distance_km(lat1, lon1, lat2, lon2):
"""
Berechnet die Distanz zwischen zwei Koordinaten per Haversine-Formel.
Rückgabe: Kilometer.
"""
earth_radius_km = 6371.0
dlat = radians(lat2 - lat1)
dlon = radians(lon2 - lon1)
a = (
sin(dlat / 2) ** 2
+ cos(radians(lat1))
* cos(radians(lat2))
* sin(dlon / 2) ** 2
)
return 2 * earth_radius_km * atan2(sqrt(a), sqrt(1 - a))
def extract_osmcal_event_id(event_url):
"""
Extrahiert die numerische OSMCal-ID aus URLs wie:
https://osmcal.org/event/4742/
"""
if not event_url:
return None
match = re.search(r"/event/(\d+)/?", event_url)
if not match:
return None
return match.group(1)
def osmcal_event_ics_url(event_url):
"""
Wandelt eine OSMCal-Web-URL in die Einzel-ICS-URL um.
Beispiel:
https://osmcal.org/event/4742/
-> https://osmcal.org/event/4742.ics
"""
event_id = extract_osmcal_event_id(event_url)
if not event_id:
return None
return f"https://osmcal.org/event/{event_id}.ics"
# ------------------------------------------------------------
# Statusdatei
# ------------------------------------------------------------
def save_state(state):
"""
Speichert die Liste bereits importierter Events.
"""
STATE_FILE.write_text(
json.dumps(state, indent=2, ensure_ascii=False),
encoding="utf-8",
)
def load_state():
"""
Lädt die lokale Import-Statusdatei.
Verhalten:
- Datei fehlt: wird mit {} angelegt.
- Datei ist leer: wird mit {} ersetzt.
- Datei ist kaputt: wird gesichert und neu mit {} angelegt.
"""
if not STATE_FILE.exists():
save_state({})
return {}
content = STATE_FILE.read_text(encoding="utf-8").strip()
if not content:
save_state({})
return {}
try:
return json.loads(content)
except json.JSONDecodeError:
backup_file = STATE_FILE.with_suffix(".json.broken")
STATE_FILE.rename(backup_file)
print(f"Warnung: {STATE_FILE.name} war kein gültiges JSON.")
print(f"Defekte Datei wurde gesichert als: {backup_file.name}")
print(f"{STATE_FILE.name} wird neu mit {{}} angelegt.")
save_state({})
return {}
def osmcal_event_key(event):
"""
Erzeugt einen stabilen Schlüssel für die Duplikatvermeidung.
Primär wird die OSMCal-URL genutzt.
Falls sie fehlt, wird ein Hash aus Titel, Startzeit und Ort gebildet.
"""
if event.get("url"):
return event["url"]
raw = "|".join(
[
str(event.get("name", "")),
str(event.get("date", {}).get("start", "")),
str(event.get("location", {}).get("short", "")),
]
)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
# ------------------------------------------------------------
# OSMCal lesen
# ------------------------------------------------------------
def fetch_osmcal_events():
"""
Holt kommende OSMCal-Events für Deutschland.
"""
response = requests.get(
OSMCAL_API,
headers={
"Accept-Language": "de",
"Client-App": "rheinneckar-events-import/0.6",
},
timeout=30,
)
response.raise_for_status()
events = response.json()
if not isinstance(events, list):
raise RuntimeError("OSMCal-Antwort ist keine Event-Liste.")
return events
def is_near_heidelberg(event):
"""
Prüft, ob ein Event im konfigurierten Radius um Heidelberg liegt.
OSMCal liefert coords als [lon, lat].
"""
location = event.get("location") or {}
coords = location.get("coords")
if not coords or len(coords) != 2:
return False
try:
lon = float(coords[0])
lat = float(coords[1])
except (TypeError, ValueError):
return False
distance = distance_km(CENTER_LAT, CENTER_LON, lat, lon)
return distance <= RADIUS_KM
def event_distance(event):
"""
Gibt die Entfernung eines Events zu Heidelberg zurück.
Rückgabe: gerundete Kilometer oder None.
"""
location = event.get("location") or {}
coords = location.get("coords")
if not coords or len(coords) != 2:
return None
try:
lon = float(coords[0])
lat = float(coords[1])
except (TypeError, ValueError):
return None
return round(distance_km(CENTER_LAT, CENTER_LON, lat, lon), 1)
def fetch_osmcal_ics_description(event_url):
"""
Holt die Originalbeschreibung aus der Einzel-ICS-Datei.
OSMCal-API liefert in der Eventliste keine Beschreibung.
Die Einzel-ICS enthält aber normalerweise DESCRIPTION.
"""
ics_url = osmcal_event_ics_url(event_url)
if not ics_url:
return None
response = requests.get(
ics_url,
headers={
"Accept-Language": "de",
"Client-App": "rheinneckar-events-import/0.6",
},
timeout=30,
)
response.raise_for_status()
calendar = Calendar.from_ical(response.content)
for component in calendar.walk():
if component.name != "VEVENT":
continue
description = component.get("DESCRIPTION")
if description:
return str(description).strip()
return None
# ------------------------------------------------------------
# Umwandlung OSMCal -> Mobilizon
# ------------------------------------------------------------
def mobilizon_address_from_osmcal(event):
"""
Wandelt OSMCal-Ortsdaten in Mobilizon AddressInput um.
Wichtig:
physicalAddress.description ist ein String, keine Liste.
"""
location = event.get("location") or {}
coords = location.get("coords")
if not coords or len(coords) != 2:
return None
try:
lon = float(coords[0])
lat = float(coords[1])
except (TypeError, ValueError):
return None
venue = clean_text(location.get("venue"))
short = clean_text(location.get("short"))
detailed = clean_text(location.get("detailed"))
description = venue or short or detailed or "Ort"
address = {
"description": description,
"geom": f"{lon};{lat}",
"country": "DE",
}
if short:
address["locality"] = short
if detailed:
address["street"] = detailed
return address
def mobilizon_description_from_osmcal(event):
"""
Baut die Mobilizon-Beschreibung.
Reihenfolge:
1. Originalbeschreibung aus der OSMCal-ICS
2. Trennlinie
3. Importhinweis und Metadaten
Links werden als HTML erzeugt, nicht als Markdown.
"""
name = clean_text(event.get("name")) or "OSMCal-Veranstaltung"
escaped_name = html.escape(name)
url = clean_text(event.get("url"))
escaped_url = html.escape(url) if url else None
location = event.get("location") or {}
location_text = clean_text(
location.get("venue")
or location.get("detailed")
or location.get("short")
)
distance = event_distance(event)
original_description = clean_text(
event.get("description")
or event.get("details")
or event.get("body")
or event.get("text")
or event.get("summary")
or event.get("content")
)
if not original_description and url:
try:
original_description = fetch_osmcal_ics_description(url)
except Exception as exc:
print(f"Warnung: Beschreibung konnte nicht aus ICS gelesen werden: {url} - {exc}")
parts = []
if original_description:
escaped_description = html.escape(original_description).replace("\n", "<br>")
parts.append(f"<p>{escaped_description}</p>")
parts.append("<hr>")
parts.append("<p>Diese Veranstaltung wurde aus dem OpenStreetMap Calendar importiert.</p>")
if escaped_url:
parts.append(
f'<p><strong>Quelle:</strong> <a href="{escaped_url}">{escaped_url}</a></p>'
)
if location_text:
parts.append(
f"<p><strong>Ort:</strong> {html.escape(location_text)}</p>"
)
if distance is not None:
parts.append(
f"<p><strong>Entfernung von Heidelberg:</strong> {distance} km</p>"
)
parts.append(f"<p><em>Originaltitel:</em> {escaped_name}</p>")
return "\n".join(parts)
# ------------------------------------------------------------
# Mobilizon schreiben
# ------------------------------------------------------------
def create_mobilizon_event(event):
"""
Legt ein Event in Mobilizon an.
Gruppenposting:
- organizerActorId = Profil-ID
- attributedToId = Gruppen-ID
Ende:
- OSMCal-Endzeit wird ignoriert.
- Mobilizon-Endzeit wird immer auf Startzeit + EVENT_DURATION_HOURS gesetzt.
"""
title = clean_text(event.get("name")) or "OSMCal-Veranstaltung"
date = event.get("date") or {}
begins_on = clean_text(date.get("start"))
if not begins_on:
raise ValueError(f"Event ohne Startzeit: {json.dumps(event, ensure_ascii=False)}")
ends_on = add_hours_to_iso_datetime(begins_on, EVENT_DURATION_HOURS)
variables = {
"organizerActorId": ORGANIZER_ACTOR_ID,
"attributedToId": ATTRIBUTED_TO_ID,
"title": title,
"description": mobilizon_description_from_osmcal(event),
"beginsOn": begins_on,
"endsOn": ends_on,
"status": "CONFIRMED",
"visibility": "PUBLIC",
"joinOptions": "FREE",
"draft": DRAFT,
"tags": ["osm", "openstreetmap", "rheinneckar"],
"onlineAddress": clean_text(event.get("url")),
"physicalAddress": mobilizon_address_from_osmcal(event),
"options": {
"commentModeration": "CLOSED",
"showStartTime": True,
"showEndTime": True,
},
}
# None-Werte entfernen, damit Mobilizon keine unnötigen Nullfelder bekommt.
variables = {
key: value
for key, value in variables.items()
if value is not None
}
response = requests.post(
MOBILIZON_API,
headers={
"Authorization": f"Bearer {MOBILIZON_TOKEN}",
"Content-Type": "application/json",
},
json={
"query": CREATE_EVENT_MUTATION,
"variables": variables,
},
timeout=30,
)
try:
payload = response.json()
except Exception:
print("Mobilizon hat keine JSON-Antwort geliefert:")
print(response.text)
raise
if response.status_code >= 400 or payload.get("errors"):
print()
print("Mobilizon-Fehler beim Import dieses Events:")
print(json.dumps(event, indent=2, ensure_ascii=False))
print()
print("Gesendete Variablen:")
print(json.dumps(variables, indent=2, ensure_ascii=False))
print()
print("Antwort von Mobilizon:")
print(json.dumps(payload, indent=2, ensure_ascii=False))
raise RuntimeError("Mobilizon-Import fehlgeschlagen.")
return payload["data"]["createEvent"]
# ------------------------------------------------------------
# Hauptprogramm
# ------------------------------------------------------------
def main():
state = load_state()
events = fetch_osmcal_events()
nearby_events = []
for event in events:
if event.get("cancelled", False):
continue
if is_near_heidelberg(event):
nearby_events.append(event)
print(f"{len(events)} Veranstaltungen aus OSMCal geladen")
print(f"{len(nearby_events)} Veranstaltungen im {RADIUS_KM:g}-km-Umkreis um Heidelberg")
imported = 0
skipped = 0
for event in nearby_events:
key = osmcal_event_key(event)
if key in state:
skipped += 1
continue
created = create_mobilizon_event(event)
state[key] = {
"osmcal_url": event.get("url"),
"mobilizon_id": created.get("id"),
"mobilizon_uuid": created.get("uuid"),
"mobilizon_url": created.get("url"),
"title": event.get("name"),
"start": event.get("date", {}).get("start"),
"distance_km": event_distance(event),
"draft": DRAFT,
"duration_hours": EVENT_DURATION_HOURS,
}
save_state(state)
imported += 1
print(
"Importiert:",
event.get("name"),
"->",
created.get("url") or created.get("uuid") or created.get("id"),
)
print()
print(f"Fertig. Importiert: {imported}, übersprungen: {skipped}")
if DRAFT:
print()
print("Hinweis: DRAFT=True. Die Events wurden als Entwurf angelegt.")
print("Für Veröffentlichung im Script DRAFT = False setzen.")
if __name__ == "__main__":
main()

117
readme.md Normal file
View File

@ -0,0 +1,117 @@
# OSMCal → Mobilizon Importer
EN: Import events from [OSMCal](https://osmcal.org/) into a Mobilizon group, e.g. `https://rheinneckar.events`.
DE: Importiert Veranstaltungen aus dem OpenStreetMap Calendar nach Mobilizon, filtert sie lokal nach Entfernung um Heidelberg und legt sie als Gruppenveranstaltungen an.
---
## Features / Funktionen
- Fetches German OSMCal events via API
Lädt deutsche OSMCal-Veranstaltungen über die API
- Filters events by radius around Heidelberg
Filtert Veranstaltungen nach Radius um Heidelberg
- Reads the original description from each event’s `.ics` file
Liest die Originalbeschreibung aus der jeweiligen `.ics`-Datei
- Imports into Mobilizon via GraphQL
Importiert nach Mobilizon per GraphQL
- Supports Mobilizon group posting
Unterstützt Veröffentlichung als Mobilizon-Gruppe
- Avoids duplicate imports using a local state file
Verhindert doppelte Importe über eine lokale Statusdatei
---
## Requirements / Voraussetzungen
- Python 3
- Mobilizon account
- Membership in the target Mobilizon group
- OAuth token for the Mobilizon API
Install dependencies:
```bash
python3 -m venv .venv
source .venv/bin/activate
pip install requests python-dotenv icalendar
#auf dem eigenen Rechner voraussetzungen schaffen
#Verzeichnis erstellen und darin ausführen:
```bash
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
## Configuration / Konfiguration
Create a .env file:
MOBILIZON_INSTANCE="https://rheinneckar.events"
MOBILIZON_TOKEN="API token"
MOBILIZON_ORGANIZER_ACTOR_ID="profile actor ID"
MOBILIZON_ATTRIBUTED_TO_ID="group actor ID"
#The helper script setup-mobilizon.py can create this .env file.
## Usage / Nutzung
Run the importer:
```bash
source .venv/bin/activate
./import-osmcal-to-mobilizon.py
Example output:
70 Veranstaltungen aus OSMCal geladen
9 Veranstaltungen im 15-km-Umkreis um Heidelberg
Fertig. Importiert: 9, übersprungen: 0
## Settings / Einstellungen
Edit these values in import-osmcal-to-mobilizon.py.
Diese Werte in dieser Datei anpassen
Radius / Radius um die angegebenen Koordinaten
RADIUS_KM = 18.0
Draft mode / Entwurfsmodus
DRAFT = True
Creates drafts / Veranstaltungen werden als Entwurf veröffentlicht.
DRAFT = False
Publishes directly / Veranstaltungen werden direkt veröffentlicht.
Event duration / Veranstaltungsdauer
EVENT_DURATION_HOURS = 3
## Duplicate handling / Umgang mit bereits importierten Veranstaltungen
Imported events are stored in:
imported-osmcal-events.json
Veranstaltungen in dieser Datei werden bei einem erneuten Lauf des Skriptes übersprungen.
Events listed there are skipped on future runs.
Reset for a new test / Zurücksetzen vor einen neuen Lauf:
```bash
echo '{}' > imported-osmcal-events.json
Vorher alte Entwürfe oder Events in Mobilizon löschen, sonst entstehen Duplikate.
Delete old Mobilizon drafts or events first, otherwise duplicates may be created.
Notes / Hinweise
DE:
Bestehende Mobilizon-Events werden nicht aktualisiert.
Bestehende Entwürfe werden nicht automatisch veröffentlicht.
Wenn DRAFT von True auf False geändert wird, werden bereits importierte Events trotzdem übersprungen.
Der Access-Token kann ablaufen und muss dann erneuert werden.
EN:
Existing Mobilizon events are not updated.
Existing drafts are not automatically published.
If DRAFT is changed from True to False, already imported events are still skipped.
The access token may expire and then has to be renewed.

10
requirements.txt Normal file
View File

@ -0,0 +1,10 @@
certifi==2026.4.22
charset-normalizer==3.4.7
icalendar==7.1.0
idna==3.13
python-dateutil==2.9.0.post0
python-dotenv==1.2.2
requests==2.33.1
six==1.17.0
tzdata==2026.2
urllib3==2.6.3

310
setup-mobilizon.py Executable file
View File

@ -0,0 +1,310 @@
#!/usr/bin/env python3
import json
import time
from pathlib import Path
import requests
INSTANCE = "https://rheinneckar.events"
# Wichtig:
# "read" wird benötigt, damit loggedUser.memberships gelesen werden darf.
# Die write:event:* Scopes werden später für das Erstellen/Aktualisieren/Löschen von Events genutzt.
SCOPES = "read write:event:create write:event:update write:event:delete"
APP_FILE = Path("mobilizon-app.json")
TOKEN_FILE = Path("mobilizon-token.json")
ENV_FILE = Path(".env")
def post_form(url, data):
response = requests.post(url, data=data, timeout=30)
print(f"POST {url} -> HTTP {response.status_code}")
try:
payload = response.json()
except Exception:
print("Keine JSON-Antwort:")
print(response.text)
raise SystemExit(1)
if response.status_code >= 400:
print("Fehlerantwort:")
print(json.dumps(payload, indent=2, ensure_ascii=False))
raise SystemExit(1)
return payload
def graphql(access_token, query, variables=None):
response = requests.post(
f"{INSTANCE}/api",
headers={
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
},
json={
"query": query,
"variables": variables or {},
},
timeout=30,
)
print(f"GraphQL -> HTTP {response.status_code}")
try:
payload = response.json()
except Exception:
print("Keine JSON-Antwort:")
print(response.text)
raise SystemExit(1)
if response.status_code >= 400 or payload.get("errors"):
print("GraphQL-Fehler:")
print(json.dumps(payload, indent=2, ensure_ascii=False))
raise SystemExit(1)
return payload["data"]
def register_app():
print("1. App wird registriert...")
app = post_form(
f"{INSTANCE}/apps",
{
"name": "OSMCal Importer",
"redirect_uri": "urn:ietf:wg:oauth:2.0:oob",
"website": INSTANCE,
"scope": SCOPES,
},
)
APP_FILE.write_text(
json.dumps(app, indent=2, ensure_ascii=False),
encoding="utf-8",
)
return app
def request_device_code(client_id):
print()
print("2. Login-Code wird angefordert...")
device = post_form(
f"{INSTANCE}/login/device/code",
{
"client_id": client_id,
"scope": SCOPES,
},
)
print()
print("Öffne diese Adresse im Browser:")
print(device["verification_uri"])
print()
print("Gib dort diesen Code ein:")
print(device["user_code"])
print()
print("Danach hier warten. Das Script fragt automatisch ab, ob du bestätigt hast.")
print()
return device
def wait_for_token(client_id, device):
interval = int(device.get("interval", 5))
expires_in = int(device.get("expires_in", 900))
deadline = time.time() + expires_in
while time.time() < deadline:
time.sleep(interval)
response = requests.post(
f"{INSTANCE}/oauth/token",
data={
"client_id": client_id,
"device_code": device["device_code"],
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
},
timeout=30,
)
try:
payload = response.json()
except Exception:
print("Keine JSON-Antwort beim Token-Abruf:")
print(response.text)
raise SystemExit(1)
if "access_token" in payload:
print("Token erhalten.")
TOKEN_FILE.write_text(
json.dumps(payload, indent=2, ensure_ascii=False),
encoding="utf-8",
)
return payload
error = payload.get("error")
if error == "authorization_pending":
print("Noch nicht bestätigt...")
continue
if error == "slow_down":
print("Server bittet um langsameres Abfragen...")
interval += 5
continue
print("Token-Abruf fehlgeschlagen:")
print(json.dumps(payload, indent=2, ensure_ascii=False))
raise SystemExit(1)
print("Der Login-Code ist abgelaufen. Script neu starten.")
raise SystemExit(1)
def get_memberships(access_token):
print()
print("3. Gruppenmitgliedschaften werden abgefragt...")
query = """
query LoggedUserMemberships($page: Int, $limit: Int) {
loggedUser {
id
memberships(page: $page, limit: $limit) {
total
elements {
role
actor {
id
type
preferredUsername
name
}
parent {
id
type
preferredUsername
name
}
}
}
}
}
"""
data = graphql(
access_token,
query,
variables={
"page": 1,
"limit": 100,
},
)
memberships_data = data.get("loggedUser", {}).get("memberships")
if not memberships_data:
print()
print("Keine Gruppenmitgliedschaften gefunden.")
print("Mögliche Ursachen:")
print("- Du bist mit diesem Account in keiner Gruppe Mitglied.")
print("- Die App wurde nicht mit dem Scope 'read' autorisiert.")
print("- Der verwendete Account ist nicht der erwartete Mobilizon-Account.")
raise SystemExit(1)
memberships = memberships_data.get("elements", [])
if not memberships:
print()
print("Du bist mit diesem Account in keiner Gruppe Mitglied.")
raise SystemExit(1)
return memberships
def choose_membership(memberships):
print()
print("Deine Gruppen:")
print()
for index, membership in enumerate(memberships, start=1):
profile = membership["actor"]
group = membership["parent"]
role = membership["role"]
profile_name = profile.get("name") or profile.get("preferredUsername") or "Profil"
group_name = group.get("name") or group.get("preferredUsername") or "Gruppe"
print(
f"{index}. Gruppe: {group_name} "
f"(group_id={group['id']}) | "
f"Profil: {profile_name} "
f"(profile_id={profile['id']}) | "
f"Rolle: {role}"
)
print()
choice = input("Welche Gruppe soll für den Import verwendet werden? Nummer eingeben: ").strip()
try:
selected = memberships[int(choice) - 1]
except Exception:
print("Ungültige Auswahl.")
raise SystemExit(1)
role = selected["role"]
if role not in ("MEMBER", "MODERATOR", "ADMINISTRATOR"):
print()
print(f"Warnung: Deine Rolle ist {role}.")
print("Falls der Import später fehlschlägt, darf dieses Profil vermutlich keine Gruppen-Events veröffentlichen.")
return selected
def write_env(app, token, membership):
organizer_actor_id = membership["actor"]["id"]
attributed_to_id = membership["parent"]["id"]
lines = [
f'MOBILIZON_INSTANCE="{INSTANCE}"',
f'MOBILIZON_CLIENT_ID="{app["client_id"]}"',
f'MOBILIZON_CLIENT_SECRET="{app.get("client_secret", "")}"',
f'MOBILIZON_TOKEN="{token["access_token"]}"',
f'MOBILIZON_REFRESH_TOKEN="{token.get("refresh_token", "")}"',
f'MOBILIZON_ORGANIZER_ACTOR_ID="{organizer_actor_id}"',
f'MOBILIZON_ATTRIBUTED_TO_ID="{attributed_to_id}"',
]
ENV_FILE.write_text("\n".join(lines) + "\n", encoding="utf-8")
ENV_FILE.chmod(0o600)
print()
print("Fertig. .env wurde geschrieben.")
print()
print("Gesetzte Werte:")
print(f"MOBILIZON_ORGANIZER_ACTOR_ID={organizer_actor_id}")
print(f"MOBILIZON_ATTRIBUTED_TO_ID={attributed_to_id}")
print()
print("Hinweis: Der Access-Token läuft nach einigen Stunden ab.")
print("Für erste Tests reicht das. Für Cron-Betrieb sollte später Refresh-Token-Unterstützung ergänzt werden.")
def main():
app = register_app()
device = request_device_code(app["client_id"])
token = wait_for_token(app["client_id"], device)
memberships = get_memberships(token["access_token"])
selected = choose_membership(memberships)
write_env(app, token, selected)
if __name__ == "__main__":
main()