Mercurial > libervia-backend
view src/plugins/plugin_xep_0055.py @ 699:e819630c6737
plugin text syntaxes: fixed bad parameter order for getParamA
author | Goffi <goffi@goffi.org> |
---|---|
date | Wed, 13 Nov 2013 14:00:07 +0100 |
parents | 694f118d0cd5 |
children | 7c806491c76a |
line wrap: on
line source
#!/usr/bin/python # -*- coding: utf-8 -*- # SAT plugin for Jabber Search (xep-0055) # Copyright (C) 2009, 2010, 2011, 2012, 2013 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 logging import debug, info, warning, error from twisted.words.protocols.jabber.xmlstream import IQ from twisted.words.protocols.jabber import jid from wokkel import data_form from sat.core.exceptions import DataError from sat.tools.xml_tools import dataForm2XML, dataFormResult2XML NS_SEARCH = 'jabber:iq:search' PLUGIN_INFO = { "name": "Jabber Search", "import_name": "XEP-0055", "type": "XEP", "protocols": ["XEP-0055"], "main": "XEP_0055", "handler": "no", "description": _("""Implementation of Jabber Search""") } class XEP_0055(object): def __init__(self, host): info(_("Jabber search plugin initialization")) self.host = host host.bridge.addMethod("getSearchUI", ".plugin", in_sign='ss', out_sign='s', method=self._getSearchUI, async=True) host.bridge.addMethod("searchRequest", ".plugin", in_sign='sa{ss}s', out_sign='s', method=self._searchRequest, async=True) def _getSearchUI(self, to_jid_s, profile_key): return self.getSearchUI(jid.JID(to_jid_s), profile_key) def getSearchUI(self, to_jid, profile_key): """ Ask for a search interface @param to_jid: XEP-0055 compliant search entity @param profile_key: %(doc_profile_key)s @return: XMLUI search interface """ client = self.host.getClient(profile_key) fields_request = IQ(client.xmlstream, 'get') fields_request["from"] = client.jid.full() fields_request["to"] = to_jid.full() fields_request.addElement('query', NS_SEARCH) d = fields_request.send(to_jid.full()) d.addCallbacks(self._fieldsOk, self._fieldsErr, callbackArgs=[client.profile], errbackArgs=[client.profile]) return d def _fieldsOk(self, answer, profile): """got fields available""" try: query_elts = answer.elements('jabber:iq:search', 'query').next() except StopIteration: info(_("No query element found")) raise DataError # FIXME: StanzaError is probably more appropriate, check the RFC try: form_elt = query_elts.elements(data_form.NS_X_DATA, 'x').next() except StopIteration: info(_("No data form found")) raise NotImplementedError("Only search through data form is implemented so far") parsed_form = data_form.Form.fromElement(form_elt) return dataForm2XML(parsed_form) def _fieldsErr(self, failure, profile): """ Called when something is wrong with fields request """ info(_("Fields request failure: %s") % str(failure.value)) return failure def _searchRequest(self, to_jid_s, search_dict, profile_key): return self.searchRequest(jid.JID(to_jid_s), search_dict, profile_key) def searchRequest(self, to_jid, search_dict, profile_key): """ Actually do a search, according to filled data @param to_jid: XEP-0055 compliant search entity @param search_dict: filled data, corresponding to the form obtained in getSearchUI @param profile_key: %(doc_profile_key)s @return: XMLUI search result """ client = self.host.getClient(profile_key) search_request = IQ(client.xmlstream, 'set') search_request["from"] = client.jid.full() search_request["to"] = to_jid.full() query_elt = search_request.addElement('query', NS_SEARCH) x_form = data_form.Form('submit') x_form.makeFields(search_dict) query_elt.addChild(x_form.toElement()) d = search_request.send(to_jid.full()) d.addCallbacks(self._searchOk, self._searchErr, callbackArgs=[client.profile], errbackArgs=[client.profile]) return d def _searchOk(self, answer, profile): """got search available""" try: query_elts = answer.elements('jabber:iq:search', 'query').next() except StopIteration: info(_("No query element found")) raise DataError # FIXME: StanzaError is probably more appropriate, check the RFC try: form_elt = query_elts.elements(data_form.NS_X_DATA, 'x').next() except StopIteration: info(_("No data form found")) raise NotImplementedError("Only search through data form is implemented so far") xmlui = dataFormResult2XML(form_elt) print "=== XMLUI ===\n%s\n\n" % xmlui return xmlui def _searchErr(self, failure, profile): """ Called when something is wrong with search request """ info(_("Search request failure: %s") % str(failure.value)) return failure