hackbot/plugins/wpsearch.py
2020-07-26 17:15:58 +02:00

155 lines
4.5 KiB
Python

#!/usr/bin/python
# -*- coding: utf-8 -*-
# file: wpsearch.py
# date: 23.07.2020
# desc: generic plugin
import json
import common
import logging
import requests
import threading
from manager import Plugin
logging = logging.getLogger()
lock = threading.Lock()
class Plugin(Plugin):
'''
'''
__module = __name__
__command = 'wpsearch'
__description = 'Seach wikipedia'
@staticmethod
def get_module():
return Plugin.__module
@staticmethod
def get_command():
return Plugin.__command
@staticmethod
def get_description():
return Plugin.__description
def __init__(self, callback=None):
self.callback = callback
def help(self):
return ('!wp-search searches in wikipedia for a given search term. '
'Arguments are language shortcut and search pattern. If no '
'language shortcut is given, en is used as default. '
'\nSyntax: !wp-search <lang> <pattern>')
def run(self, stanza):
'''
'''
lang = 'en'
pattern = None
call_msg = 'Call "!help {}"'.format(self.get_command())
no_args_msg = ' '.join(('No search pattern!', call_msg))
muc_nick = common.get_nick_from_stanza(stanza)
arguments = common.get_arguments_from_body(stanza)
logging.debug('Arguments: {}'.format(arguments))
if arguments is False:
logging.warning('No arguments for wp search. Abort.')
self.callback(': '.join((muc_nick, no_args_msg)))
return
else:
if len(arguments) == 1:
pattern = arguments[0]
elif len(arguments) > 1:
logging.warning('Not implemented yet.')
self.callback('Not implemented yet')
return
logging.debug('Start thread for wp search')
self.callback('Search started')
api_thread = ApiThread(self.callback)
api_thread.run(lang, pattern, muc_nick)
class ApiThread(threading.Thread):
'''
The thread who fetched and returns the wp search.
'''
def __init__(self, callback):
threading.Thread.__init__(self)
self.callback = callback
def run(self, lang, pattern, muc_nick):
'''
Starts the thread.
'''
data = self.get_file(lang, pattern)
if data == False:
self.callback('Error while connecting WP')
else:
# TODO: check if its a json needed !
logging.debug(data)
msg = self.string_factory(data)
self.callback(': '.join((muc_nick, msg)))
def get_file(self, lang, pattern):
'''
Grabs the API file, parse it and returns a json ... otherwise none.
param 1: string
param 2: string
param 3: string
returns: json or none
'''
api_url = 'https://{}.wikipedia.org/w/api.php'.format(lang)
api_params = {
'action': 'query',
'prop': 'extracts|info',
'explaintext': '',
'redirects': '',
'exchars': 200,
'continue': '',
'format': 'json',
'titles': pattern,
'inprop': 'url',
'formatversion': 2
}
try:
logging.debug('Try to fetch {}'.format(api_url))
session = requests.Session()
response = session.get(url = api_url, params = api_params)
data = response.json()
logging.debug('{}: successfull fetched'.format(api_url))
except:
logging.debug('{}: failed to fetch'.format(api_url))
return False
return data
def string_factory(self, data):
'''
param 1: json
returns: string
'''
msg = ''
if 'redirects' in data['query'].keys():
for i in data['query']['redirects']:
rfr = i['from']
rto = i['to']
msg = '\n'.join((msg, 'Redirect from {} to {}'.format( \
rfr, rto)))
logging.debug('Message: {}'.format(msg))
pages = data['query']['pages']
for i in pages:
if 'extract' in i.keys():
msg = '\n'.join((msg, 'Summary:', i['extract']))
logging.debug('Message: {}'.format(msg))
else: msg = '\n'.join((msg, 'Nothing found'))
msg = '\n'.join((msg, 'URL:', i['fullurl']))
logging.debug('Message: {}'.format(msg))
return msg