Welcome to csbotā€™s documentation!Ā¶

Contents:

How to write pluginsĀ¶

Anatomy of a pluginĀ¶

Plugins are automatically discovered if they match the right pattern. They must

  • subclass csbot.plugin.Plugin, and
  • live under the package specified by csbot.core.Bot.PLUGIN_PACKAGE (csbot.plugins by default).

For example, a minimal plugin that does nothing might live in csbot/plugins/nothing.py and look like:

from csbot.plugin import Plugin

class Nothing(Plugin):
    pass

A pluginā€™s name is its class name in lowercase [1] and must be unique, so plugin classes should be named meaningfully. Changing a plugin name will cause it to lose access to its associated configuration and database, so try not to do that unless youā€™re prepared to migrate these things.

The vast majority of interaction with the outside world is through subscribing to events and registering commands.

EventsĀ¶

Root events are generated when the bot receives data from the IRC server, and further events may be generated while handling an event.

All events are represented by the Event class, which is a dictionary of event-related information with some additional helpful attributes. See Events for further information on the Event class and available events.

Events are hooked with the Plugin.hook() decorator. The decorated method will be called for every event that matches the specified event_type, with the event object as the only argument. For example, a basic logging plugin that prints sent and received data:

class Logger(Plugin):
    @Plugin.hook('core.raw.sent')
    def sent(self, e):
        print('<-- ' + e['message'])

    @Plugin.hook('core.raw.received')
    def received(self, e):
        print('--> ' + e['message'])

A single handler can hook more than one event:

class MessagePrinter(Plugin):
    @Plugin.hook('core.message.privmsg')
    @Plugin.hook('core.message.notice')
    def got_message(self, e):
        """Print out all messages, ignoring if they were PRIVMSG or NOTICE."""
        print(e['message'])

CommandsĀ¶

Registering commands provides a more structured way for users to interact with a plugin. A command can be any unique, non-empty sequence of non-whitespace characters, and are invoked when prefixed with the botā€™s configured command prefix. Command events use the CommandEvent class, extending a core.message.privmsg Event and adding the arguments() method and the command and data items.

class CommandTest(Plugin):
    @Plugin.command('test')
    def hello(self, e):
        print(e['command'] + ' invoked with arguments ' + repr(e.arguments()))

A single handler can be registered for more than one command, e.g. to give aliases, and commands and hooks can be freely mixed.

class Friendly(Plugin):
    @Plugin.hook('core.channel.joined')
    @Plugin.command('hello')
    @Plugin.command('hi')
    def hello(self, e):
        e.protocol.msg(e['channel'], 'Hello, ' + nick(e['user']))

Responding: the BotProtocol objectĀ¶

In the above example the Event.protocol attribute was used to respond back to the IRC server. This attribute is an instance of BotProtocol, which subclasses twisted.words.protocols.irc.IRCClient for IRC protocol support. The documentation for IRCClient is the best place to find out what methods are supported when responding to an event or command.

ConfigurationĀ¶

Basic string key/value configuration can be stored in an INI-style file. A pluginā€™s config attribute is a shortcut to a configuration section with the same name as the plugin. The Python 3 configparser is used instead of the Python 2 ConfigParser because it supports the mapping access protocol, i.e. it acts like a dictionary in addition to supporting its own API.

An example of using plugin configuration:

class Say(Plugin):
    @Plugin.command('say')
    def say(self, e):
        if self.config.getboolean('shout', False):
            e.protocol.msg(e['reply_to'], e['data'].upper() + '!')
        else:
            e.protocol.msg(e['reply_to'], e['data'])

For even more convenience, automatic fallback values are supported through the CONFIG_DEFAULTS attribute when using the config_get() or config_getboolean() methods instead of the corresponding methods on config. This is encouraged, since it makes it clear what configuration the plugin supports and what the default values are by looking at just one part of the plugin source code. The above example would look like this:

class Say(Plugin):
    CONFIG_DEFAULTS = {
        'shout': False,
    }

    @Plugin.command('say')
    def say(self, e):
        if self.config_getboolean('shout'):
            e.protocol.msg(e['reply_to'], e['data'].upper() + '!')
        else:
            e.protocol.msg(e['reply_to'], e['data'])

Configuration can be changed at runtime, but wonā€™t be saved. This allows for temporary state changes, whilst ensuring the startup state of the bot reflects the configuration file. For example, the above plugin could be modified with a toggle for the ā€œshoutā€ mode:

class Say(Plugin):
    # ...
    @Plugin.command('toggle')
    def toggle(self, e):
        self.config['shout'] = not self.config_get('shout')

DatabaseĀ¶

The bot supports easy access to MongoDB through PyMongo. Plugins have a db attribute which is a pymongo.database.Database, unique to the plugin and created as needed. Refer to the PyMongo documentation for further guidance on using the API.

[1]This can be changed by overriding the plugin_name() class method if absolutely necessary.

EventsĀ¶

All events are represented by Event instances. Every event has the following attributes:

Event.bot = None

The Bot which triggered the event.

Event.event_type = None

The name of the event.

Event.datetime = None

The value of datetime.datetime.now() when the event was triggered.

Event instances are also dictionaries, and the keys present depend on the particular event type. The following sections describe each event, specified as event_type(keys).

Raw eventsĀ¶

These events are very low-level and most plugins shouldnā€™t need them.

core.raw.connected

Client established connection.

core.raw.disconnected

Client lost connection.

core.raw.sent(message)

Client sent message to the server.

core.raw.received(message)

Client received message from the server.

Bot eventsĀ¶

These events represent changes in the botā€™s state.

core.self.connected

IRC connection successfully established.

core.self.joined(channel)

Client joined channel.

core.self.left(channel)

Client left channel.

Message eventsĀ¶

These events occur when messages are received by the bot.

core.message.privmsg(channel, user, message, is_private, reply_to)

Received message from user which was sent to channel. If the message was sent directly to the client, i.e. channel is the clientā€™s nick and not a channel name, then is_private will be True and any response should be to user, not channel. reply_to is the channel/user any response should be sent to.

core.message.notice(channel, user, message, is_private, reply_to)

As core.message.privmsg, but representing a NOTICE rather than a PRIVMSG. Bear in mind that according to RFC 1459 ā€œautomatic replies must never be sent in response to a NOTICE messageā€ - this definitely applies to bot functionality!

core.message.action(channel, user, message, is_private, reply_to)

Received a CTCP ACTION of message from user sent to channel. Other arguments are as for core.message.privmsg.

Channel eventsĀ¶

These events occur when something about the channel changes, e.g. people joining or leaving, the topic changing, etc.

core.channel.joined(channel, user)

user joined channel.

core.channel.left(channel, user)

user left channel.

core.channel.names(channel, names, raw_names)

Received the list of users currently in the channel, in response to a NAMES command.

cores.channel.topic(channel, author, topic)

Fired whenever the channel topic is changed, and also immediately after joining a channel. The author field will usually be the server name when joining a channel (on Freenode, at least), and the nick of the user setting the topic when the topic has been changed.

User eventsĀ¶

These events occur when a user changes state in some way, i.e. actions that arenā€™t limited to a single channel.

core.user.quit(user, message)
core.user.renamed(oldnick, newnick)

csbotĀ¶

csbot packageĀ¶

SubpackagesĀ¶

csbot.plugins packageĀ¶
SubmodulesĀ¶
csbot.plugins.auth moduleĀ¶
class csbot.plugins.auth.PermissionDB[source]Ā¶

Bases: collections.defaultdict

A helper class for assembling the permissions database.

process(entity, permissions)[source]Ā¶

Process a configuration entry, where entity is an account name, @group name or * and permissions is a space-separated list of permissions to grant.

get_permissions(entity)[source]Ā¶

Get the set of permissions for entity.

The union of the permissions for entity and the universal (*) permissions is returned. If entity is None, only the universal permissions are returned.

check(entity, permission, channel=None)[source]Ā¶

Check if entity has permission.

If channel is present, check for a channel permission, otherwise check for a bot permission. Compatible wildcard permissions are also checked.

