sopel.modules.help.post_to_ubuntu() - Code Metrics - Inspection of "Merge pull request #1651 from ddstreet/pastebin-ub..." - sopel-irc/sopel - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( c752b1...225321 )

by dgw

created 2019-09-26 19:05 UTC

sopel.modules.help.post_to_ubuntu() A

↳ Parent: sopel.modules.help

Complexity

Conditions

Size

Total Lines	17
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	14
dl	0
loc	17
rs	9.7
c	0
b	0
f	0
cc	3
nop	1

# coding=utf-8
"""
help.py - Sopel Help Module
Copyright 2008, Sean B. Palmer, inamidst.com
Copyright © 2013, Elad Alfassa, <[email protected]>
Copyright © 2018, Adam Erdman, pandorah.org
Copyright © 2019, Tomasz Kurcz, github.com/uint
Copyright © 2019, dgw, technobabbl.es
Licensed under the Eiffel Forum License 2.

https://sopel.chat
"""
from __future__ import unicode_literals, absolute_import, print_function, division

import re
import collections
import socket
import textwrap

import requests

from sopel.config.types import ChoiceAttribute, ValidatedAttribute, StaticSection
from sopel.logger import get_logger
from sopel.module import commands, rule, example, priority
from sopel.tools import SopelMemory


SETTING_CACHE_NAMESPACE = 'help-setting-cache'  # Set top-level memory key name
LOGGER = get_logger(__name__)

# Settings that should require the help listing to be regenerated, or
# re-POSTed to paste, if they are changed during runtime.
# Keys are module names, and values are lists of setting names
# specific to that module.
TRACKED_SETTINGS = {
    'help': [
        'output',
        'show_server_host',
    ]
}


class PostingException(Exception):
    """Custom exception type for errors posting help to the chosen pastebin."""
    pass


# Pastebin handlers


def _requests_post_catch_errors(*args, **kwargs):
    try:
        response = requests.post(*args, **kwargs)
        response.raise_for_status()
    except (
            requests.exceptions.Timeout,
            requests.exceptions.TooManyRedirects,
            requests.exceptions.RequestException,
            requests.exceptions.HTTPError
    ):
        # We re-raise all expected exception types to a generic "posting error"
        # that's easy for callers to expect, and then we pass the original
        # exception through to provide some debugging info
        LOGGER.exception('Error during POST request')
        raise PostingException('Could not communicate with remote service')

    # remaining handling (e.g. errors inside the response) is left to the caller
    return response


def post_to_clbin(msg):
    try:
        result = _requests_post_catch_errors('https://clbin.com/', data={'clbin': msg})
    except PostingException:
        raise

    result = result.text
    if '://clbin.com/' in result:
        # find/replace just in case the site tries to be sneaky and save on SSL overhead,
        # though it will probably send us an HTTPS link without any tricks.
        return result.replace('http://', 'https://', 1)
    else:
        LOGGER.error("Invalid result %s", result)
        raise PostingException('clbin result did not contain expected URL base.')


def post_to_0x0(msg):
    try:
        result = _requests_post_catch_errors('https://0x0.st', files={'file': msg})
    except PostingException:
        raise

    result = result.text
    if '://0x0.st' in result:
        # find/replace just in case the site tries to be sneaky and save on SSL overhead,
        # though it will probably send us an HTTPS link without any tricks.
        return result.replace('http://', 'https://', 1)
    else:
        LOGGER.error('Invalid result %s', result)
        raise PostingException('0x0.st result did not contain expected URL base.')


def post_to_hastebin(msg):
    try:
        result = _requests_post_catch_errors('https://hastebin.com/documents', data=msg)
    except PostingException:
        raise

    try:
        result = result.json()
    except ValueError:
        LOGGER.error("Invalid Hastebin response %s", result)
        raise PostingException('Could not parse response from Hastebin!')

    if 'key' not in result:
        LOGGER.error("Invalid result %s", result)
        raise PostingException('Hastebin result did not contain expected URL base.')
    return "https://hastebin.com/" + result['key']


