311 lines
8.2 KiB
Python
Executable File
311 lines
8.2 KiB
Python
Executable File
#!/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()
|