class csbot.plugins.auth.Auth(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

PLUGIN_DEPENDS = ['usertrack']Ā¶
setup()[source]Ā¶

Plugin setup.

  • Replace all ProvidedByPlugin attributes.
  • Fire all plugin integration methods.
  • Register all commands provided by the plugin.
check(nick, perm, channel=None)[source]Ā¶
check_or_error(e, perm, channel=None)[source]Ā¶
csbot.plugins.calc moduleĀ¶
csbot.plugins.calc.is_too_long(n)[source]Ā¶
csbot.plugins.calc.guarded_power(a, b)[source]Ā¶

A limited power function to make sure that commands do not take too long to process.

csbot.plugins.calc.guarded_lshift(a, b)[source]Ā¶
csbot.plugins.calc.guarded_rshift(a, b)[source]Ā¶
csbot.plugins.calc.guarded_factorial(a)[source]Ā¶
class csbot.plugins.calc.CalcEval[source]Ā¶

Bases: ast.NodeVisitor

visit_Module(node)[source]Ā¶
visit_Expr(node)[source]Ā¶
visit_BinOp(node)[source]Ā¶
visit_UnaryOp(node)[source]Ā¶
visit_Compare(node)[source]Ā¶
visit_Call(node)[source]Ā¶
visit_Name(node)[source]Ā¶
visit_Num(node)[source]Ā¶
visit_NameConstant(node)[source]Ā¶
visit_Str(node)[source]Ā¶
generic_visit(node)[source]Ā¶

Fallback visitor which always raises an exception.

We evaluate expressions by using return values of node visitors, and generic_visit() returns None, therefore if itā€™s called we know this is an expression we donā€™t support and should give an error.

exception csbot.plugins.calc.CalcError[source]Ā¶

Bases: Exception

class csbot.plugins.calc.Calc(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

A plugin that calculates things.

do_some_calc(e)[source]Ā¶

What? You donā€™t have a calculator handy?

csbot.plugins.cron moduleĀ¶
class csbot.plugins.cron.Cron(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

Time, that most mysterious of things. What is it? Is it discrete or continuous? What was before time? Does that even make sense to ask? This plugin will attempt to address some, perhaps all, of these questions.

More seriously, this plugin allows the scheduling of events. Due to computers being the constructs of fallible humans, itā€™s not guaranteed that a callback will be run precisely when you want it to be. Furthermore, if you schedule multiple events at the same time, donā€™t make any assumptions about the order in which theyā€™ll be called.

Example of usage:

class MyPlugin(Plugin):

cron = Plugin.use(ā€˜cronā€™)

def setup(self):

ā€¦ self.cron.after(

ā€œhello worldā€, datetime.timedelta(days=1), ā€œcallbackā€)
def callback(self, when):
self.log.info(uā€™I got called at {}ā€™.format(when))

@Plugin.hook(ā€˜cron.hourlyā€™) def hourlyevent(self, e):

self.log.info(uā€™An hour has passedā€™)
tasksĀ¶

Descriptor for plugin attributes that get (and cache) a value from another plugin.

See Plugin.use().

setup()[source]Ā¶

Plugin setup.

  • Replace all ProvidedByPlugin attributes.
  • Fire all plugin integration methods.
  • Register all commands provided by the plugin.
teardown()[source]Ā¶

Plugin teardown.

  • Unregister all commands provided by the plugin.
fire_event(now, name)[source]Ā¶

Fire off a regular event.

This gets called by the scheduler at the appropriate time.

provide(plugin_name)[source]Ā¶

Return the crond for the given plugin.

match_task(owner, name=None, args=None, kwargs=None)[source]Ā¶

Create a MongoDB search for a task definition.

schedule(owner, name, when, interval=None, callback=None, args=None, kwargs=None)[source]Ā¶

Schedule a new task.

Parameters:
  • owner ā€“ The plugin which created the task
  • name ā€“ The name of the task
  • when ā€“ The datetime to trigger the task at
  • interval ā€“ Optionally, reschedule at when + interval when triggered. Gives rise to repeating tasks.
  • callback ā€“ Call owner.callback when triggered; if None, call owner.name.
  • args ā€“ Callback positional arguments.
  • kwargs ā€“ Callback keyword arguments.

The signature of a task is (owner, name, args, kwargs), and trying to create a task with the same signature as an existing task will raise DuplicateTaskError. Any subset of the signature can be used to unschedule() all matching tasks (owner is mandatory).

unschedule(owner, name=None, args=None, kwargs=None)[source]Ā¶

Unschedule a task.

Removes all existing tasks that match based on the criteria passed as arguments (see match_task()).

This could result in the scheduler having nothing to do in its next call, but this isnā€™t a problem as itā€™s not a very intensive function, so thereā€™s no point in rescheduling it here.

schedule_event_runner()[source]Ā¶

Schedule the event runner.

Set up a delayed call for event_runner() to happen no sooner than is required by the next scheduled task. If a different call already exists it is replaced.

event_runner()[source]Ā¶

Run pending tasks.

Run all tasks which have a trigger time in the past, and then reschedule self to run in time for the next task.

exception csbot.plugins.cron.DuplicateTaskError[source]Ā¶

Bases: Exception

Task with a given signature already exists.

This can be raised by Cron.schedule() if a plugin tries to register two events with the same name.

class csbot.plugins.cron.PluginCron(cron, plugin)[source]Ā¶

Bases: object

Interface to the cron methods restricted to plugin as the task owner..

All of the scheduling functions have a signature of the form (name, time, method_name, *args, **kwargs).

This means that at the appropriate time, the method plugin.method_name will be called with the arguments (time, *args, **kwargs), where the time argument is the time it was supposed to be run by the scheduler (which may not be identical to teh actual time it is run).

These functions will raise a DuplicateNameException if you try to schedule two events with the same name.

schedule(name, when, interval=None, callback=None, args=None, kwargs=None)[source]Ā¶

Pass through to Cron.schedule(), adding owner argument.

after(_delay, _name, _method_name, *args, **kwargs)[source]Ā¶

Schedule an event to occur after the timedelta delay has passed.

at(_when, _name, _method_name, *args, **kwargs)[source]Ā¶

Schedule an event to occur at a given time.

every(_freq, _name, _method_name, *args, **kwargs)[source]Ā¶

Schedule an event to occur every time the delay passes.

unschedule(name, args=None, kwargs=None)[source]Ā¶

Pass through to Cron.unschedule(), adding owner argument.

unschedule_all()[source]Ā¶

Unschedule all tasks for this plugin.

This could be supported by unschedule(), but itā€™s nice to prevent code accidentally wiping all of a pluginā€™s tasks.

csbot.plugins.csyork moduleĀ¶
class csbot.plugins.csyork.CSYork(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

Amusing replacements for various #cs-york members

respond(e)[source]Ā¶
csbot.plugins.github moduleĀ¶
GitHub Deployment TrackingĀ¶

GitHubā€™s Deployments API allows a repository to track deployment activity. For example, deployments of the main instance of csbot can be seen at https://github.com/HackSoc/csbot/deployments.

Getting csbot to report deployments to your repository during bot startup requires the following:

  • SOURCE_COMMIT environment variable set to the current git revision (the Docker image has this baked in)
  • --env-name command-line option (defaults to development)
  • --github-repo command-line option with the repository to report deployments to (e.g. HackSoc/csbot)
  • --github-token command-line option with a GitHub ā€œpersonal access tokenā€ that has repo_deployment scope

Note

Deployments API functionality is implemented in csbot.cli, not here.

GitHub WebhooksĀ¶

The GitHub plugin provides a webhook endpoint that will turn incoming events into messages that are sent to IRC channels. To use the GitHub webhook, the webserver and webhook plugins must be enabled in addition to this one, and the csbot webserver must be exposed to the internet somehow.

Follow the GitHub documentation to create a webhook on the desired repository, with the following settings:

  • Payload URL: see webhook for how webhook URL routing works
  • Content type: application/json
  • Secret: the same value as chosen for the secret plugin option, for signing payloads
  • Which events ā€¦: Configure for whichever events you want to handle
ConfigurationĀ¶

The following configuration options are supported in the [github] config section:

Setting Description
secret The secret used when creating the GitHub webhook. Optional, will not verify payload signatures if unset.
notify Space-separated list of IRC channels to send messages to.
fmt/[...] Format strings to use for particular events, for turning an event into an IRC message. See below.
fmt.[...] Re-usable format string fragments. See below.

secret and notify can be overridden on a per-repository basis, in a [github/{repo}] config section, e.g. [github/HackSoc/csbot].

Event format stringsĀ¶

When writing format strings to handle GitHub webhook events, itā€™s essential to refer to the GitHub Event Types & Payloads documentation.

Each event event_type, and possibly an event_subtype. The event_type always corresponds to the ā€œWebhook event nameā€ defined by GitHubā€™s documentation, e.g. release for ReleaseEvent. The event_subtype is generally the action from the payload, if that event type has one (but see below for exceptions).

The plugin will attempt to find the most specific config option that exists to supply a format string:

  • For an event with event_type and event_subtype, will try fmt/event_type/event_subtype, fmt/event_type/* and fmt/*
  • For an event with no event_subtype, will try fmt/event_type and fmt/*

The first config option that exists will be used, and if that format string is empty (zero-length string, None, False) then no message will be sent. This means itā€™s possible to set a useful format string for fmt/issues/*, but then set an empty format for fmt/issues/labeled and fmt/issues/unlabeled to ignore some unwanted noise.

The string is formatted with the context of the entire webhook payload, plus additional keys for event_type, event_subtype and event_name (which is {event_type}/{event_subtype} if there is an event_subtype, otherwise {event_type}. (But see below for exceptions where additional context exists.)

Re-usable format stringsĀ¶

There are a lot of recurring structures in the GitHub webhook payloads, and those will usually want to be formatted in similar ways in resulting messages. For example, it might be desirable to start every message with the repository name and user that caused the event. Instead of duplicating the same fragment of format string for each event type, which makes the format strings long and hard to maintain, a format string fragment can be defined as a fmt.name config option, and referenced in another format string as {fmt.name}. These fragments will get formatted with the same context as the top-level format string.

Customised event handlingĀ¶

To represent certain events more clearly, additional processing is required, either to extend the string format context or to introduce an event_subtype where there is no action in the payload. This is the approach needed when thinking ā€œI wish string formatting had conditionalsā€. Implementing such handling is done by creating a handle_{event_type} method, which should ultimately call generic_handler with appropriate arguments.

There is already customised handling for the following:

  • push
    • Sets event_subtype: forced for forced update of a ref, and pushed for regular pushes
    • Sets count: number of commits pushed to the ref
    • Sets short_ref: only the final element of the long ref name, e.g. v1.0 from refs/tags/v1.0
  • pull_request
    • Overrides event_subtype with merged if PR was closed due to a merge
  • pull_request_review
    • Sets review_state to a human-readable version of the review state
Module contentsĀ¶
class csbot.plugins.github.GitHub(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

PLUGIN_DEPENDS = ['webhook']Ā¶
CONFIG_DEFAULTS = {'debug_payloads': False, 'fmt/*': None, 'notify': '', 'secret': ''}Ā¶
CONFIG_ENVVARS = {'secret': ['GITHUB_WEBHOOK_SECRET']}Ā¶
config_get(key, repo=None)[source]Ā¶

A special implementation of Plugin.config_get() which looks at a repo-based configuration subsection before the pluginā€™s configuration section.

classmethod find_by_matchers(matchers, d, default=<object object>)[source]Ā¶
generic_handler(data, event_type, event_subtype=None, event_subtype_key='action', context=None)[source]Ā¶
handle_pull_request(data, event_type)[source]Ā¶
handle_pull_request_review(data, event_type)[source]Ā¶
handle_push(data, event_type)[source]Ā¶
webhook(e)[source]Ā¶
class csbot.plugins.github.MessageFormatter(config_get)[source]Ā¶

Bases: string.Formatter

get_field(field_name, args, kwargs)[source]Ā¶
csbot.plugins.helix moduleĀ¶
class csbot.plugins.helix.Helix(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

The premier csbot plugin, allowing mere mortals to put questions to the mighty helix, and receive his divine wisdom.

Notes:
  • The popular online version basically just selects a random outcome, and saves it with a random url so that it can be reused if the same question is asked.
  • Iā€™m lazy, so Iā€™m just going to hash whatever the person puts in and mod the resulting value (taken from hex) to pick out an element of the outcomes list. That way if the same questions gets asked twice, it gets (hopefully) the same answer.
outcomes = ['It is certain', 'It is decidedly so', 'Without a doubt', 'Yes definitely', 'You may rely on it', 'As I see it, yes', 'Most likely', 'Outlook good', 'Yes', 'Signs point to yes', 'Reply hazy try again', 'Ask again later', 'Better not tell you now', 'Cannot predict now', 'Concentrate and ask again ', "Don't count on it", 'My reply is no', 'My sources say no', 'Outlook not so good', 'Very doubtful', 'no.', 'START', 'A', 'B', 'UP', 'DOWN', 'LEFT', 'RIGHT', 'SELECT', 'START', 'A', 'B', 'UP', 'DOWN', 'LEFT', 'RIGHT', 'SELECT']Ā¶
setup()[source]Ā¶

Plugin setup.

  • Replace all ProvidedByPlugin attributes.
  • Fire all plugin integration methods.
  • Register all commands provided by the plugin.
ask_the_almighty_helix(e)[source]Ā¶

Ask and you shall recieve.

csbot.plugins.hoogle moduleĀ¶
class csbot.plugins.hoogle.Hoogle(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

CONFIG_DEFAULTS = {'results': 5}Ā¶
setup()[source]Ā¶

Plugin setup.

  • Replace all ProvidedByPlugin attributes.
  • Fire all plugin integration methods.
  • Register all commands provided by the plugin.
search_hoogle(e)[source]Ā¶

Search Hoogle with a given string and return the first few (exact number configurable) results.

csbot.plugins.imgur moduleĀ¶
exception csbot.plugins.imgur.ImgurError[source]Ā¶

Bases: Exception

class csbot.plugins.imgur.Imgur(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

CONFIG_DEFAULTS = {'client_id': None, 'client_secret': None}Ā¶
CONFIG_ENVVARS = {'client_id': ['IMGUR_CLIENT_ID'], 'client_secret': ['IMGUR_CLIENT_SECRET']}Ā¶
integrate_with_linkinfo(linkinfo)[source]Ā¶
csbot.plugins.last moduleĀ¶
class csbot.plugins.last.Last(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

Utility plugin to record the last message (and time said) of a user. Records both messages and actions individually, and allows querying on either.

dbĀ¶

Descriptor for plugin attributes that get (and cache) a value from another plugin.

See Plugin.use().

last(nick, channel=None, msgtype=None)[source]Ā¶

Get the last thing said (including actions) by a given nick, optionally filtering by channel.

last_message(nick, channel=None)[source]Ā¶

Get the last message sent by a nick, optionally filtering by channel.

last_action(nick, channel=None)[source]Ā¶

Get the last action sent by a nick, optionally filtering by channel.

last_command(nick, channel=None)[source]Ā¶

Get the last command sent by a nick, optionally filtering by channel.

record_message(event)[source]Ā¶

Record the receipt of a new message.

record_command(event)[source]Ā¶

Record the receipt of a new command.

record_action(event)[source]Ā¶

Record the receipt of a new action.

record(event, nick, channel, msgtype, msg)[source]Ā¶

Record a new message, of a given type.

show_seen(event)[source]Ā¶
csbot.plugins.linkinfo moduleĀ¶
class csbot.plugins.linkinfo.LinkInfoHandler(filter: Callable[[urllib.parse.ParseResult], LinkInfoFilterResult], handler: Callable[[urllib.parse.ParseResult, LinkInfoFilterResult], Optional[LinkInfoResult]], exclusive: bool)[source]Ā¶

Bases: typing.Generic

class csbot.plugins.linkinfo.LinkInfoResult(url: str, text: str, is_error: bool = False, nsfw: bool = False, is_redundant: bool = False)[source]Ā¶

Bases: object

urlĀ¶

The URL requested

textĀ¶

Information about the URL

is_errorĀ¶

Is an error?

nsfwĀ¶

URL is not safe for work?

is_redundantĀ¶

URL information is redundant? (e.g. duplicated in URL string)

get_message()[source]Ā¶
class csbot.plugins.linkinfo.LinkInfo(*args, **kwargs)[source]Ā¶

Bases: csbot.plugin.Plugin

class Config(raw_data=None, trusted_data=None, deserialize_mapping=None, init=True, partial=True, strict=True, validate=False, app_data=None, lazy=False, **kwargs)[source]Ā¶

Bases: csbot.config.Config

scan_limit = <IntType() instance on Config as 'scan_limit'>Ā¶
minimum_slug_length = <IntType() instance on Config as 'minimum_slug_length'>Ā¶
max_file_ext_length = <IntType() instance on Config as 'max_file_ext_length'>Ā¶
minimum_path_match = <FloatType() instance on Config as 'minimum_path_match'>Ā¶
rate_limit_time = <IntType() instance on Config as 'rate_limit_time'>Ā¶
rate_limit_count = <IntType() instance on Config as 'rate_limit_count'>Ā¶
max_response_size = <IntType() instance on Config as 'max_response_size'>Ā¶
register_handler(filter, handler, exclusive=False)[source]Ā¶

Add a URL handler.

filter should be a function that returns a True-like or False-like value to indicate whether handler should be run for a particular URL. The URL is supplied as a urlparse:ParseResult instance.

If handler is called, it will be as handler(url, filter(url)). The filter result is useful for accessing the results of a regular expression filter, for example. The result should be a LinkInfoResult instance. If the result is None instead, the processing will fall through to the next handler; this is the best way to signal that a handler doesnā€™t know what to do with a particular URL.

If exclusive is True, the fall-through behaviour will not happen, instead terminating the handling with the result of calling handler.

register_exclude(filter)[source]Ā¶

Add a URL exclusion filter.

filter should be a function that returns a True-like or False-like value to indicate whether or not a URL should be excluded from the default title-scraping behaviour (after all registered handlers have been tried). The URL is supplied as a urlparse.ParseResult instance.

Get information about a URL.

Using the original_url string, run the chain of URL handlers and excludes to get a LinkInfoResult.

Handle the ā€œlinkā€ command.

Fetch information about a specified URL, e.g. !link http://google.com. The link can be explicitly marked as NSFW by including the string anywhere in the trailing string, e.g. !link http://lots-of-porn.com nsfw.

scan_privmsg(e)[source]Ā¶

Scan the data of PRIVMSG events for URLs and respond with information about them.

scrape_html_title(url)[source]Ā¶

Scrape the <title> tag contents from the HTML page at url.

Returns a LinkInfoResult.

csbot.plugins.logger moduleĀ¶
class csbot.plugins.logger.Logger(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

raw_log = <Logger csbot.raw_log (WARNING)>Ā¶
pretty_log = <Logger csbot.pretty_log (WARNING)>Ā¶
raw_received(event)[source]Ā¶
raw_sent(event)[source]Ā¶
connected(event)[source]Ā¶
disconnected(event)[source]Ā¶
signedon(event)[source]Ā¶
joined(event)[source]Ā¶
left(event)[source]Ā¶
user_joined(event)[source]Ā¶
user_left(event)[source]Ā¶
names(event)[source]Ā¶
topic(event)[source]Ā¶
privmsg(event)[source]Ā¶
notice(event)[source]Ā¶
action(event)[source]Ā¶
quit(event)[source]Ā¶
renamed(event)[source]Ā¶
command(event)[source]Ā¶

Tag a command to be registered by setup().

Additional keyword arguments are added to a metadata dictionary that gets stored with the command. This is a good place to put, for example, the help string for the command:

@Plugin.command('foo', help='foo: does something amazing')
def foo_command(self, e):
    pass
csbot.plugins.mongodb moduleĀ¶
class csbot.plugins.mongodb.MongoDB(*args, **kwargs)[source]Ā¶

Bases: csbot.plugin.Plugin

A plugin that provides access to a MongoDB server via pymongo.

CONFIG_DEFAULTS = {'mode': 'uri', 'uri': 'mongodb://localhost:27017/csbot'}Ā¶
CONFIG_ENVVARS = {'uri': ['MONGOLAB_URI', 'MONGODB_URI']}Ā¶
provide(plugin_name, collection)[source]Ā¶

Get a MongoDB collection for {plugin_name}__{collection}.

csbot.plugins.termdates moduleĀ¶
class csbot.plugins.termdates.Term(key: str, start_date: datetime.datetime)[source]Ā¶

Bases: object

first_mondayĀ¶
last_fridayĀ¶
get_week_number(date: datetime.date) → int[source]Ā¶

Get the ā€œterm week numberā€ of a date relative to this term.

The first week of term is week 1, not week 0. Week 1 starts at the Monday of the termā€™s start date, even if the termā€™s start date is not Monday. Any date before the start of the term gives a negative week number.

get_week_start(week_number: int) → datetime.datetime[source]Ā¶

Get the start date of a specific week number relative to this term.

The first week of term is week 1, not week 0, although this method allows both. When referring to the first week of term, the start date is the term start date (which may not be a Monday). All other weeks start on their Monday.

class csbot.plugins.termdates.TermDates(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

A wonderful plugin allowing old people (graduates) to keep track of the ever-changing calendar.

DATE_FORMAT = '%Y-%m-%d'Ā¶
TERM_KEYS = ('aut', 'spr', 'sum')Ā¶
db_termsĀ¶

Descriptor for plugin attributes that get (and cache) a value from another plugin.

See Plugin.use().

terms = NoneĀ¶
setup()[source]Ā¶

Plugin setup.

  • Replace all ProvidedByPlugin attributes.
  • Fire all plugin integration methods.
  • Register all commands provided by the plugin.
initialisedĀ¶

If no term dates have been set, the calendar is uninitialised and canā€™t be asked about term thing.

termdates(e)[source]Ā¶
week(e)[source]Ā¶
termdates_set(e)[source]Ā¶
csbot.plugins.topic moduleĀ¶
class csbot.plugins.topic.Topic(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

PLUGIN_DEPENDS = ['auth']Ā¶
CONFIG_DEFAULTS = {'end': '', 'history': 5, 'sep': '|', 'start': ''}Ā¶
setup()[source]Ā¶

Plugin setup.

  • Replace all ProvidedByPlugin attributes.
  • Fire all plugin integration methods.
  • Register all commands provided by the plugin.
config_get(key, channel=None)[source]Ā¶

A special implementation of Plugin.config_get() which looks at a channel-based configuration subsection before the pluginā€™s configuration section.

topic_changed(e)[source]Ā¶
topic(e)[source]Ā¶
topic_history(e)[source]Ā¶
topic_undo(e)[source]Ā¶
topic_append(e)[source]Ā¶
topic_pop(e)[source]Ā¶
topic_replace(e)[source]Ā¶
topic_insert(e)[source]Ā¶
csbot.plugins.usertrack moduleĀ¶
class csbot.plugins.usertrack.UserDict[source]Ā¶

Bases: collections.defaultdict

static create_user(nick)[source]Ā¶
copy_or_create(nick)[source]Ā¶
class csbot.plugins.usertrack.UserTrack(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

setup()[source]Ā¶

Plugin setup.

  • Replace all ProvidedByPlugin attributes.
  • Fire all plugin integration methods.
  • Register all commands provided by the plugin.
get_user(nick)[source]Ā¶

Get a copy of the user record for nick.

account_command(e)[source]Ā¶
csbot.plugins.webhook moduleĀ¶

Uses webserver to create a generic URL for incoming webhooks so that other plugins can handle webhook events.

To act as a webhook handler, a plugin should hook the webhook.{service} event, for example:

class MyPlugin(Plugin):
    @Plugin.hook('webhook.myplugin')
    async def webhook(self, e):
        self.log.info(f'Handling {e["request"]}')

The request key of the event contains the aiohttp.web.Request object.

Note

The webhook plugin only responds to POST requests.

ConfigurationĀ¶

The following configuration options are supported in the [webhook] config section:

Setting Description
prefix URL prefix for the web server sub-application. Default: /webhook.
url_secret Extra URL component to make valid endpoints hard to guess.
URL Format & Request HandlingĀ¶

The URL path for a webhook is {prefix}/{service}/{url_secret}. The host and port elements, plus any additional prefix, are determined by the webserver plugin and/or any reverse-proxy that is in front of it.

For example, the main deployment of csbot received webhooks at https://{host}/csbot/webhook/{service}/{url_secret} and sits behind nginx with the following configuration:

location /csbot/ {
    proxy_pass http://localhost:8180/;
}
Module contentsĀ¶
class csbot.plugins.webhook.Webhook(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

CONFIG_DEFAULTS = {'prefix': '/webhook', 'url_secret': ''}Ā¶
CONFIG_ENVVARS = {'url_secret': ['WEBHOOK_SECRET']}Ā¶
create_app(e)[source]Ā¶
request_handler(request)[source]Ā¶
csbot.plugins.webserver moduleĀ¶

Creates a web server using aiohttp so that other plugins can register URL handlers.

To register a URL handler, a plugin should hook the webserver.build event and create a sub-application, for example:

class MyPlugin(Plugin):
    @Plugin.hook('webserver.build')
    def create_app(self, e):
        with e['webserver'].create_subapp('/my_plugin') as app:
            app.add_routes([web.get('/{item}', self.request_handler)])

    async def request_handler(self, request):
        return web.Response(text=f'No {request.match_info["item"]} here, oh dear!')
ConfigurationĀ¶

The following configuration options are supported in the [webserver] config section:

Setting Description
host Hostname/IP address to listen on. Default: localhost.
port Port to listen on. Default: 1337.
Module contentsĀ¶
class csbot.plugins.webserver.WebServer(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

CONFIG_DEFAULTS = {'host': 'localhost', 'port': 1337}Ā¶
setup()[source]Ā¶

Plugin setup.

  • Replace all ProvidedByPlugin attributes.
  • Fire all plugin integration methods.
  • Register all commands provided by the plugin.
teardown()[source]Ā¶

Plugin teardown.

  • Unregister all commands provided by the plugin.
create_subapp(prefix)[source]Ā¶
csbot.plugins.whois moduleĀ¶
class csbot.plugins.whois.Whois(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

Associate data with a user and a channel. Users can update their own data, and it persists over nick changes.

PLUGIN_DEPENDS = ['usertrack']Ā¶
whoisdbĀ¶

Descriptor for plugin attributes that get (and cache) a value from another plugin.

See Plugin.use().

whois_lookup(nick, channel, db=None)[source]Ā¶

Performs a whois lookup for a nick

whois_set(nick, whois_str, channel=None, db=None)[source]Ā¶
whois_unset(nick, channel=None, db=None)[source]Ā¶
whois(e)[source]Ā¶

Look up a user by nick, and return what data they have set for themselves (or an error message if there is no data)

setdefault(e)[source]Ā¶
set(e)[source]Ā¶

Allow a user to associate data with themselves for this channel.

unset(e)[source]Ā¶
unsetdefault(e)[source]Ā¶
identify_user(nick, channel=None)[source]Ā¶

Identify a user: by account if authed, if not, by nick. Produces a dict suitable for throwing at mongo.

csbot.plugins.xkcd moduleĀ¶
csbot.plugins.xkcd.fix_json_unicode(data)[source]Ā¶

Attempts to fix the unicode & HTML silliness that is included in the json data. Why Randall, Why?

class csbot.plugins.xkcd.xkcd(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

A plugin that does some xkcd things. Based on williebot xkcd plugin.

exception XKCDError[source]Ā¶

Bases: Exception

randall_is_awesome(e)[source]Ā¶

Well, Randall sucks at unicode actually :(

linkinfo_integrate(linkinfo)[source]Ā¶

Handle recognised xkcd urls.

csbot.plugins.xkcd.get_info(number=None)[source]Ā¶

Gets the json data for a particular comic (or the latest, if none provided).

csbot.plugins.youtube moduleĀ¶
csbot.plugins.youtube.get_yt_id(url)[source]Ā¶

Gets the video ID from a urllib ParseResult object.

exception csbot.plugins.youtube.YoutubeError(http_error)[source]Ā¶

Bases: Exception

Signifies some error occurred accessing the Youtube API.

This is only used for actual errors, e.g. invalid API key, not failure to find any data matching a query.

Pass the HttpError from the API call as an argument.

class csbot.plugins.youtube.Youtube(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

A plugin that does some youtube things. Based on williebot youtube plugin.

CONFIG_DEFAULTS = {'api_key': ''}Ā¶
CONFIG_ENVVARS = {'api_key': ['YOUTUBE_DATA_API_KEY']}Ā¶
RESPONSE = '"{title}" [{duration}] (by {uploader} at {uploaded}) | Views: {views}'Ā¶
CMD_RESPONSE = '"{title}" [{duration}] (by {uploader} at {uploaded}) | Views: {views} | {link}'Ā¶
all_hail_our_google_overlords(e)[source]Ā¶

I for one, welcome our Google overlords.

get_video_json(id)[source]Ā¶
linkinfo_integrate(linkinfo)[source]Ā¶

Handle recognised youtube urls.

Module contentsĀ¶

SubmodulesĀ¶

csbot.cli moduleĀ¶
csbot.cli.load_ini(f)[source]Ā¶
csbot.cli.load_json(f)[source]Ā¶
csbot.cli.load_toml(f)[source]Ā¶
csbot.cli.github_report_deploy(github_token, github_repo, env_name, revision)[source]Ā¶
csbot.cli.rollbar_report_deploy(rollbar_token, env_name, revision)[source]Ā¶
csbot.config moduleĀ¶
class csbot.config.Config(raw_data=None, trusted_data=None, deserialize_mapping=None, init=True, partial=True, strict=True, validate=False, app_data=None, lazy=False, **kwargs)[source]Ā¶

Bases: schematics.deprecated.Model

Base class for configuration schemas.

Use option(), option_list() and option_map() to create fields in the schema. Schemas are also valid option types, so deeper structures can be defined.

>>> class MyConfig(Config):
...     delay = option(float, default=0.5, help="Number of seconds to wait")
...     notify = option_list(str, help="Users to notify")
csbot.config.ConfigErrorĀ¶

alias of schematics.exceptions.DataError

csbot.config.example_mode()[source]Ā¶

For the duration of this context manager, try to use example values before default values.

class csbot.config.WordList(min_size=None, max_size=None, **kwargs)[source]Ā¶

Bases: schematics.types.compound.ListType

A list of strings that also accepts a space-separated string instead.

convert(value, context=None)[source]Ā¶
MESSAGES = {'choices': <schematics.translator.LazyText object>, 'required': <schematics.translator.LazyText object>}Ā¶
csbot.config.is_config(obj: Any) → bool[source]Ā¶

Is obj a configuration class or instance?

csbot.config.is_allowable_type(cls: Type[CT_co]) → bool[source]Ā¶

Is cls allowed as a configuration option type?

csbot.config.structure(data: Mapping[str, Any], cls: Type[csbot.config.Config]) → csbot.config.Config[source]Ā¶

Create an instance of cls from plain Python structure data.

csbot.config.unstructure(obj: csbot.config.Config) → Mapping[str, Any][source]Ā¶

Get plain Python structured data from obj.

csbot.config.loads(s: str, cls: Type[csbot.config.Config]) → csbot.config.Config[source]Ā¶

Create an instance of cls from the TOML in s.

csbot.config.dumps(obj: csbot.config.Config) → str[source]Ā¶

Get TOML string representation of obj.

csbot.config.load(f: TextIO, cls: Type[csbot.config.Config]) → csbot.config.Config[source]Ā¶

Create an instance of cls from the TOML in f.

csbot.config.dump(obj: csbot.config.Config, f: TextIO)[source]Ā¶

Write TOML representation of obj to f.

csbot.config.option(cls: Type[_B], *, required: bool = None, default: Union[None, _B, Callable[[], Union[None, _B]]] = None, example: Union[None, _B, Callable[[], Union[None, _B]]] = None, env: Union[str, List[str]] = None, help: str)[source]Ā¶

Create a configuration option that contains a value of type cls.

Parameters:
  • cls ā€“ Option type (see is_allowable_type())
  • required ā€“ A non-None value is required? (default: False if default is None, otherwise True)
  • default ā€“ Default value if no value is supplied (default: None)
  • example ā€“ Default value when generating example configuration (default: None)
  • env ā€“ Environment variables to try if no value is supplied, before using default (default: [])
  • help ā€“ Description of option, included when generating example configuration
csbot.config.option_list(cls: Type[_B], *, default: Union[None, List[_B], Callable[[], Union[None, List[_B]]]] = None, example: Union[None, List[_B], Callable[[], Union[None, List[_B]]]] = None, help: str)[source]Ā¶

Create a configuration option that contains a list of cls values.

Parameters:
  • cls ā€“ Option type (see is_allowable_type())
  • default ā€“ Default value if no value is supplied (default: empty list)
  • example ā€“ Default value when generating example configuration (default: empty list)
  • help ā€“ Description of option, included when generating example configuration
csbot.config.option_map(cls: Type[_B], *, default: Union[None, Dict[str, _B], Callable[[], Union[None, Dict[str, _B]]]] = None, example: Union[None, Dict[str, _B], Callable[[], Union[None, Dict[str, _B]]]] = None, help: str)[source]Ā¶

Create a configuration option that contains a mapping of string keys to cls values.

Parameters:
  • cls ā€“ Option type (see is_allowable_type())
  • default ā€“ Default value if no value is supplied (default: empty list)
  • example ā€“ Default value when generating example configuration (default: empty list)
  • help ā€“ Description of option, included when generating example configuration
csbot.config.make_example(cls: Type[csbot.config.Config]) → csbot.config.Config[source]Ā¶

Create an instance of cls without supplying data, using ā€œexampleā€ or ā€œdefaultā€ values for each option.

class csbot.config.TomlExampleGenerator(*, commented=False)[source]Ā¶

Bases: object

generate(obj: Union[csbot.config.Config, Type[csbot.config.Config]], stream: TextIO, prefix: List[str] = None)[source]Ā¶

Generate an example from obj and write it to stream.

csbot.config.generate_toml_example(obj: Union[csbot.config.Config, Type[csbot.config.Config]], commented: bool = False) → str[source]Ā¶

Generate an example configuration from obj as a TOML string.

csbot.core moduleĀ¶
exception csbot.core.PluginError[source]Ā¶

Bases: Exception

class csbot.core.Bot(config=None, *, plugins: Sequence[Type[csbot.plugin.Plugin]] = None, loop=None)[source]Ā¶

Bases: csbot.plugin.SpecialPlugin, csbot.irc.IRCClient

class Config(raw_data=None, trusted_data=None, deserialize_mapping=None, init=True, partial=True, strict=True, validate=False, app_data=None, lazy=False, **kwargs)[source]Ā¶

Bases: csbot.config.Config

ircv3 = <BooleanType() instance on Config as 'ircv3'>Ā¶
nickname = <StringType() instance on Config as 'nickname'>Ā¶
username = <StringType() instance on Config as 'username'>Ā¶
realname = <StringType() instance on Config as 'realname'>Ā¶
auth_method = <StringType() instance on Config as 'auth_method'>Ā¶
password = <StringType() instance on Config as 'password'>Ā¶
irc_host = <StringType() instance on Config as 'irc_host'>Ā¶
irc_port = <IntType() instance on Config as 'irc_port'>Ā¶
command_prefix = <StringType() instance on Config as 'command_prefix'>Ā¶
channels = <WordList(StringType) instance on Config as 'channels'>Ā¶
plugins = <WordList(StringType) instance on Config as 'plugins'>Ā¶
use_notice = <IntType() instance on Config as 'use_notice'>Ā¶
client_ping = <IntType() instance on Config as 'client_ping'>Ā¶
bind_addr = <StringType() instance on Config as 'bind_addr'>Ā¶
rate_limit_period = <IntType() instance on Config as 'rate_limit_period'>Ā¶
rate_limit_count = <IntType() instance on Config as 'rate_limit_count'>Ā¶
available_plugins = NoneĀ¶

Dictionary containing available plugins for loading, using straight.plugin to discover plugin classes under a namespace.

bot_setup()[source]Ā¶

Load plugins defined in configuration and run setup methods.

bot_teardown()[source]Ā¶

Run plugin teardown methods.

post_event(event)[source]Ā¶
register_command(cmd, metadata, f, tag=None)[source]Ā¶
unregister_command(cmd, tag=None)[source]Ā¶
unregister_commands(tag)[source]Ā¶
signedOn(event)[source]Ā¶
privmsg(event)[source]Ā¶

Handle commands inside PRIVMSGs.

show_commands(e)[source]Ā¶
show_plugins(e)[source]Ā¶
emit_new(event_type, data=None)[source]Ā¶

Shorthand for firing a new event.

emit(event)[source]Ā¶

Shorthand for firing an existing event.

line_sent(line: str)[source]Ā¶

Callback for sent raw IRC message.

Subclasses can implement this to get access to the actual message that was sent (which may have been truncated from what was passed to send_line()).

line_received(line)[source]Ā¶

Callback for received raw IRC message.

recent_messagesĀ¶
on_welcome()[source]Ā¶

Successfully signed on to the server.

on_joined(channel)[source]Ā¶

Joined a channel.

on_left(channel)[source]Ā¶

Left a channel.

on_privmsg(user, channel, message)[source]Ā¶

Received a message, either directly or in a channel.

on_notice(user, channel, message)[source]Ā¶

Received a notice, either directly or in a channel.

on_action(user, channel, message)[source]Ā¶

Received CTCP ACTION. Common enough to deserve its own event.

on_user_joined(user, channel)[source]Ā¶

User joined a channel.

on_user_left(user, channel, message)[source]Ā¶

User left a channel.

on_user_quit(user, message)[source]Ā¶

User disconnected.

on_user_renamed(oldnick, newnick)[source]Ā¶

User changed nick.

on_topic_changed(user, channel, topic)[source]Ā¶

user changed the topic of channel to topic.

irc_RPL_NAMREPLY(msg)[source]Ā¶
irc_RPL_ENDOFNAMES(msg)[source]Ā¶
on_names(channel, names, raw_names)[source]Ā¶

Called when the NAMES list for a channel has been received.

identify(target)[source]Ā¶

Find the account for a user or all users in a channel.

connection_lost(exc)[source]Ā¶

Handle a broken connection by attempting to reconnect.

Wonā€™t reconnect if the broken connection was deliberate (i.e. close() was called).

connection_made()[source]Ā¶

Callback for successful connection.

Register with the IRC server.

fire_command(event)[source]Ā¶

Dispatch a command event to its callback.

irc_354(msg)[source]Ā¶

Handle ā€œformatted WHOā€ responses.

on_user_identified(user, account)[source]Ā¶
irc_ACCOUNT(msg)[source]Ā¶

Account change notification from account-notify capability.

irc_JOIN(msg)[source]Ā¶

Re-implement JOIN handler to account for extended-join info.

reply(to, message)[source]Ā¶

Reply to a nick/channel.

This is not implemented because it should be replaced in the constructor with a reference to a real method, e.g. self.reply = self.msg.

classmethod write_example_config(f, plugins=None, commented=False)[source]Ā¶
csbot.events moduleĀ¶
class csbot.events.HybridEventRunner(get_handlers, loop=None)[source]Ā¶

Bases: object

A hybrid synchronous/asynchronous event runner.

get_handlers is called for each event passed to post_event(), and should return an iterable of callables to handle that event, each of which will be called with the event object.

Events are processed in the order they are received, with all handlers for an event being called before the handlers for the next event. If a handler returns an awaitable, it is added to a set of asynchronous tasks to wait on.

The future returned by post_event() completes only when all events have been processed and all asynchronous tasks have completed.

Parameters:
  • get_handlers ā€“ Get functions to call for an event
  • loop ā€“ asyncio event loop to use (default: use current loop)
post_event(event)[source]Ā¶

Post event to be handled soon.

event is added to the queue of events.

Returns a future which resolves when the handlers of event (and all events generated during those handlers) have completed.

class csbot.events.Event(bot, event_type, data=None)[source]Ā¶

Bases: dict

IRC event information.

Events are dicts of event information, plus some attributes which are applicable for all events.

bot = NoneĀ¶

The Bot which triggered the event.

event_type = NoneĀ¶

The name of the event.

datetime = NoneĀ¶

The value of datetime.datetime.now() when the event was triggered.

classmethod extend(event, event_type=None, data=None)[source]Ā¶

Create a new event by extending an existing event.

The main purpose of this classmethod is to duplicate an event as a new event type, preserving existing information. For example:

reply(message)[source]Ā¶

Send a reply.

For messages that have a reply_to key, instruct the bot to send a reply.

class csbot.events.CommandEvent(bot, event_type, data=None)[source]Ā¶

Bases: csbot.events.Event

classmethod parse_command(event, prefix, nick)[source]Ā¶

Attempt to create a CommandEvent from a core.message.privmsg event.

A command is signified by event[ā€œmessageā€] starting with the command prefix string followed by one or more non-space characters.

Returns None if event[ā€˜messageā€™] wasnā€™t recognised as being a command.

arguments()[source]Ā¶

Parse self[ā€œdataā€] into a list of arguments using parse_arguments(). This might raise a ValueError if the string cannot be parsed, e.g. if there are unmatched quotes.

csbot.irc moduleĀ¶
exception csbot.irc.IRCParseError[source]Ā¶

Bases: Exception

Raised by IRCMessage.parse() when a message canā€™t be parsed.

class csbot.irc.IRCMessage(raw: str, prefix: Optional[str], command: str, params: List[str], command_name: str)[source]Ā¶

Bases: object

Represents an IRC message.

The IRC message format, paraphrased and simplified from RFC2812, is:

message = [":" prefix " "] command {" " parameter} [" :" trailing]

Has the following attributes:

Parameters:
  • raw (str) ā€“ The raw IRC message
  • prefix (str or None) ā€“ Prefix part of the message, usually the origin
  • command (str) ā€“ IRC command
  • params (list of str) ā€“ List of command parameters (including trailing)
  • command_name (str) ā€“ Name of IRC command (see below)

The command_name attribute is intended to be the ā€œreadableā€ form of the command. Usually it will be the same as command, but numeric replies recognised in RFC2812 will have their corresponding name instead.

rawĀ¶
prefixĀ¶
commandĀ¶
paramsĀ¶
command_nameĀ¶
REGEX = re.compile('(:(?P<prefix>\\S+) )?(?P<command>\\S+)(?P<params>( (?!:)\\S+)*)( :(?P<trailing>.*))?')Ā¶

Regular expression to extract message components from a message.

FORCE_TRAILING = {'PRIVMSG', 'QUIT', 'USER'}Ā¶

Commands to force trailing parameter (:blah) for

classmethod parse(line)[source]Ā¶

Create an IRCMessage object by parsing a raw message.

classmethod create(command, params=None, prefix=None)[source]Ā¶

Create an IRCMessage from its core components.

The raw and command_name attributes will be generated based on the message details.

prettyĀ¶

Get a more readable version of the raw IRC message.

Pretty much identical to the raw IRC message, but numeric commands that have names end up being NUMERIC/NAME.

pad_params(length, default=None)[source]Ā¶

Pad parameters to length with default.

Useful when a command has optional parameters:

>>> msg = IRCMessage.parse(':nick!user@host KICK #channel other')
>>> channel, nick, reason = msg.params
Traceback (most recent call last):
  ...
ValueError: need more than 2 values to unpack
>>> channel, nick, reason = msg.pad_params(3)
class csbot.irc.IRCUser(raw: str, nick: str, user: Optional[str], host: Optional[str])[source]Ā¶

Bases: object

Provide access to the parts of an IRC user string.

The following parts of the user string are available, set to None if that part of the string is absent:

Parameters:
  • raw ā€“ Raw user string
  • nick ā€“ Nick of the user
  • user ā€“ Username of the user (excluding leading ~)
  • host ā€“ Hostname of the user
>>> IRCUser.parse('my_nick!some_user@host.name')
IRCUser(raw='my_nick!some_user@host.name', nick='my_nick', user='some_user', host='host.name')
rawĀ¶
nickĀ¶
userĀ¶
hostĀ¶
REGEX = re.compile('(?P<raw>(?P<nick>[^!]+)(!~*(?P<user>[^@]+))?(@(?P<host>.+))?)')Ā¶

Username parsing regex. Stripping out the ā€œ~ā€ might be a Freenode peculiarityā€¦

classmethod parse(raw)[source]Ā¶

Create an IRCUser from a raw user string.

class csbot.irc.IRCCodec[source]Ā¶

Bases: codecs.Codec

The encoding scheme to use for IRC messages.

IRC messages are ā€œjust bytesā€ with no encoding made explicit in the protocol definition or the messages. Ideally weā€™d like to handle IRC messages as proper strings.

encode(input, errors='strict')[source]Ā¶

Encode a message as UTF-8.

decode(input, errors='strict')[source]Ā¶

Decode a message.

IRC messages could pretty much be in any encoding. Here we just try the two most likely candidates: UTF-8, falling back to CP1252. Unfortunately, any encoding where every byte is valid (e.g. CP1252) makes it impossible to detect encoding errors - if input isnā€™t UTF-8 or CP1252-compatible, the result might be a bit odd.

exception csbot.irc.IRCClientError[source]Ā¶

Bases: Exception

class csbot.irc.IRCClient(*, loop=None, **kwargs)[source]Ā¶

Bases: object

Internet Relay Chat client protocol.

A line-oriented protocol for communicating with IRC servers. It handles receiving data at several layers of abstraction:

It also handles sending data at several layers of abstraction:

  • send_line(): raw IRC command, e.g. self.send_line('JOIN #cs-york-dev')
  • send(): IRCMessage, e.g. self.send(IRCMessage.create('JOIN', params=['#cs-york-dev']))
  • <action>(...): e.g. self.join('#cs-york-dev').

The API and implementation is inspired by irc3 and Twisted.

  • TODO: NAMES
  • TODO: MODE
  • TODO: More sophisticated CTCP? (see Twisted)
  • TODO: MOTD?
  • TODO: SSL
codec = <csbot.irc.IRCCodec object>Ā¶

Codec for encoding/decoding IRC messages.

static DEFAULTS()Ā¶

Generate a default configuration. Easier to call this and update the result than relying on dict.copy().

available_capabilities = NoneĀ¶

Available client capabilities

enabled_capabilities = NoneĀ¶

Enabled client capabilities

disconnect()[source]Ā¶

Disconnect from the IRC server.

Use quit() for a more graceful disconnect.

line_received(line: str)[source]Ā¶

Callback for received raw IRC message.

line_sent(line: str)[source]Ā¶

Callback for sent raw IRC message.

Subclasses can implement this to get access to the actual message that was sent (which may have been truncated from what was passed to send_line()).

message_received(msg)[source]Ā¶

Callback for received parsed IRC message.

send_line(data: str)[source]Ā¶

Send a raw IRC message to the server.

Encodes, terminates and sends data to the server. If the line would be longer than the maximum allowed by the IRC specification, it is trimmed to fit (without breaking UTF-8 sequences).

If rate limiting is enabled, the message may not be sent immediately.

send(msg)[source]Ā¶

Send an IRCMessage.

class Waiter(predicate: Callable[[csbot.irc.IRCMessage], Tuple[bool, Any]], future: _asyncio.Future)[source]Ā¶

Bases: object

PredicateType = typing.Callable[[csbot.irc.IRCMessage], typing.Tuple[bool, typing.Any]]Ā¶
wait_for_message(predicate: Callable[[csbot.irc.IRCMessage], Tuple[bool, Any]]) → _asyncio.Future[source]Ā¶

Wait for a message that matches predicate.

predicate should return a (did_match, result) tuple, where did_match is a boolean indicating if the message is a match, and result is the value to return.

Returns a future that is resolved with result on the first matching message.

process_wait_for_message(msg)[source]Ā¶
request_capabilities(*, enable: Iterable[str] = None, disable: Iterable[str] = None) → Awaitable[bool][source]Ā¶

Request a change to the enabled IRCv3 capabilities.

enable and disable are sets of capability names, with disable taking precedence.

Returns a future which resolves with True if the request is successful, or False otherwise.

set_nick(nick)[source]Ā¶

Ask the server to set our nick.

join(channel)[source]Ā¶

Join a channel.

leave(channel, message=None)[source]Ā¶

Leave a channel, with an optional message.

quit(message=None, reconnect=False)[source]Ā¶

Leave the server.

If reconnect is False, then the client will not attempt to reconnect after the server closes the connection.

msg(to, message)[source]Ā¶

Send message to a channel/nick.

act(to, action)[source]Ā¶

Send action as a CTCP ACTION to a channel/nick.

notice(to, message)[source]Ā¶

Send message as a NOTICE to a channel/nick.

set_topic(channel, topic)[source]Ā¶

Try and set a channelā€™s topic.

get_topic(channel)[source]Ā¶

Ask server to send the topic for channel.

Will cause on_topic_changed() at some point in the future.

ctcp_query(to, command, data=None)[source]Ā¶

Send CTCP query.

ctcp_reply(to, command, data=None)[source]Ā¶

Send CTCP reply.

irc_RPL_WELCOME(msg)[source]Ā¶

Received welcome from server, now we can start communicating.

Welcome should include the accepted nick as the first parameter. This may be different to the nick we requested (e.g. truncated to a maximum length); if this is the case we store the new nick and fire the on_nick_changed() event.

irc_ERR_NICKNAMEINUSE(msg)[source]Ā¶

Attempted nick is in use, try another.

Adds an underscore to the end of the current nick. If the server truncated the nick, replaces the last non-underscore with an underscore.

irc_PING(msg)[source]Ā¶

IRC PING/PONG keepalive.

irc_CAP(msg)[source]Ā¶

Dispatch CAP subcommands to their own methods.

irc_CAP_LS(msg)[source]Ā¶

Response to CAP LS, giving list of available capabilities.

irc_CAP_ACK(msg)[source]Ā¶

Response to CAP REQ, acknowledging capability changes.

irc_CAP_NAK(msg)[source]Ā¶

Response to CAP REQ, rejecting capability changes.

irc_NICK(msg)[source]Ā¶

Somebodyā€™s nick changed.

irc_JOIN(msg)[source]Ā¶

Somebody joined a channel.

irc_PART(msg)[source]Ā¶

Somebody left a channel.

irc_KICK(msg)[source]Ā¶

Somebody was kicked from a channel.

irc_QUIT(msg)[source]Ā¶

Somebody quit the server.

irc_TOPIC(msg)[source]Ā¶

A channelā€™s topic changed.

irc_RPL_TOPIC(msg)[source]Ā¶

Topic notification, usually after joining a channel.

irc_PRIVMSG(msg)[source]Ā¶

Received a PRIVMSG.

TODO: Implement CTCP queries.

irc_NOTICE(msg)[source]Ā¶

Received a NOTICE.

TODO: Implement CTCP replies.

on_capabilities_available(capabilities)[source]Ā¶

Client capabilities are available.

Called with a set of client capability names when we get a response to CAP LS.

on_capability_enabled(name)[source]Ā¶

Client capability enabled.

Called when enabling client capability name has been acknowledged.

on_capability_disabled(name)[source]Ā¶

Client capability disabled.

Called when disabling client capability name has been acknowledged.

on_welcome()[source]Ā¶

Successfully signed on to the server.

on_nick_changed(nick)[source]Ā¶

Changed nick.

on_joined(channel)[source]Ā¶

Joined a channel.

on_left(channel)[source]Ā¶

Left a channel.

on_kicked(channel, by, reason)[source]Ā¶

Kicked from a channel.

on_privmsg(user, to, message)[source]Ā¶

Received a message, either directly or in a channel.

on_notice(user, to, message)[source]Ā¶

Received a notice, either directly or in a channel.

on_action(user, to, action)[source]Ā¶

Received CTCP ACTION. Common enough to deserve its own event.

on_ctcp_query_ACTION(user, to, data)[source]Ā¶

Turn CTCP ACTION into on_action() event.

on_user_renamed(oldnick, newnick)[source]Ā¶

User changed nick.

connect()[source]Ā¶

Connect to the IRC server.

connection_lost(exc)[source]Ā¶

Handle a broken connection by attempting to reconnect.

Wonā€™t reconnect if the broken connection was deliberate (i.e. close() was called).

connection_made()[source]Ā¶

Callback for successful connection.

Register with the IRC server.

on_user_joined(user, channel)[source]Ā¶

User joined a channel.

read_loop()[source]Ā¶

Read and dispatch lines until the connection closes.

run(run_once=False)[source]Ā¶

Run the bot, reconnecting when the connection is lost.

on_user_left(user, channel, message)[source]Ā¶

User left a channel.

on_user_kicked(user, channel, by, reason)[source]Ā¶

User kicked from a channel.

on_user_quit(user, message)[source]Ā¶

User disconnected.

on_topic_changed(user, channel, topic)[source]Ā¶

user changed the topic of channel to topic.

csbot.irc.main()[source]Ā¶
csbot.plugin moduleĀ¶
csbot.plugin.find_plugins()[source]Ā¶

Find available plugins.

Returns a list of discovered plugin classes.

csbot.plugin.build_plugin_dict(plugins)[source]Ā¶

Build a dictionary mapping the value of plugin_name() to each plugin class in plugins. PluginDuplicate is raised if more than one plugin has the same name.

class csbot.plugin.LazyMethod(obj, name)[source]Ā¶

Bases: object

exception csbot.plugin.PluginDuplicate[source]Ā¶

Bases: Exception

exception csbot.plugin.PluginDependencyUnmet[source]Ā¶

Bases: Exception

exception csbot.plugin.PluginFeatureError[source]Ā¶

Bases: Exception

exception csbot.plugin.PluginConfigError[source]Ā¶

Bases: Exception

class csbot.plugin.PluginManager(loaded, available, plugins, args)[source]Ā¶

Bases: collections.abc.Mapping

A simple plugin manager and proxy.

The plugin manager is responsible for loading plugins and proxying method calls to all plugins. In addition to accepting loaded, a list of existing plugin objects, it will attempt to load each of plugins from available (a mapping of plugin name to plugin class), passing args to the constructors.

Attempting to load missing or duplicate plugins will log errors and warnings respectively, but will not result in an exception or any change of state. A plugin classā€™ dependencies are checked before loading and a PluginDependencyUnmet is raised if any are missing.

The Mapping interface is implemented to provide easy querying and access to the loaded plugins. All attributes that do not start with a _ are treated as methods that will be proxied through to every plugin in the order they were loaded (loaded before plugins) with the same arguments.

plugins = NoneĀ¶

Loaded plugins.

class csbot.plugin.ProvidedByPlugin(plugin: str, kwargs: Mapping[str, Any], name: str = None)[source]Ā¶

Bases: object

Descriptor for plugin attributes that get (and cache) a value from another plugin.

See Plugin.use().

class csbot.plugin.PluginMeta(name, bases, attrs)[source]Ā¶

Bases: type

Metaclass for Plugin that collects methods tagged with plugin feature decorators.

classmethod current()[source]Ā¶
class csbot.plugin.Plugin(bot)[source]Ā¶

Bases: object

Bot plugin base class.

All bot plugins should inherit from this class. It provides convenience methods for hooking events, registering commands, accessing MongoDB and manipulating the configuration file.

CONFIG_DEFAULTS = {}Ā¶

Default configuration values, used automatically by config_get().

CONFIG_ENVVARS = {}Ā¶

Configuration environment variables, used automatically by config_get().

PLUGIN_DEPENDS = []Ā¶

Plugins that missing_dependencies() should check for.

log = NoneĀ¶

The pluginā€™s logger, created by default using the plugin classā€™ containing module name as the logger name.

classmethod plugin_name()[source]Ā¶

Get the name of the plugin, by default the class name in lowercase.

classmethod qualified_name()[source]Ā¶

Get the fully qualified class name, most useful when complaining about duplicate plugins names.

classmethod missing_dependencies(plugins)[source]Ā¶

Return elements from PLUGIN_DEPENDS that are not in the container plugins.

This should be used with some container of already loaded plugin names (e.g. a dictionary or set) to find out which dependencies are missing.

static hook(hook)[source]Ā¶
static command(cmd, **metadata)[source]Ā¶

Tag a command to be registered by setup().

Additional keyword arguments are added to a metadata dictionary that gets stored with the command. This is a good place to put, for example, the help string for the command:

@Plugin.command('foo', help='foo: does something amazing')
def foo_command(self, e):
    pass
static integrate_with(*otherplugins)[source]Ā¶

Tag a method as providing integration with otherplugins.

During setup(), all methods tagged with this decorator will be run if all of the named plugins are loaded. The actual plugin objects will be passed as arguments to the method in the same order.

Note

The order that integration methods are called in cannot be guaranteed, because attribute order is not preserved during class creation.

static use(other, **kwargs)[source]Ā¶

Create a property that will be provided by another plugin.

Returns a ProvidedByPlugin instance. PluginMeta will collect attributes of this type, and add other as an implicit plugin dependency. setup() will replace it with a value acquired from the plugin named by other. For example:

class Foo(Plugin):
    stuff = Plugin.use('mongodb', collection='stuff')

will cause setup() to replace the stuff attribute with:

self.bot.plugins[other].provide(self.plugin_name(), **kwargs)
get_hooks(hook: str) → List[Callable][source]Ā¶

Get a list of this pluginā€™s handlers for hook.

provide(plugin_name, **kwarg)[source]Ā¶

Provide a value for a Plugin.use() usage.

setup()[source]Ā¶

Plugin setup.

  • Replace all ProvidedByPlugin attributes.
  • Fire all plugin integration methods.
  • Register all commands provided by the plugin.
teardown()[source]Ā¶

Plugin teardown.

  • Unregister all commands provided by the plugin.
configĀ¶

Get the configuration section for this plugin.

Uses the [plugin_name] section of the configuration file, creating an empty section if it doesnā€™t exist.

See also

configparser

subconfig(subsection)[source]Ā¶

Get a configuration subsection for this plugin.

Uses the [plugin_name/subsection] section of the configuration file, creating an empty section if it doesnā€™t exist.

config_get(key)[source]Ā¶

Convenience wrapper proxying get() on config.

Given a key, this method tries the following in order:

self.config[key]
for v in self.CONFIG_ENVVARS[key]:
    os.environ[v]
self.CONFIG_DEFAULTS[key]

KeyError is raised if none of the methods succeed.

config_getboolean(key)[source]Ā¶

Identical to config_get(), but proxying getboolean.

class csbot.plugin.SpecialPlugin(bot)[source]Ā¶

Bases: csbot.plugin.Plugin

A special plugin with a special name that expects to be handled specially. Probably shouldnā€™t have too many of these or they wonā€™t feel special anymore.

classmethod plugin_name()[source]Ā¶

Change the plugin name to something that canā€™t possibly result from a class name by prepending a @.

csbot.util moduleĀ¶
csbot.util.nick(user)[source]Ā¶

Get nick from user string.

>>> nick('csyorkbot!~csbot@example.com')
'csyorkbot'
csbot.util.username(user)[source]Ā¶

Get username from user string.

>>> username('csyorkbot!~csbot@example.com')
'csbot'
csbot.util.host(user)[source]Ā¶

Get hostname from user string.

>>> host('csyorkbot!~csbot@example.com')
'example.com'
csbot.util.is_channel(channel)[source]Ā¶

Check if channel is a channel or private chat.

>>> is_channel('#cs-york')
True
>>> is_channel('csyorkbot')
False
csbot.util.parse_arguments(raw)[source]Ā¶

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)[source]Ā¶

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)[source]Ā¶

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

csbot.util.cap_string(s, n)[source]Ā¶

If a string is longer than a particular length, it gets truncated and has ā€˜ā€¦ā€™ added to the end.

csbot.util.ordinal(value)[source]Ā¶

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)[source]Ā¶
csbot.util.is_ascii(s)[source]Ā¶

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)[source]Ā¶

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[source]Ā¶

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]][source]Ā¶

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)>)[source]Ā¶

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[source]Ā¶

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

start()[source]Ā¶

Start async task to process calls.

stop(clear=True)[source]Ā¶

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()[source]Ā¶
csbot.util.type_validator(_obj, attrib: attr._make.Attribute, value)[source]Ā¶

An attrs validator that inspects the attribute type.

class csbot.util.PrettyStreamHandler(stream=None, colour=None)[source]Ā¶

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)[source]Ā¶

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)[source]Ā¶

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)[source]Ā¶

Module contentsĀ¶

Indices and tablesĀ¶