csbot.util module

csbot.util.nick(user)

Get nick from user string.

>>> nick('csyorkbot!~csbot@example.com')
'csyorkbot'

csbot.util.username(user)

Get username from user string.

>>> username('csyorkbot!~csbot@example.com')
'csbot'

csbot.util.host(user)

Get hostname from user string.

>>> host('csyorkbot!~csbot@example.com')
'example.com'

csbot.util.is_channel(channel)

Check if channel is a channel or private chat.

>>> is_channel('#cs-york')
True
>>> is_channel('csyorkbot')
False

csbot.util.parse_arguments(raw)

Parse raw into a list of arguments using shlex.

The shlex lexer is customised to be more appropriate for grouping natural language arguments by only treating " as a quote character. This allows ' to be used naturally. A ValueError will be raised if the string couldn’t be parsed.

>>> parse_arguments("a test string")
['a', 'test', 'string']
>>> parse_arguments("apostrophes aren't a problem")
['apostrophes', "aren't", 'a', 'problem']
>>> parse_arguments('"string grouping" is useful')
['string grouping', 'is', 'useful']
>>> parse_arguments('just remember to "match your quotes')
Traceback (most recent call last):
  File "<stdin>", line 1, in ?
ValueError: No closing quotation

csbot.util.simple_http_get(url, stream=False)

A deliberately dumb wrapper around requests.get().

This should be used for the vast majority of HTTP GET requests. It turns off SSL certificate verification and sets a non-default User-Agent, thereby succeeding at most “just get the content” requests. Note that it can generate a ConnectionError exception if the url is not resolvable.

stream controls the “streaming mode” of the HTTP client, i.e. deferring the acquisition of the response body. Use this if you need to impose a maximum size or process a large response. The entire content must be consumed or ``response.close()`` must be called.

csbot.util.pairwise(iterable)

Pairs elements of an iterable together, e.g. s -> (s0,s1), (s1,s2), (s2, s3), …

csbot.util.cap_string(s, n)

If a string is longer than a particular length, it gets truncated and has ‘…’ added to the end.

csbot.util.ordinal(value)

Converts zero or a postive integer (or their string representations) to an ordinal value.

http://code.activestate.com/recipes/576888-format-a-number-as-an-ordinal/

>>> for i in range(1,13):
...     ordinal(i)
...
u'1st'
u'2nd'
u'3rd'
u'4th'
u'5th'
u'6th'
u'7th'
u'8th'
u'9th'
u'10th'
u'11th'
u'12th'

>>> for i in (100, '111', '112',1011):
...     ordinal(i)
...
u'100th'
u'111th'
u'112th'
u'1011th'

csbot.util.pluralize(n, singular, plural)

csbot.util.is_ascii(s)

Returns true if all characters in a string can be represented in ASCII.

csbot.util.maybe_future(result, *, on_error=None, log=<Logger csbot.util (WARNING)>, loop=None)

Make result a future if possible, otherwise return None.

If result is not None but also not awaitable, it is passed to on_error if supplied, otherwise logged as a warning on log.

csbot.util.truncate_utf8(b: bytes, maxlen: int, ellipsis: bytes = b'...') → bytes

Trim b to a maximum of maxlen bytes (including ellipsis if longer), without breaking UTF-8 sequences.

csbot.util.topological_sort(data: Dict[T, Set[T]]) → Iterator[Set[T]]

Get topological ordering from dependency data.

Generates sets of items with equal ordering position.

class csbot.util.RateLimited(f, *, period: float = 2.0, count: int = 5, loop=None, log=<Logger csbot.util (WARNING)>)

Bases: object

An asynchronous wrapper around calling f that is rate limited to count calls per period seconds.

Calling the rate limiter returns a future that completes with the result of calling f with the same arguments. start() and stop() control whether or not calls are actually processed.

get_delay() → float

Get number of seconds to wait before processing the next call.

start()

Start async task to process calls.

stop(clear=True)

Stop async call processing.

If clear is True (the default), any pending calls not yet processed have their futures cancelled. If it’s False, then those pending calls will still be queued when start() is called again.

Returns list of (args, kwargs) pairs of cancelled calls.

run()

csbot.util.type_validator(_obj, attrib: attr._make.Attribute, value)

An attrs validator that inspects the attribute type.

class csbot.util.PrettyStreamHandler(stream=None, colour=None)

Bases: logging.StreamHandler

Wrap log messages with severity-dependent ANSI terminal colours.

Use in place of logging.StreamHandler to have log messages coloured according to severity.

>>> handler = PrettyStreamHandler()
>>> handler.setFormatter(logging.Formatter('[%(levelname)-8s] %(message)s'))
>>> logging.getLogger('').addHandler(handler)

stream corresponds to the same argument to logging.StreamHandler, defaulting to stderr.

colour overrides TTY detection to force colour on or off.

This source for this class is released into the public domain.

Code author: Alan Briolat <alan.briolat@gmail.com>

COLOURS = {10: '\x1b[36m', 30: '\x1b[33m', 40: '\x1b[31m', 50: '\x1b[31;7m'}

Mapping from logging levels to ANSI colours.

COLOUR_END = '\x1b[0m'

ANSI code for resetting the terminal to default colour.

format(record)

Get a coloured, formatted message for a log record.

Calls logging.StreamHandler.format() and applies a colour to the message if appropriate.

csbot.util.maybe_future_result(result, **kwargs)

Get actual result from result.

If result is awaitable, return the result of awaiting it, otherwise just return result.

csbot.util.simple_http_get_async(url, **kwargs)