def post_to_termbin(msg):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(10)  # the bot may NOT wait forever for a response; that would be bad
    try:
        sock.connect(('termbin.com', 9999))
        sock.sendall(msg)
        sock.shutdown(socket.SHUT_WR)
        response = ""
        while 1:
            data = sock.recv(1024)
            if data == "":
                break
            response += data
        sock.close()
    except socket.error:
        LOGGER.exception('Error during communication with termbin')
        raise PostingException('Error uploading to termbin')

    # find/replace just in case the site tries to be sneaky and save on SSL overhead,
    # though it will probably send us an HTTPS link without any tricks.
    return response.strip('\x00\n').replace('http://', 'https://', 1)


def post_to_ubuntu(msg):
    data = {
        'poster': 'sopel',
        'syntax': 'text',
        'expiration': '',
        'content': msg,
    }
    try:
        result = _requests_post_catch_errors('https://pastebin.ubuntu.com/', data=data)
    except PostingException:
        raise

    if not re.match(r'https://pastebin.ubuntu.com/p/[^/]+/', result.url):
        LOGGER.error("Invalid Ubuntu pastebin response url %s", result.url)
        raise PostingException('Invalid response from Ubuntu pastebin: %s' % result.url)

    return result.url


PASTEBIN_PROVIDERS = {
    'clbin': post_to_clbin,
    '0x0': post_to_0x0,
    'hastebin': post_to_hastebin,
    'termbin': post_to_termbin,
    'ubuntu': post_to_ubuntu,
}


class HelpSection(StaticSection):
    """Configuration section for this module."""
    output = ChoiceAttribute('output',
                             list(PASTEBIN_PROVIDERS),
                             default='clbin')
    """The pastebin provider to use for help output."""
    show_server_host = ValidatedAttribute('show_server_host', bool, default=True)
    """Show the IRC server's hostname/IP in the first line of the help listing?"""


def configure(config):
    """
    | name | example | purpose |
    | ---- | ------- | ------- |
    | output | clbin | The pastebin provider to use for help output |
    | show\\_server\\_host | True | Whether to show the IRC server's hostname/IP at the top of command listings |
    """
    config.define_section('help', HelpSection)
    provider_list = ', '.join(PASTEBIN_PROVIDERS)
    config.help.configure_setting(
        'output',
        'Pick a pastebin provider: {}: '.format(provider_list)
    )
    config.help.configure_setting(
        'show_server_host',
        'Should the help command show the IRC server\'s hostname/IP in the listing?'
    )


def setup(bot):
    bot.config.define_section('help', HelpSection)

    # Initialize memory
    if SETTING_CACHE_NAMESPACE not in bot.memory:

        bot.memory[SETTING_CACHE_NAMESPACE] = SopelMemory()

    # Initialize settings cache
    for section in TRACKED_SETTINGS:
        if section not in bot.memory[SETTING_CACHE_NAMESPACE]:
            bot.memory[SETTING_CACHE_NAMESPACE][section] = SopelMemory()

    update_cache(bot)  # Populate cache

    bot.config.define_section('help', HelpSection)


def update_cache(bot):
    for section, setting_names_list in TRACKED_SETTINGS.items():
        for setting_name in setting_names_list:
            bot.memory[SETTING_CACHE_NAMESPACE][section][setting_name] = getattr(getattr(bot.config, section), setting_name)


def is_cache_valid(bot):
    for section, setting_names_list in TRACKED_SETTINGS.items():
        for setting_name in setting_names_list:
            if bot.memory[SETTING_CACHE_NAMESPACE][section][setting_name] != getattr(getattr(bot.config, section), setting_name):
                return False
    return True


