Mercurial > libervia-backend
view libervia/backend/plugins/plugin_comp_ap_gateway/events.py @ 4163:3b3cd9453d9b
plugin XEP-0308: implement Last Message Correction
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 28 Nov 2023 17:38:31 +0100 |
parents | 4b842c1fb686 |
children | 49019947cc76 |
line wrap: on
line source
#!/usr/bin/env python3 # Libervia ActivityPub Gateway # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. from typing import Tuple import mimetypes import html import shortuuid from twisted.words.xish import domish from twisted.words.protocols.jabber import jid from libervia.backend.core.i18n import _ from libervia.backend.core.log import getLogger from libervia.backend.core import exceptions from libervia.backend.tools.common import date_utils, uri from .constants import NS_AP_PUBLIC, TYPE_ACTOR, TYPE_EVENT, TYPE_ITEM log = getLogger(__name__) # direct copy of what Mobilizon uses AP_EVENTS_CONTEXT = { "@language": "und", "Hashtag": "as:Hashtag", "PostalAddress": "sc:PostalAddress", "PropertyValue": "sc:PropertyValue", "address": {"@id": "sc:address", "@type": "sc:PostalAddress"}, "addressCountry": "sc:addressCountry", "addressLocality": "sc:addressLocality", "addressRegion": "sc:addressRegion", "anonymousParticipationEnabled": {"@id": "mz:anonymousParticipationEnabled", "@type": "sc:Boolean"}, "category": "sc:category", "commentsEnabled": {"@id": "pt:commentsEnabled", "@type": "sc:Boolean"}, "discoverable": "toot:discoverable", "discussions": {"@id": "mz:discussions", "@type": "@id"}, "events": {"@id": "mz:events", "@type": "@id"}, "ical": "http://www.w3.org/2002/12/cal/ical#", "inLanguage": "sc:inLanguage", "isOnline": {"@id": "mz:isOnline", "@type": "sc:Boolean"}, "joinMode": {"@id": "mz:joinMode", "@type": "mz:joinModeType"}, "joinModeType": {"@id": "mz:joinModeType", "@type": "rdfs:Class"}, "location": {"@id": "sc:location", "@type": "sc:Place"}, "manuallyApprovesFollowers": "as:manuallyApprovesFollowers", "maximumAttendeeCapacity": "sc:maximumAttendeeCapacity", "memberCount": {"@id": "mz:memberCount", "@type": "sc:Integer"}, "members": {"@id": "mz:members", "@type": "@id"}, "mz": "https://joinmobilizon.org/ns#", "openness": {"@id": "mz:openness", "@type": "@id"}, "participantCount": {"@id": "mz:participantCount", "@type": "sc:Integer"}, "participationMessage": {"@id": "mz:participationMessage", "@type": "sc:Text"}, "postalCode": "sc:postalCode", "posts": {"@id": "mz:posts", "@type": "@id"}, "propertyID": "sc:propertyID", "pt": "https://joinpeertube.org/ns#", "remainingAttendeeCapacity": "sc:remainingAttendeeCapacity", "repliesModerationOption": {"@id": "mz:repliesModerationOption", "@type": "mz:repliesModerationOptionType"}, "repliesModerationOptionType": {"@id": "mz:repliesModerationOptionType", "@type": "rdfs:Class"}, "resources": {"@id": "mz:resources", "@type": "@id"}, "sc": "http://schema.org#", "streetAddress": "sc:streetAddress", "timezone": {"@id": "mz:timezone", "@type": "sc:Text"}, "todos": {"@id": "mz:todos", "@type": "@id"}, "toot": "http://joinmastodon.org/ns#", "uuid": "sc:identifier", "value": "sc:value" } class APEvents: """XMPP Events <=> AP Events conversion""" def __init__(self, apg): self.host = apg.host self.apg = apg self._events = self.host.plugins["XEP-0471"] async def event_data_2_ap_item( self, event_data: dict, author_jid: jid.JID, is_new: bool=True ) -> dict: """Convert event data to AP activity @param event_data: event data as used in [plugin_exp_events] @param author_jid: jid of the published of the event @param is_new: if True, the item is a new one (no instance has been found in cache). If True, a "Create" activity will be generated, otherwise an "Update" one will be @return: AP activity wrapping an Event object """ if not event_data.get("id"): event_data["id"] = shortuuid.uuid() ap_account = await self.apg.get_ap_account_from_jid_and_node( author_jid, self._events.namespace ) url_actor = self.apg.build_apurl(TYPE_ACTOR, ap_account) url_item = self.apg.build_apurl(TYPE_ITEM, ap_account, event_data["id"]) ap_object = { "actor": url_actor, "attributedTo": url_actor, "to": [NS_AP_PUBLIC], "id": url_item, "type": TYPE_EVENT, "name": next(iter(event_data["name"].values())), "startTime": date_utils.date_fmt(event_data["start"], "iso"), "endTime": date_utils.date_fmt(event_data["end"], "iso"), "url": url_item, } attachment = ap_object["attachment"] = [] # FIXME: we only handle URL head-picture for now # TODO: handle jingle and use file metadata try: head_picture_url = event_data["head-picture"]["sources"][0]["url"] except (KeyError, IndexError, TypeError): pass else: media_type = mimetypes.guess_type(head_picture_url, False)[0] or "image/jpeg" attachment.append({ "name": "Banner", "type": "Document", "mediaType": media_type, "url": head_picture_url, }) descriptions = event_data.get("descriptions") if descriptions: for description in descriptions: content = description["description"] if description["type"] == "xhtml": break else: content = f"<p>{html.escape(content)}</p>" # type: ignore ap_object["content"] = content categories = event_data.get("categories") if categories: tag = ap_object["tag"] = [] for category in categories: tag.append({ "name": f"#{category['term']}", "type": "Hashtag", }) locations = event_data.get("locations") if locations: ap_loc = ap_object["location"] = {} # we only use the first found location location = locations[0] for source, dest in ( ("description", "name"), ("lat", "latitude"), ("lon", "longitude"), ): value = location.get(source) if value is not None: ap_loc[dest] = value for source, dest in ( ("country", "addressCountry"), ("locality", "addressLocality"), ("region", "addressRegion"), ("postalcode", "postalCode"), ("street", "streetAddress"), ): value = location.get(source) if value is not None: ap_loc.setdefault("address", {})[dest] = value if event_data.get("comments"): ap_object["commentsEnabled"] = True extra = event_data.get("extra") if extra: status = extra.get("status") if status: ap_object["ical:status"] = status.upper() website = extra.get("website") if website: attachment.append({ "href": website, "mediaType": "text/html", "name": "Website", "type": "Link" }) accessibility = extra.get("accessibility") if accessibility: wheelchair = accessibility.get("wheelchair") if wheelchair: if wheelchair == "full": ap_wc_value = "fully" elif wheelchair == "partial": ap_wc_value = "partially" elif wheelchair == "no": ap_wc_value = "no" else: log.error(f"unexpected wheelchair value: {wheelchair}") ap_wc_value = None if ap_wc_value is not None: attachment.append({ "propertyID": "mz:accessibility:wheelchairAccessible", "type": "PropertyValue", "value": ap_wc_value }) activity = self.apg.create_activity( "Create" if is_new else "Update", url_actor, ap_object, activity_id=url_item ) activity["@context"].append(AP_EVENTS_CONTEXT) return activity async def ap_item_2_event_data(self, ap_item: dict) -> dict: """Convert AP activity or object to event data @param ap_item: ActivityPub item to convert Can be either an activity of an object @return: AP Item's Object and event data @raise exceptions.DataError: something is invalid in the AP item """ is_activity = self.apg.is_activity(ap_item) if is_activity: ap_object = await self.apg.ap_get_object(ap_item, "object") if not ap_object: log.warning(f'No "object" found in AP item {ap_item!r}') raise exceptions.DataError else: ap_object = ap_item # id if "_repeated" in ap_item: # if the event is repeated, we use the original one ID repeated_uri = ap_item["_repeated"]["uri"] parsed_uri = uri.parse_xmpp_uri(repeated_uri) object_id = parsed_uri["item"] else: object_id = ap_object.get("id") if not object_id: raise exceptions.DataError('"id" is missing in AP object') if ap_item["type"] != TYPE_EVENT: raise exceptions.DataError("AP Object is not an event") # author actor = await self.apg.ap_get_sender_actor(ap_object) account = await self.apg.get_ap_account_from_id(actor) author_jid = self.apg.get_local_jid_from_account(account).full() # name, start, end event_data = { "id": object_id, "name": {"": ap_object.get("name") or "unnamed"}, "start": date_utils.date_parse(ap_object["startTime"]), "end": date_utils.date_parse(ap_object["endTime"]), } # attachments/extra event_data["extra"] = extra = {} attachments = ap_object.get("attachment") or [] for attachment in attachments: name = attachment.get("name") if name == "Banner": try: url = attachment["url"] except KeyError: log.warning(f"invalid attachment: {attachment}") continue event_data["head-picture"] = {"sources": [{"url": url}]} elif name == "Website": try: url = attachment["href"] except KeyError: log.warning(f"invalid attachment: {attachment}") continue extra["website"] = url else: log.debug(f"unmanaged attachment: {attachment}") # description content = ap_object.get("content") if content: event_data["descriptions"] = [{ "type": "xhtml", "description": content }] # categories tags = ap_object.get("tag") if tags: categories = event_data["categories"] = [] for tag in tags: if tag.get("type") == "Hashtag": try: term = tag["name"][1:] except KeyError: log.warning(f"invalid tag: {tag}") continue categories.append({"term": term}) #location ap_location = ap_object.get("location") if ap_location: location = {} for source, dest in ( ("name", "description"), ("latitude", "lat"), ("longitude", "lon"), ): value = ap_location.get(source) if value is not None: location[dest] = value address = ap_location.get("address") if address: for source, dest in ( ("addressCountry", "country"), ("addressLocality", "locality"), ("addressRegion", "region"), ("postalCode", "postalcode"), ("streetAddress", "street"), ): value = address.get(source) if value is not None: location[dest] = value if location: event_data["locations"] = [location] # rsvp # So far Mobilizon seems to only handle participate/don't participate, thus we use # a simple "yes"/"no" form. rsvp_data = {"fields": []} event_data["rsvp"] = [rsvp_data] rsvp_data["fields"].append({ "type": "list-single", "name": "attending", "label": "Attending", "options": [ {"label": "yes", "value": "yes"}, {"label": "no", "value": "no"} ], "required": True }) # comments if ap_object.get("commentsEnabled"): __, comments_node = await self.apg.get_comments_nodes(object_id, None) event_data["comments"] = { "service": author_jid, "node": comments_node, } # extra # part of extra come from "attachment" above status = ap_object.get("ical:status") if status is None: pass elif status in ("CONFIRMED", "CANCELLED", "TENTATIVE"): extra["status"] = status.lower() else: log.warning(f"unknown event status: {status}") return event_data async def ap_item_2_event_data_and_elt( self, ap_item: dict ) -> Tuple[dict, domish.Element]: """Convert AP item to parsed event data and corresponding item element""" event_data = await self.ap_item_2_event_data(ap_item) event_elt = self._events.event_data_2_event_elt(event_data) item_elt = domish.Element((None, "item")) item_elt["id"] = event_data["id"] item_elt.addChild(event_elt) return event_data, item_elt async def ap_item_2_event_elt(self, ap_item: dict) -> domish.Element: """Convert AP item to XMPP item element""" __, item_elt = await self.ap_item_2_event_data_and_elt(ap_item) return item_elt