Mercurial > libervia-backend
diff sat/plugins/plugin_misc_uri_finder.py @ 2562:26edcf3a30eb
core, setup: huge cleaning:
- moved directories from src and frontends/src to sat and sat_frontends, which is the recommanded naming convention
- move twisted directory to root
- removed all hacks from setup.py, and added missing dependencies, it is now clean
- use https URL for website in setup.py
- removed "Environment :: X11 Applications :: GTK", as wix is deprecated and removed
- renamed sat.sh to sat and fixed its installation
- added python_requires to specify Python version needed
- replaced glib2reactor which use deprecated code by gtk3reactor
sat can now be installed directly from virtualenv without using --system-site-packages anymore \o/
author | Goffi <goffi@goffi.org> |
---|---|
date | Mon, 02 Apr 2018 19:44:50 +0200 |
parents | src/plugins/plugin_misc_uri_finder.py@0062d3e79d12 |
children | 003b8b4b56a7 |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/sat/plugins/plugin_misc_uri_finder.py Mon Apr 02 19:44:50 2018 +0200 @@ -0,0 +1,96 @@ +#!/usr/bin/env python2 +# -*- coding: utf-8 -*- + +# SAT plugin to find URIs +# Copyright (C) 2009-2018 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 sat.core.i18n import _ +from sat.core.constants import Const as C +from sat.core.log import getLogger +from twisted.internet import defer +import textwrap +log = getLogger(__name__) +import json +import os.path +import os +import re + +PLUGIN_INFO = { + C.PI_NAME: _("URI finder"), + C.PI_IMPORT_NAME: "uri_finder", + C.PI_TYPE: "EXP", + C.PI_PROTOCOLS: [], + C.PI_DEPENDENCIES: [], + C.PI_MAIN: "URIFinder", + C.PI_HANDLER: "no", + C.PI_DESCRIPTION: textwrap.dedent(_(u"""\ + Plugin to find URIs in well know location. + This allows to retrieve settings to work with a project (e.g. pubsub node used for merge-requests). + """)) +} + + +SEARCH_FILES = ('readme', 'contributing') + + +class URIFinder(object): + + def __init__(self, host): + log.info(_(u"URI finder plugin initialization")) + self.host = host + host.bridge.addMethod("URIFind", ".plugin", + in_sign='sas', out_sign='a{sa{ss}}', + method=self.find, + async=True) + + def find(self, path, keys): + """Look for URI in well known locations + + @param path(unicode): path to start with + @param keys(list[unicode]): keys lookeds after + e.g.: "tickets", "merge-requests" + @return (dict[unicode, unicode]): map from key to found uri + """ + keys_re = u'|'.join(keys) + label_re = r'"(?P<label>[^"]+)"' + uri_re = re.compile(ur'(?P<key>{keys_re})[ :]? +(?P<uri>xmpp:\S+)(?:.*use {label_re} label)?'.format( + keys_re=keys_re, label_re = label_re)) + path = os.path.normpath(path) + if not os.path.isdir(path) or not os.path.isabs(path): + raise ValueError(u'path must be an absolute path to a directory') + + found_uris = {} + while path != u'/': + for filename in os.listdir(path): + name, __ = os.path.splitext(filename) + if name.lower() in SEARCH_FILES: + file_path = os.path.join(path, filename) + with open(file_path) as f: + for m in uri_re.finditer(f.read()): + key = m.group(u'key') + uri = m.group(u'uri') + label = m.group(u'label') + if key in found_uris: + log.warning(_(u"Ignoring already found uri for key \"{key}\"").format(key=key)) + else: + uri_data = found_uris[key] = {u'uri': uri} + if label is not None: + uri_data[u'labels'] = json.dumps([label]) + if found_uris: + break + path = os.path.dirname(path) + + return defer.succeed(found_uris)