@rule('$nick' r'(?i)(help|doc) +([A-Za-z]+)(?:\?+)?$')
@example('.help tell')
@commands('help', 'commands')
@priority('low')
def help(bot, trigger):
    """Shows a command's documentation, and an example if available. With no arguments, lists all commands."""
    if trigger.group(2):
        name = trigger.group(2)
        name = name.lower()

        # number of lines of help to show
        threshold = 3

        if name in bot.doc:
            # count lines we're going to send
            # lines in command docstring, plus one line for example(s) if present (they're sent all on one line)
            if len(bot.doc[name][0]) + int(bool(bot.doc[name][1])) > threshold:
                if trigger.nick != trigger.sender:  # don't say that if asked in private
                    bot.reply('The documentation for this command is too long; '
                              'I\'m sending it to you in a private message.')

                def msgfun(l):
                    bot.say(l, trigger.nick)
            else:
                msgfun = bot.reply

            for line in bot.doc[name][0]:
                msgfun(line)
            if bot.doc[name][1]:
                # Build a nice, grammatically-correct list of examples
                examples = ', '.join(bot.doc[name][1][:-2] + [' or '.join(bot.doc[name][1][-2:])])
                msgfun('e.g. ' + examples)
    else:
        # This'll probably catch most cases, without having to spend the time
        # actually creating the list first. Maybe worth storing the link and a
        # heuristic in the DB, too, so it persists across restarts. Would need a
        # command to regenerate, too...
        if (
            'command-list' in bot.memory and
            bot.memory['command-list'][0] == len(bot.command_groups) and
            is_cache_valid(bot)
        ):
            url = bot.memory['command-list'][1]
        else:
            bot.say("Hang on, I'm creating a list.")
            msgs = []

            name_length = max(6, max(len(k) for k in bot.command_groups.keys()))
            for category, cmds in collections.OrderedDict(sorted(bot.command_groups.items())).items():
                category = category.upper().ljust(name_length)
                cmds = set(cmds)  # remove duplicates
                cmds = '  '.join(cmds)
                msg = category + '  ' + cmds
                indent = ' ' * (name_length + 2)
                # Honestly not sure why this is a list here
                msgs.append('\n'.join(textwrap.wrap(msg, subsequent_indent=indent)))

            url = create_list(bot, '\n\n'.join(msgs))
            if not url:
                return
            bot.memory['command-list'] = (len(bot.command_groups), url)
            update_cache(bot)
        bot.say("I've posted a list of my commands at {0} - You can see "
                "more info about any of these commands by doing {1}help "
                "<command> (e.g. {1}help time)"
                .format(url, bot.config.core.help_prefix))


def create_list(bot, msg):
    """Creates & uploads the command list.

    Returns the URL from the chosen pastebin provider.
    """
    msg = 'Command listing for {}{}\n\n{}'.format(
        bot.nick,
        ('@' + bot.config.core.host) if bot.config.help.show_server_host else '',
        msg)

    try:
        result = PASTEBIN_PROVIDERS[bot.config.help.output](msg)
    except PostingException:
        bot.say("Sorry! Something went wrong.")
        LOGGER.exception("Error posting commands")
        return
    return result


@rule('$nick' r'(?i)help(?:[?!]+)?$')
@priority('low')
def help2(bot, trigger):
    response = (
        "Hi, I'm a bot. Say {1}commands to me in private for a list "
        "of my commands, or see https://sopel.chat for more "
        "general details. My owner is {0}."
        .format(bot.config.core.owner, bot.config.core.help_prefix))
    bot.reply(response)


