sopel/tools/__init__.py
2017-11-22 19:26:40 -05:00

353 lines
11 KiB
Python
Executable File

# coding=utf-8
"""Useful miscellaneous tools and shortcuts for Sopel modules
*Availability: 3+*
"""
# tools.py - Sopel misc tools
# Copyright 2008, Sean B. Palmer, inamidst.com
# Copyright © 2012, Elad Alfassa <elad@fedoraproject.org>
# Copyright 2012, Elsie Powell, embolalia.com
# Licensed under the Eiffel Forum License 2.
# https://sopel.chat
from __future__ import unicode_literals, absolute_import, print_function, division
import sys
import os
import re
import threading
import codecs
import traceback
from collections import defaultdict
from tools._events import events # NOQA
if sys.version_info.major >= 3:
raw_input = input
unicode = str
iteritems = dict.items
itervalues = dict.values
iterkeys = dict.keys
else:
iteritems = dict.iteritems
itervalues = dict.itervalues
iterkeys = dict.iterkeys
_channel_prefixes = ('#', '&', '+', '!')
def get_input(prompt):
"""Get decoded input from the terminal (equivalent to python 3's ``input``).
"""
if sys.version_info.major >= 3:
return input(prompt)
else:
return raw_input(prompt).decode('utf8')
def get_raising_file_and_line(tb=None):
"""Return the file and line number of the statement that raised the tb.
Returns: (filename, lineno) tuple
"""
if not tb:
tb = sys.exc_info()[2]
filename, lineno, _context, _line = traceback.extract_tb(tb)[-1]
return filename, lineno
def get_command_regexp(prefix, command):
"""Return a compiled regexp object that implements the command."""
# Escape all whitespace with a single backslash. This ensures that regexp
# in the prefix is treated as it was before the actual regexp was changed
# to use the verbose syntax.
prefix = re.sub(r"(\s)", r"\\\1", prefix)
# This regexp match equivalently and produce the same
# groups 1 and 2 as the old regexp: r'^%s(%s)(?: +(.*))?$'
# The only differences should be handling all whitespace
# like spaces and the addition of groups 3-6.
pattern = r"""
(?:{prefix})({command}) # Command as group 1.
(?:\s+ # Whitespace to end command.
( # Rest of the line as group 2.
(?:(\S+))? # Parameters 1-4 as groups 3-6.
(?:\s+(\S+))?
(?:\s+(\S+))?
(?:\s+(\S+))?
.* # Accept anything after the parameters.
# Leave it up to the module to parse
# the line.
))? # Group 2 must be None, if there are no
# parameters.
$ # EoL, so there are no partial matches.
""".format(prefix=prefix, command=command)
return re.compile(pattern, re.IGNORECASE | re.VERBOSE)
def deprecated(old):
def new(*args, **kwargs):
print('Function %s is deprecated.' % old.__name__, file=sys.stderr)
trace = traceback.extract_stack()
for line in traceback.format_list(trace[:-1]):
stderr(line[:-1])
return old(*args, **kwargs)
new.__doc__ = old.__doc__
new.__name__ = old.__name__
return new
# from
# http://parand.com/say/index.php/2007/07/13/simple-multi-dimensional-dictionaries-in-python/
# A simple class to make mutli dimensional dict easy to use
class Ddict(dict):
"""Class for multi-dimensional ``dict``.
A simple helper class to ease the creation of multi-dimensional ``dict``\s.
"""
def __init__(self, default=None):
self.default = default
def __getitem__(self, key):
if key not in self:
self[key] = self.default()
return dict.__getitem__(self, key)
class Identifier(unicode):
"""A `unicode` subclass which acts appropriately for IRC identifiers.
When used as normal `unicode` objects, case will be preserved.
However, when comparing two Identifier objects, or comparing a Identifier
object with a `unicode` object, the comparison will be case insensitive.
This case insensitivity includes the case convention conventions regarding
``[]``, ``{}``, ``|``, ``\\``, ``^`` and ``~`` described in RFC 2812.
"""
def __new__(cls, identifier):
# According to RFC2812, identifiers have to be in the ASCII range.
# However, I think it's best to let the IRCd determine that, and we'll
# just assume unicode. It won't hurt anything, and is more internally
# consistent. And who knows, maybe there's another use case for this
# weird case convention.
s = unicode.__new__(cls, identifier)
s._lowered = Identifier._lower(identifier)
return s
def lower(self):
"""Return the identifier converted to lower-case per RFC 2812."""
return self._lowered
@staticmethod
def _lower(identifier):
"""Returns `identifier` in lower case per RFC 2812."""
# The tilde replacement isn't needed for identifiers, but is for
# channels, which may be useful at some point in the future.
low = identifier.lower().replace('{', '[').replace('}', ']')
low = low.replace('|', '\\').replace('^', '~')
return low
def __repr__(self):
return "%s(%r)" % (
self.__class__.__name__,
self.__str__()
)
def __hash__(self):
return self._lowered.__hash__()
def __lt__(self, other):
if isinstance(other, Identifier):
return self._lowered < other._lowered
return self._lowered < Identifier._lower(other)
def __le__(self, other):
if isinstance(other, Identifier):
return self._lowered <= other._lowered
return self._lowered <= Identifier._lower(other)
def __gt__(self, other):
if isinstance(other, Identifier):
return self._lowered > other._lowered
return self._lowered > Identifier._lower(other)
def __ge__(self, other):
if isinstance(other, Identifier):
return self._lowered >= other._lowered
return self._lowered >= Identifier._lower(other)
def __eq__(self, other):
if isinstance(other, Identifier):
return self._lowered == other._lowered
return self._lowered == Identifier._lower(other)
def __ne__(self, other):
return not (self == other)
def is_nick(self):
"""Returns True if the Identifier is a nickname (as opposed to channel)
"""
return self and not self.startswith(_channel_prefixes)
class OutputRedirect(object):
"""Redirect te output to the terminal and a log file.
A simplified object used to write to both the terminal and a log file.
"""
def __init__(self, logpath, stderr=False, quiet=False):
"""Create an object which will to to a file and the terminal.
Create an object which will log to the file at ``logpath`` as well as
the terminal.
If ``stderr`` is given and true, it will write to stderr rather than
stdout.
If ``quiet`` is given and True, data will be written to the log file
only, but not the terminal.
"""
self.logpath = logpath
self.stderr = stderr
self.quiet = quiet
def write(self, string):
"""Write the given ``string`` to the logfile and terminal."""
if not self.quiet:
try:
if self.stderr:
sys.__stderr__.write(string)
else:
sys.__stdout__.write(string)
except:
pass
with codecs.open(self.logpath, 'ab', encoding="utf8",
errors='xmlcharrefreplace') as logfile:
try:
logfile.write(string)
except UnicodeDecodeError:
# we got an invalid string, safely encode it to utf-8
logfile.write(unicode(string, 'utf8', errors="replace"))
def flush(self):
if self.stderr:
sys.__stderr__.flush()
else:
sys.__stdout__.flush()
# These seems to trace back to when we thought we needed a try/except on prints,
# because it looked like that was why we were having problems. We'll drop it in
# 4.0^H^H^H5.0^H^H^H6.0^H^H^Hsome version when someone can be bothered.
@deprecated
def stdout(string):
print(string)
def stderr(string):
"""Print the given ``string`` to stderr.
This is equivalent to ``print >> sys.stderr, string``
"""
print(string, file=sys.stderr)
def check_pid(pid):
"""Check if a process is running with the given ``PID``.
*Availability: Only on POSIX systems*
Return ``True`` if there is a process running with the given ``PID``.
"""
try:
os.kill(pid, 0)
except OSError:
return False
else:
return True
def get_hostmask_regex(mask):
"""Return a compiled `re.RegexObject` for an IRC hostmask"""
mask = re.escape(mask)
mask = mask.replace(r'\*', '.*')
return re.compile(mask + '$', re.I)
class SopelMemory(dict):
"""A simple thread-safe dict implementation.
*Availability: 4.0; available as ``Sopel.SopelMemory`` in 3.1.0 - 3.2.0*
In order to prevent exceptions when iterating over the values and changing
them at the same time from different threads, we use a blocking lock on
``__setitem__`` and ``contains``.
"""
def __init__(self, *args):
dict.__init__(self, *args)
self.lock = threading.Lock()
def __setitem__(self, key, value):
self.lock.acquire()
result = dict.__setitem__(self, key, value)
self.lock.release()
return result
def __contains__(self, key):
"""Check if a key is in the dict.
It locks it for writes when doing so.
"""
self.lock.acquire()
result = dict.__contains__(self, key)
self.lock.release()
return result
def contains(self, key):
"""Backwards compatability with 3.x, use `in` operator instead."""
return self.__contains__(key)
class SopelMemoryWithDefault(defaultdict):
"""Same as SopelMemory, but subclasses from collections.defaultdict."""
def __init__(self, *args):
defaultdict.__init__(self, *args)
self.lock = threading.Lock()
def __setitem__(self, key, value):
self.lock.acquire()
result = defaultdict.__setitem__(self, key, value)
self.lock.release()
return result
def __contains__(self, key):
"""Check if a key is in the dict.
It locks it for writes when doing so.
"""
self.lock.acquire()
result = defaultdict.__contains__(self, key)
self.lock.release()
return result
def contains(self, key):
"""Backwards compatability with 3.x, use `in` operator instead."""
return self.__contains__(key)