1			# coding=utf-8
2			"""
3			help.py - Sopel Help Module
4			Copyright 2008, Sean B. Palmer, inamidst.com
5			Copyright © 2013, Elad Alfassa, <[email protected]>
6			Copyright © 2018, Adam Erdman, pandorah.org
7			Copyright © 2019, Tomasz Kurcz, github.com/uint
8			Copyright © 2019, dgw, technobabbl.es
9			Licensed under the Eiffel Forum License 2.
10
11			https://sopel.chat
12			"""
13			from __future__ import unicode_literals, absolute_import, print_function, division
14
15			import re
16			import collections
17			import socket
18			import textwrap
19
20			import requests
21
22			from sopel.config.types import ChoiceAttribute, ValidatedAttribute, StaticSection
23			from sopel.logger import get_logger
24			from sopel.module import commands, rule, example, priority
25			from sopel.tools import SopelMemory
26
27
28			SETTING_CACHE_NAMESPACE = 'help-setting-cache' # Set top-level memory key name
29			LOGGER = get_logger(__name__)
30
31			# Settings that should require the help listing to be regenerated, or
32			# re-POSTed to paste, if they are changed during runtime.
33			# Keys are module names, and values are lists of setting names
34			# specific to that module.
35			TRACKED_SETTINGS = {
36			'help': [
37			'output',
38			'show_server_host',
39			]
40			}
41
42
43			class PostingException(Exception):
44			"""Custom exception type for errors posting help to the chosen pastebin."""
45			pass
46
47
48			# Pastebin handlers
49
50
51			def _requests_post_catch_errors(args, *kwargs):
52			try:
53			response = requests.post(args, *kwargs)
54			response.raise_for_status()
55			except (
56			requests.exceptions.Timeout,
57			requests.exceptions.TooManyRedirects,
58			requests.exceptions.RequestException,
59			requests.exceptions.HTTPError
60			):
61			# We re-raise all expected exception types to a generic "posting error"
62			# that's easy for callers to expect, and then we pass the original
63			# exception through to provide some debugging info
64			LOGGER.exception('Error during POST request')
65			raise PostingException('Could not communicate with remote service')
66
67			# remaining handling (e.g. errors inside the response) is left to the caller
68			return response
69
70
71			def post_to_clbin(msg):
72			try:
73			result = _requests_post_catch_errors('https://clbin.com/', data={'clbin': msg})
74			except PostingException:
75			raise
76
77			result = result.text
78			if '://clbin.com/' in result:
79			# find/replace just in case the site tries to be sneaky and save on SSL overhead,
80			# though it will probably send us an HTTPS link without any tricks.
81			return result.replace('http://', 'https://', 1)
82			else:
83			LOGGER.error("Invalid result %s", result)
84			raise PostingException('clbin result did not contain expected URL base.')
85
86
87			def post_to_0x0(msg):
88			try:
89			result = _requests_post_catch_errors('https://0x0.st', files={'file': msg})
90			except PostingException:
91			raise
92
93			result = result.text
94			if '://0x0.st' in result:
95			# find/replace just in case the site tries to be sneaky and save on SSL overhead,
96			# though it will probably send us an HTTPS link without any tricks.
97			return result.replace('http://', 'https://', 1)
98			else:
99			LOGGER.error('Invalid result %s', result)
100			raise PostingException('0x0.st result did not contain expected URL base.')
101
102
103			def post_to_hastebin(msg):
104			try:
105			result = _requests_post_catch_errors('https://hastebin.com/documents', data=msg)
106			except PostingException:
107			raise
108
109			try:
110			result = result.json()
111			except ValueError:
112			LOGGER.error("Invalid Hastebin response %s", result)
113			raise PostingException('Could not parse response from Hastebin!')
114
115			if 'key' not in result:
116			LOGGER.error("Invalid result %s", result)
117			raise PostingException('Hastebin result did not contain expected URL base.')
118			return "https://hastebin.com/" + result['key']
119
120
121			def post_to_termbin(msg):
122			sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
123			sock.settimeout(10) # the bot may NOT wait forever for a response; that would be bad
124			try:
125			sock.connect(('termbin.com', 9999))
126			sock.sendall(msg)
127			sock.shutdown(socket.SHUT_WR)
128			response = ""
129			while 1:
130			data = sock.recv(1024)
131			if data == "":
132			break
133			response += data
134			sock.close()
135			except socket.error:
136			LOGGER.exception('Error during communication with termbin')
137			raise PostingException('Error uploading to termbin')
138
139			# find/replace just in case the site tries to be sneaky and save on SSL overhead,
140			# though it will probably send us an HTTPS link without any tricks.
141			return response.strip('\x00\n').replace('http://', 'https://', 1)
142
143
144			def post_to_ubuntu(msg):
145			data = {
146			'poster': 'sopel',
147			'syntax': 'text',
148			'expiration': '',
149			'content': msg,
150			}
151			try:
152			result = _requests_post_catch_errors('https://pastebin.ubuntu.com/', data=data)
153			except PostingException:
154			raise
155
156			if not re.match(r'https://pastebin.ubuntu.com/p/[^/]+/', result.url):
157			LOGGER.error("Invalid Ubuntu pastebin response url %s", result.url)
158			raise PostingException('Invalid response from Ubuntu pastebin: %s' % result.url)
159
160			return result.url
161
162
163			PASTEBIN_PROVIDERS = {
164			'clbin': post_to_clbin,
165			'0x0': post_to_0x0,
166			'hastebin': post_to_hastebin,
167			'termbin': post_to_termbin,
168			'ubuntu': post_to_ubuntu,
169			}
170
171
172			class HelpSection(StaticSection):
173			"""Configuration section for this module."""
174			output = ChoiceAttribute('output',
175			list(PASTEBIN_PROVIDERS),
176			default='clbin')
177			"""The pastebin provider to use for help output."""
178			show_server_host = ValidatedAttribute('show_server_host', bool, default=True)
179			"""Show the IRC server's hostname/IP in the first line of the help listing?"""
180
181
182			def configure(config):
183			"""
184			\| name \| example \| purpose \|
185			\| ---- \| ------- \| ------- \|
186			\| output \| clbin \| The pastebin provider to use for help output \|
187			\| show\\_server\\_host \| True \| Whether to show the IRC server's hostname/IP at the top of command listings \|
188			"""
189			config.define_section('help', HelpSection)
190			provider_list = ', '.join(PASTEBIN_PROVIDERS)
191			config.help.configure_setting(
192			'output',
193			'Pick a pastebin provider: {}: '.format(provider_list)
194			)
195			config.help.configure_setting(
196			'show_server_host',
197			'Should the help command show the IRC server\'s hostname/IP in the listing?'
198			)
199
200
201			def setup(bot):
202			bot.config.define_section('help', HelpSection)
203
204			# Initialize memory
205			if SETTING_CACHE_NAMESPACE not in bot.memory:
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-04-25 23:30 UTC by Report Bug Copy Issue Report The variable `SETTING_CACHE_NAMESPACE` does not seem to be defined. Loading history...
206			bot.memory[SETTING_CACHE_NAMESPACE] = SopelMemory()
207
208			# Initialize settings cache
209			for section in TRACKED_SETTINGS:
210			if section not in bot.memory[SETTING_CACHE_NAMESPACE]:
211			bot.memory[SETTING_CACHE_NAMESPACE][section] = SopelMemory()
212
213			update_cache(bot) # Populate cache
214
215			bot.config.define_section('help', HelpSection)
216
217
218			def update_cache(bot):
219			for section, setting_names_list in TRACKED_SETTINGS.items():
220			for setting_name in setting_names_list:
221			bot.memory[SETTING_CACHE_NAMESPACE][section][setting_name] = getattr(getattr(bot.config, section), setting_name)
222
223
224			def is_cache_valid(bot):
225			for section, setting_names_list in TRACKED_SETTINGS.items():
226			for setting_name in setting_names_list:
227			if bot.memory[SETTING_CACHE_NAMESPACE][section][setting_name] != getattr(getattr(bot.config, section), setting_name):
228			return False
229			return True
230
231
232			@rule('$nick' r'(?i)(help\|doc) +([A-Za-z]+)(?:\?+)?$')
233			@example('.help tell')
234			@commands('help', 'commands')
235			@priority('low')
236			def help(bot, trigger):
237			"""Shows a command's documentation, and an example if available. With no arguments, lists all commands."""
238			if trigger.group(2):
239			name = trigger.group(2)
240			name = name.lower()
241
242			# number of lines of help to show
243			threshold = 3
244
245			if name in bot.doc:
246			# count lines we're going to send
247			# lines in command docstring, plus one line for example(s) if present (they're sent all on one line)
248			if len(bot.doc[name][0]) + int(bool(bot.doc[name][1])) > threshold:
249			if trigger.nick != trigger.sender: # don't say that if asked in private
250			bot.reply('The documentation for this command is too long; '
251			'I\'m sending it to you in a private message.')
252
253			def msgfun(l):
254			bot.say(l, trigger.nick)
255			else:
256			msgfun = bot.reply
257
258			for line in bot.doc[name][0]:
259			msgfun(line)
260			if bot.doc[name][1]:
261			# Build a nice, grammatically-correct list of examples
262			examples = ', '.join(bot.doc[name][1][:-2] + [' or '.join(bot.doc[name][1][-2:])])
263			msgfun('e.g. ' + examples)
264			else:
265			# This'll probably catch most cases, without having to spend the time
266			# actually creating the list first. Maybe worth storing the link and a
267			# heuristic in the DB, too, so it persists across restarts. Would need a
268			# command to regenerate, too...
269			if (
270			'command-list' in bot.memory and
271			bot.memory['command-list'][0] == len(bot.command_groups) and
272			is_cache_valid(bot)
273			):
274			url = bot.memory['command-list'][1]
275			else:
276			bot.say("Hang on, I'm creating a list.")
277			msgs = []
278
279			name_length = max(6, max(len(k) for k in bot.command_groups.keys()))
280			for category, cmds in collections.OrderedDict(sorted(bot.command_groups.items())).items():
281			category = category.upper().ljust(name_length)
282			cmds = set(cmds) # remove duplicates
283			cmds = ' '.join(cmds)
284			msg = category + ' ' + cmds
285			indent = ' ' * (name_length + 2)
286			# Honestly not sure why this is a list here
287			msgs.append('\n'.join(textwrap.wrap(msg, subsequent_indent=indent)))
288
289			url = create_list(bot, '\n\n'.join(msgs))
290			if not url:
291			return
292			bot.memory['command-list'] = (len(bot.command_groups), url)
293			update_cache(bot)
294			bot.say("I've posted a list of my commands at {0} - You can see "
295			"more info about any of these commands by doing {1}help "
296			"<command> (e.g. {1}help time)"
297			.format(url, bot.config.core.help_prefix))
298
299
300			def create_list(bot, msg):
301			"""Creates & uploads the command list.
302
303			Returns the URL from the chosen pastebin provider.
304			"""
305			msg = 'Command listing for {}{}\n\n{}'.format(
306			bot.nick,
307			('@' + bot.config.core.host) if bot.config.help.show_server_host else '',
308			msg)
309
310			try:
311			result = PASTEBIN_PROVIDERS[bot.config.help.output](msg)
312			except PostingException:
313			bot.say("Sorry! Something went wrong.")
314			LOGGER.exception("Error posting commands")
315			return
316			return result
317
318
319			@rule('$nick' r'(?i)help(?:[?!]+)?$')
320			@priority('low')
321			def help2(bot, trigger):
322			response = (
323			"Hi, I'm a bot. Say {1}commands to me in private for a list "
324			"of my commands, or see https://sopel.chat for more "
325			"general details. My owner is {0}."
326			.format(bot.config.core.owner, bot.config.core.help_prefix))
327			bot.reply(response)
328

sopel-irc / sopel

Push — master ( c752b1...225321 )

sopel.modules.help.post_to_ubuntu() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like