text:
from __future__ import print_function

from future import standard_library

standard_library.install_aliases()
import collections
import glob
import os
import re
import sys
import traceback

if "mtimes" not in globals():
    mtimes = {}

if "lastfiles" not in globals():
    lastfiles = set()

def make_signature(f):
    return f.__code__.co_filename, f.__name__, f.__code__.co_firstlineno

def format_plug(plug, kind="", lpad=0, width=40):
    out = " " * lpad + "%s:%s:%s" % make_signature(plug[0])
    if kind == "command":
        out += " " * (50 - len(out)) + plug[1]["name"]

    if kind == "event":
        out += " " * (50 - len(out)) + ", ".join(plug[1]["events"])

    if kind == "regex":
        out += " " * (50 - len(out)) + plug[1]["regex"]

    return out

def reload(init=False):
    changed = False

    if init:
        bot.plugs = collections.defaultdict(list)
        bot.threads = {}

    core_fileset = set(glob.glob(os.path.join("core", "*.py")))

    for filename in core_fileset:
        mtime = os.stat(filename).st_mtime
        if mtime != mtimes.get(filename):
            mtimes[filename] = mtime

            changed = True

            try:
                eval(compile(open(filename, "r").read(), filename, "exec"), globals())
            except Exception:
                traceback.print_exc()
                if init:  # stop if there's an error (syntax?) in a core
                    sys.exit()  # script on startup
                continue

            if filename == os.path.join("core", "reload.py"):
                reload(init=init)
                return

    fileset = set(glob.glob(os.path.join("plugins", "*.py")))

    # remove deleted/moved plugins
    for name, data in bot.plugs.items():
        bot.plugs[name] = [x for x in data if x[0]._filename in fileset]

    for filename in list(mtimes):
        if filename not in fileset and filename not in core_fileset:
            mtimes.pop(filename)

    for func, handler in list(bot.threads.items()):
        if func._filename not in fileset:
            handler.stop()
            del bot.threads[func]

    # compile new plugins
    for filename in fileset:
        mtime = os.stat(filename).st_mtime
        if mtime != mtimes.get(filename):
            mtimes[filename] = mtime

            changed = True

            try:
                code = compile(open(filename, "r").read(), filename, "exec")
                namespace = {}
                eval(code, namespace)
            except Exception:
                traceback.print_exc()
                continue

            # remove plugins already loaded from this filename
            for name, data in bot.plugs.items():
                bot.plugs[name] = [x for x in data if x[0]._filename != filename]

            for func, handler in list(bot.threads.items()):
                if func._filename == filename:
                    handler.stop()
                    del bot.threads[func]

            for obj in namespace.values():
                if hasattr(obj, "_hook"):  # check for magic
                    if obj._thread:
                        bot.threads[obj] = Handler(obj)

                    for type, data in obj._hook:
                        bot.plugs[type] += [data]

                        if not init:
                            print(
                                "### new plugin (type: %s) loaded:" % type,
                                format_plug(data),
                            )

    if changed:
        bot.commands = {}
        for plug in bot.plugs["command"]:
            name = plug[1]["name"].lower()
            if not re.match(r"^\w+$", name):
                print(
                    '### ERROR: invalid command name "%s" (%s)'
                    % (name, format_plug(plug))
                )
                continue
            if name in bot.commands:
                print(
                    "### ERROR: command '%s' already registered (%s, %s)"
                    % (name, format_plug(bot.commands[name]), format_plug(plug))
                )
                continue
            bot.commands[name] = plug

        bot.events = collections.defaultdict(list)
        for func, args in bot.plugs["event"]:
            for event in args["events"]:
                bot.events[event].append((func, args))

    if init:
        print("  plugin listing:")

        if bot.commands:
            # hack to make commands with multiple aliases
            # print nicely

            print("    command:")
            commands = collections.defaultdict(list)

            for name, (func, args) in bot.commands.items():
                commands[make_signature(func)].append(name)

            for sig, names in sorted(commands.items()):
                names.sort(key=lambda x: (-len(x), x))  # long names first
                out = " " * 6 + "%s:%s:%s" % sig
                out += " " * (50 - len(out)) + ", ".join(names)
                print(out)

        for kind, plugs in sorted(bot.plugs.items()):
            if kind == "command":
                continue
            print("    %s:" % kind)
            for plug in plugs:
                print(format_plug(plug, kind=kind, lpad=6))
        print()

repo_name: rmmh/skybot

repo_language: Python


text:
import os
import sqlite3

def get_db_connection(conn, name=""):
    "returns an sqlite3 connection to a persistent database"

    if not name:
        name = "%s.%s.db" % (conn.nick, conn.server_host)

    filename = os.path.join(bot.persist_dir, name)
    return sqlite3.connect(filename, timeout=10)

bot.get_db_connection = get_db_connection


text:
from __future__ import print_function
from builtins import map
from builtins import object
import re
import socket
import time
import _thread
import queue

from ssl import wrap_socket, CERT_NONE, CERT_REQUIRED, SSLError

DEFAULT_NAME = "skybot"
DEFAULT_REALNAME = "Python bot - http://github.com/rmmh/skybot"
DEFAULT_NICKSERV_NAME = "nickserv"
DEFAULT_NICKSERV_COMMAND = "IDENTIFY %s"

def decode(txt):
    for codec in ("utf-8", "iso-8859-1", "shift_jis", "cp1252"):
        try:
            return txt.decode(codec)
        except UnicodeDecodeError:
            continue

    return txt.decode("utf-8", "ignore")

def censor(text, censored_strings=None):
    text = re.sub("[\n\r]+", " ", text)

    if not censored_strings:
        return text

    words = map(re.escape, censored_strings)
    pattern = "(%s)" % "|".join(words)

    text = re.sub(pattern, "[censored]", text)

    return text

class crlf_tcp(object):

    "Handles tcp connections that consist of utf-8 lines ending with crlf"

    def __init__(self, host, port, timeout=300):
        self.ibuffer = b""
        self.obuffer = b""
        self.oqueue = queue.Queue()  # lines to be sent out
        self.iqueue = queue.Queue()  # lines that were received
        self.socket = self.create_socket()
        self.host = host
        self.port = port
        self.timeout = timeout

    def create_socket(self):
        return socket.socket(socket.AF_INET, socket.TCP_NODELAY)

    def run(self):
        while True:
            try:
                self.socket.connect((self.host, self.port))
            except socket.timeout:
                print("timed out connecting to %s:%s" % (self.host, self.port))
                time.sleep(60)
            else:
                break
        _thread.start_new_thread(self.recv_loop, ())
        _thread.start_new_thread(self.send_loop, ())

    def recv_from_socket(self, nbytes):
        return self.socket.recv(nbytes)

    def get_timeout_exception_type(self):
        return socket.timeout

    def handle_receive_exception(self, error, last_timestamp):
        if time.time() - last_timestamp > self.timeout:
            self.iqueue.put(StopIteration)
            self.socket.close()
            return True
        return False

    def recv_loop(self):
        last_timestamp = time.time()
        while True:
            try:
                data = self.recv_from_socket(4096)
                self.ibuffer += data
                if data:
                    last_timestamp = time.time()
                else:
                    if time.time() - last_timestamp > self.timeout:
                        self.iqueue.put(StopIteration)
                        self.socket.close()
                        return
                    time.sleep(1)
            except (self.get_timeout_exception_type(), socket.error) as e:
                if self.handle_receive_exception(e, last_timestamp):
                    return
                continue

            while b"\r\n" in self.ibuffer:
                line, self.ibuffer = self.ibuffer.split(b"\r\n", 1)
                self.iqueue.put(decode(line))

    def send_loop(self):
        while True:
            line = self.oqueue.get().splitlines()[0][:500]
            print(">>> %s" % line)
            self.obuffer += line.encode("utf-8", "replace") + b"\r\n"
            while self.obuffer:
                sent = self.socket.send(self.obuffer)
                self.obuffer = self.obuffer[sent:]

class crlf_ssl_tcp(crlf_tcp):

    "Handles ssl tcp connetions that consist of utf-8 lines ending with crlf"

    def __init__(self, host, port, ignore_cert_errors, timeout=300):
        self.ignore_cert_errors = ignore_cert_errors
        crlf_tcp.__init__(self, host, port, timeout)

    def create_socket(self):
        return wrap_socket(
            crlf_tcp.create_socket(self),
            server_side=False,
            cert_reqs=CERT_NONE if self.ignore_cert_errors else CERT_REQUIRED,
        )

    def recv_from_socket(self, nbytes):
        return self.socket.read(nbytes)

    def get_timeout_exception_type(self):
        return SSLError

    def handle_receive_exception(self, error, last_timestamp):
        return crlf_tcp.handle_receive_exception(self, error, last_timestamp)

def zip_channels(channels):
    channels.sort(key=lambda x: " " not in x)  # keyed channels first
    chans = []
    keys = []
    for channel in channels:
        if " " in channel:
            chan, key = channel.split(" ")
            chans.append(chan)
            keys.append(key)
        else:
            chans.append(channel)
    chans = ",".join(chans)
    if keys:
        return [chans, ",".join(keys)]
    else:
        return [chans]

def test_zip_channels():
    assert zip_channels(["#a", "#b c", "#d"]) == ["#b,#a,#d", "c"]
    assert zip_channels(["#a", "#b"]) == ["#a,#b"]

class IRC(object):
    IRC_PREFIX_REM = re.compile(r"(.*?) (.*?) (.*)").match
    IRC_NOPROFEIX_REM = re.compile(r"()(.*?) (.*)").match
    IRC_NETMASK_REM = re.compile(r":?([^!@]*)!?([^@]*)@?(.*)").match
    IRC_PARAM_REF = re.compile(r"(?:^|(?<= ))(:.*|[^ ]+)").findall

    "handles the IRC protocol"
    # see the docs/ folder for more information on the protocol

    def __init__(self, conf):
        self.conn = None

        self.nick = DEFAULT_NAME
        self.user = DEFAULT_NAME
        self.realname = DEFAULT_REALNAME
        self.user_mode = None

        self.server_host = None
        self.server_port = 6667
        self.server_password = None

        self.nickserv_password = None
        self.nickserv_name = DEFAULT_NICKSERV_NAME
        self.nickserv_command = DEFAULT_NICKSERV_COMMAND

        self.channels = []
        self.admins = []
        self.censored_strings = []

        self.out = queue.Queue()  # responses from the server are placed here
        # format: [rawline, prefix, command, params,
        # nick, user, host, paramlist, msg]

        self.set_conf(conf)

        self.connect()

        _thread.start_new_thread(self.parse_loop, ())

    def set_conf(self, conf):
        self.nick = conf.get("nick", DEFAULT_NAME)
        self.user = conf.get("user", DEFAULT_NAME)
        self.realname = conf.get("realname", DEFAULT_REALNAME)
        self.user_mode = conf.get("mode", None)

        self.server_host = conf["server"]
        self.server_port = conf.get("port", 6667)
        self.server_password = conf.get("server_password", None)

        self.nickserv_password = conf.get("nickserv_password", None)
        self.nickserv_name = conf.get("nickserv_name", DEFAULT_NICKSERV_NAME)
        self.nickserv_command = conf.get("nickserv_command", DEFAULT_NICKSERV_COMMAND)

        self.channels = conf.get("channels", [])
        self.admins = conf.get("admins", [])
        self.censored_strings = conf.get("censored_strings", [])

        if self.conn is not None:
            self.join_channels()

    def create_connection(self):
        return crlf_tcp(self.server_host, self.server_port)

    def connect(self):
        self.conn = self.create_connection()
        _thread.start_new_thread(self.conn.run, ())
        self.cmd("NICK", [self.nick])
        self.cmd("USER", [self.user, "3", "*", self.realname])
        if self.server_password:
            self.cmd("PASS", [self.server_password])

    def parse_loop(self):
        while True:
            msg = self.conn.iqueue.get()

            if msg == StopIteration:
                self.connect()
                continue

            if msg.startswith(":"):  # has a prefix
                prefix, command, params = self.IRC_PREFIX_REM(msg).groups()
            else:
                prefix, command, params = self.IRC_NOPROFEIX_REM(msg).groups()
            nick, user, host = self.IRC_NETMASK_REM(prefix).groups()
            paramlist = self.IRC_PARAM_REF(params)
            lastparam = ""
            if paramlist:
                if paramlist[-1].startswith(":"):
                    paramlist[-1] = paramlist[-1][1:]
                lastparam = paramlist[-1]
            self.out.put(
                [msg, prefix, command, params, nick, user, host, paramlist, lastparam]
            )

            if command == "PING":
                self.cmd("PONG", paramlist)

    def join(self, channel):
        self.cmd("JOIN", channel.split(" "))  # [chan, password]

    def join_channels(self):
        if self.channels:
            # TODO: send multiple join commands for large channel lists
            self.cmd("JOIN", zip_channels(self.channels))

    def msg(self, target, text):
        self.cmd("PRIVMSG", [target, text])

    def cmd(self, command, params=None):
        if params:
            params[-1] = ":" + params[-1]

            params = [censor(p, self.censored_strings) for p in params]

            self.send(command + " " + " ".join(params))
        else:
            self.send(command)

    def send(self, str):
        self.conn.oqueue.put(str)

class FakeIRC(IRC):
    def __init__(self, conf):
        self.set_conf(conf)
        self.out = queue.Queue()  # responses from the server are placed here

        self.f = open(fn, "rb")

        _thread.start_new_thread(self.parse_loop, ())

    def parse_loop(self):
        while True:
            msg = decode(self.f.readline()[9:])

            if msg == "":
                print("!!!!DONE READING FILE!!!!")
                return

            if msg.startswith(":"):  # has a prefix
                prefix, command, params = irc_prefix_rem(msg).groups()
            else:
                prefix, command, params = irc_noprefix_rem(msg).groups()
            nick, user, host = irc_netmask_rem(prefix).groups()
            paramlist = irc_param_ref(params)
            lastparam = ""
            if paramlist:
                if paramlist[-1].startswith(":"):
                    paramlist[-1] = paramlist[-1][1:]
                lastparam = paramlist[-1]
            self.out.put(
                [msg, prefix, command, params, nick, user, host, paramlist, lastparam]
            )
            if command == "PING":
                self.cmd("PONG", [params])

    def cmd(self, command, params=None):
        pass

class SSLIRC(IRC):
    def __init__(self, conf):
        super(SSLIRC, self).__init__(conf=conf)

        self.server_port = 6697
        self.server_ignore_cert = False

    def set_conf(self, conf):
        super(SSLIRC, self).set_conf(conf)

        self.server_port = conf.get("port", 6697)
        self.server_ignore_cert = conf.get("ignore_cert", False)

    def create_connection(self):
        return crlf_ssl_tcp(self.server_host, self.server_port, self.server_ignore_cert)


text:
from __future__ import print_function
import inspect
import json
import os

def find_config():
    # for backwards compatibility, look for either 'config' or 'config.json'
    if os.path.exists("config"):
        return "config"
    return "config.json"

def save(conf):
    json.dump(conf, open(find_config(), "w"), sort_keys=True, indent=2)

if not os.path.exists(find_config()):
    open("config.json", "w").write(
        inspect.cleandoc(
            r"""
        {
          "connections":
          {
            "local irc":
            {
              "server": "localhost",
              "nick": "skybot",
              "channels": ["#test"]
            }
          },
          "prefix": ".",
          "disabled_plugins": [],
          "disabled_commands": [],
          "acls": {},
          "api_keys": {},
          "censored_strings":
          [
            "DCC SEND",
            "1nj3ct",
            "thewrestlinggame",
            "startkeylogger",
            "hybux",
            "\\0",
            "\\x01",
            "!coz",
            "!tell /x"
          ]
        }"""
        )
        + "\n"
    )

def config():
    # reload config from file if file has changed
    config_mtime = os.stat(find_config()).st_mtime
    if bot._config_mtime != config_mtime:
        try:
            bot.config = json.load(open(find_config()))
            bot._config_mtime = config_mtime

            for name, conf in bot.config["connections"].items():
                conf.setdefault(
                    "censored_strings", bot.config.get("censored_strings", [])
                )

                if name in bot.conns:
                    bot.conns[name].set_conf(conf)
                else:
                    if conf.get("ssl"):
                        bot.conns[name] = SSLIRC(conf)
                    else:
                        bot.conns[name] = IRC(conf)
        except ValueError as e:
            print("ERROR: malformed config!", e)

bot._config_mtime = 0


text:
from __future__ import print_function
from builtins import str
from builtins import range
from builtins import object
import re
import _thread
import traceback
from queue import Queue
from future.builtins import str

_thread.stack_size(1024 * 512)  # reduce vm size

class Input(dict):
    def __init__(
        self, conn, raw, prefix, command, params, nick, user, host, paraml, msg
    ):

        server = conn.server_host

        chan = paraml[0].lower()
        if chan == conn.nick.lower():  # is a PM
            chan = nick

        def say(msg):
            conn.msg(chan, msg)

        def reply(msg):
            if chan == nick:  # PMs don't need prefixes
                self.say(msg)
            else:
                self.say(nick + ": " + msg)

        def pm(msg, nick=nick):
            conn.msg(nick, msg)

        def set_nick(nick):
            conn.set_nick(nick)

        def me(msg):
            self.say("\x01%s %s\x01" % ("ACTION", msg))

        def notice(msg):
            conn.cmd("NOTICE", [nick, msg])

        def kick(target=None, reason=None):
            conn.cmd("KICK", [chan, target or nick, reason or ""])

        def ban(target=None):
            conn.cmd("MODE", [chan, "+b", target or host])

        def unban(target=None):
            conn.cmd("MODE", [chan, "-b", target or host])

        dict.__init__(
            self,
            conn=conn,
            raw=raw,
            prefix=prefix,
            command=command,
            params=params,
            nick=nick,
            user=user,
            host=host,
            paraml=paraml,
            msg=msg,
            server=server,
            chan=chan,
            notice=notice,
            say=say,
            reply=reply,
            pm=pm,
            bot=bot,
            kick=kick,
            ban=ban,
            unban=unban,
            me=me,
            set_nick=set_nick,
            lastparam=paraml[-1],
        )

    # make dict keys accessible as attributes
    def __getattr__(self, key):
        return self[key]

    def __setattr__(self, key, value):
        self[key] = value

def run(func, input):
    args = func._args

    if "inp" not in input:
        input.inp = input.paraml

    if args:
        if "db" in args and "db" not in input:
            input.db = get_db_connection(input.conn)
        if "input" in args:
            input.input = input
        if 0 in args:
            out = func(input.inp, **input)
        else:
            kw = dict((key, input[key]) for key in args if key in input)
            out = func(input.inp, **kw)
    else:
        out = func(input.inp)
    if out is not None:
        input.reply(str(out))

def do_sieve(sieve, bot, input, func, type, args):
    try:
        return sieve(bot, input, func, type, args)
    except Exception:
        print("sieve error", end=" ")
        traceback.print_exc()
        return None

class Handler(object):

    """Runs plugins in their own threads (ensures order)"""

    def __init__(self, func):
        self.func = func
        self.input_queue = Queue()
        _thread.start_new_thread(self.start, ())

    def start(self):
        uses_db = "db" in self.func._args
        db_conns = {}
        while True:
            input = self.input_queue.get()

            if input == StopIteration:
                break

            if uses_db:
                db = db_conns.get(input.conn)
                if db is None:
                    db = bot.get_db_connection(input.conn)
                    db_conns[input.conn] = db
                input.db = db

            try:
                run(self.func, input)
            except:
                traceback.print_exc()

    def stop(self):
        self.input_queue.put(StopIteration)

    def put(self, value):
        self.input_queue.put(value)

def dispatch(input, kind, func, args, autohelp=False):
    for (sieve,) in bot.plugs["sieve"]:
        input = do_sieve(sieve, bot, input, func, kind, args)
        if input == None:
            return

    if (
        autohelp
        and args.get("autohelp", True)
        and not input.inp
        and func.__doc__ is not None
    ):
        input.reply(func.__doc__)
        return

    if hasattr(func, "_apikeys"):
        bot_keys = bot.config.get("api_keys", {})
        keys = {key: bot_keys.get(key) for key in func._apikeys}

        missing = [keyname for keyname, value in keys.items() if value is None]
        if missing:
            input.reply("error: missing api keys - {}".format(missing))
            return

        # Return a single key as just the value, and multiple keys as a dict.
        if len(keys) == 1:
            input.api_key = list(keys.values())[0]
        else:
            input.api_key = keys

    if func._thread:
        bot.threads[func].put(input)
    else:
        _thread.start_new_thread(run, (func, input))

def match_command(command):
    commands = list(bot.commands)

    # do some fuzzy matching
    prefix = [x for x in commands if x.startswith(command)]
    if len(prefix) == 1:
        return prefix[0]
    elif prefix and command not in prefix:
        return prefix

    return command

def make_command_re(bot_prefix, is_private, bot_nick):
    if not isinstance(bot_prefix, list):
        bot_prefix = [bot_prefix]
    if is_private:
        bot_prefix.append("")  # empty prefix
    bot_prefix = "|".join(re.escape(p) for p in bot_prefix)
    bot_prefix += "|" + bot_nick + r"[:,]+\s+"
    command_re = r"(?:%s)(\w+)(?:$|\s+)(.*)" % bot_prefix
    return re.compile(command_re)

def test_make_command_re():
    match = make_command_re(".", False, "bot").match
    assert not match("foo")
    assert not match("bot foo")
    for _ in range(2):
        assert match(".test").groups() == ("test", "")
        assert match("bot: foo args").groups() == ("foo", "args")
        match = make_command_re(".", True, "bot").match
    assert match("foo").groups() == ("foo", "")
    match = make_command_re([".", "!"], False, "bot").match
    assert match("!foo args").groups() == ("foo", "args")

def main(conn, out):
    inp = Input(conn, *out)

    # EVENTS
    for func, args in bot.events[inp.command] + bot.events["*"]:
        dispatch(Input(conn, *out), "event", func, args)

    if inp.command == "PRIVMSG":
        # COMMANDS
        config_prefix = bot.config.get("prefix", ".")
        is_private = inp.chan == inp.nick  # no prefix required
        command_re = make_command_re(config_prefix, is_private, inp.conn.nick)

        m = command_re.match(inp.lastparam)

        if m:
            trigger = m.group(1).lower()
            command = match_command(trigger)

            if isinstance(command, list):  # multiple potential matches
                input = Input(conn, *out)
                input.reply(
                    "did you mean %s or %s?" % (", ".join(command[:-1]), command[-1])
                )
            elif command in bot.commands:
                input = Input(conn, *out)
                input.trigger = trigger
                input.inp_unstripped = m.group(2)
                input.inp = input.inp_unstripped.strip()

                func, args = bot.commands[command]
                dispatch(input, "command", func, args, autohelp=True)

        # REGEXES
        for func, args in bot.plugs["regex"]:
            m = args["re"].search(inp.lastparam)
            if m:
                input = Input(conn, *out)
                input.inp = m

                dispatch(input, "regex", func, args)


text:
from __future__ import unicode_literals

import random

from util import hook, http

def api_get(query, key, is_image=None, num=1):
    url = (
        "https://www.googleapis.com/customsearch/v1?cx=007629729846476161907:ud5nlxktgcw"
        "&fields=items(title,link,snippet)&safe=off&nfpr=1"
        + ("&searchType=image" if is_image else "")
    )
    return http.get_json(url, key=key, q=query, num=num)

@hook.api_key("google")
@hook.command("can i get a picture of")
@hook.command("can you grab me a picture of")
@hook.command("give me a print out of")
@hook.command
def gis(inp, api_key=None):
    """.gis <term> -- finds an image using google images (safesearch off)"""

    parsed = api_get(inp, api_key, is_image=True, num=10)
    if "items" not in parsed:
        return "no images found"
    return random.choice(parsed["items"])["link"]

@hook.api_key("google")
@hook.command("g")
@hook.command
def google(inp, api_key=None):
    """.g/.google <query> -- returns first google search result"""

    parsed = api_get(inp, api_key)
    if "items" not in parsed:
        return "no results found"

    out = '{link} -- \x02{title}\x02: "{snippet}"'.format(**parsed["items"][0])
    out = " ".join(out.split())

    if len(out) > 300:
        out = out[: out.rfind(" ")] + '..."'

    return out


text:
"""
log.py: written by Scaevolus 2009
"""

from __future__ import print_function, unicode_literals

from builtins import str
import os
import codecs
import time
import re

from util import hook

log_fds = {}  # '%(net)s %(chan)s' : (filename, fd)

timestamp_format = "%H:%M:%S"

formats = {
    "PRIVMSG": "<%(nick)s> %(msg)s",
    "PART": "-!- %(nick)s [%(user)s@%(host)s] has left %(chan)s",
    "JOIN": "-!- %(nick)s [%(user)s@%(host)s] has joined %(param0)s",
    "MODE": "-!- mode/%(chan)s [%(param_tail)s] by %(nick)s",
    "KICK": "-!- %(param1)s was kicked from %(chan)s by %(nick)s [%(msg)s]",
    "TOPIC": "-!- %(nick)s changed the topic of %(chan)s to: %(msg)s",
    "QUIT": "-!- %(nick)s has quit [%(msg)s]",
    "PING": "",
    "NOTICE": "",
}

ctcp_formats = {"ACTION": "* %(nick)s %(ctcpmsg)s"}

irc_color_re = re.compile(r"(\x03(\d+,\d+|\d)|[\x0f\x02\x16\x1f])")

def get_log_filename(dir, server, chan):
    return os.path.join(
        dir, "log", gmtime("%Y"), server, (gmtime("%%s.%m-%d.log") % chan).lower()
    )

def gmtime(format):
    return time.strftime(format, time.gmtime())

def beautify(input):
    format = formats.get(input.command, "%(raw)s")
    args = dict(input)

    leng = len(args["paraml"])
    for n, p in enumerate(args["paraml"]):
        args["param" + str(n)] = p
        args["param_" + str(abs(n - leng))] = p

    args["param_tail"] = " ".join(args["paraml"][1:])
    args["msg"] = irc_color_re.sub("", args["msg"])

    if input.command == "PRIVMSG" and input.msg.count("\x01") >= 2:
        ctcp = input.msg.split("\x01", 2)[1].split(" ", 1)
        if len(ctcp) == 1:
            ctcp += [""]
        args["ctcpcmd"], args["ctcpmsg"] = ctcp
        format = ctcp_formats.get(
            args["ctcpcmd"],
            "%(nick)s [%(user)s@%(host)s] requested unknown CTCP "
            "%(ctcpcmd)s from %(chan)s: %(ctcpmsg)s",
        )

    return format % args

def get_log_fd(dir, server, chan):
    fn = get_log_filename(dir, server, chan)
    cache_key = "%s %s" % (server, chan)
    filename, fd = log_fds.get(cache_key, ("", 0))

    if fn != filename:  # we need to open a file for writing
        if fd != 0:  # is a valid fd
            fd.flush()
            fd.close()
        dir = os.path.split(fn)[0]
        if not os.path.exists(dir):
            os.makedirs(dir)
        fd = codecs.open(fn, "a", "utf-8")
        log_fds[cache_key] = (fn, fd)

    return fd

@hook.singlethread
@hook.event("*")
def log(paraml, input=None, bot=None):
    timestamp = gmtime(timestamp_format)

    if input.command == "QUIT":  # these are temporary fixes until proper
        input.chan = "quit"  # presence tracking is implemented
    if input.command == "NICK":
        input.chan = "nick"

    beau = beautify(input)

    if beau == "":  # don't log this
        return

    if input.chan:
        fd = get_log_fd(bot.persist_dir, input.server, input.chan)
        fd.write(timestamp + " " + beau + "\n")

    print(timestamp, input.chan, beau)


text:
from __future__ import unicode_literals

import re
from util import hook, http

soundcloud_track_re = (
    r"soundcloud.com/([-_a-z0-9]+)/([-_a-z0-9/]+)",
    re.I,
)

BASE_URL = "https://soundcloud.com/"

@hook.regex(*soundcloud_track_re)
def soundcloud_track(match):
    url = BASE_URL + match.group(1) + "/" + match.group(2)
    d = http.get_html(url)

    # iso 8601 fmt expected, e.g. PT01H53M22S
    duration = d.xpath('.//meta[@itemprop="duration"]/@content')[
        0].replace("PT", "").replace("00H", "").lower().strip('0')

    # iso 8601 fmt expected, e.g. 2022-12-30T21:09:15Z
    published = d.find('.//time[@pubdate]').text[:10]

    title = d.find('.//a[@itemprop="url"]').text
    name = d.xpath('.//meta[@itemprop="name"]/@content')[0]
    plays = d.xpath('.//meta[@property="soundcloud:play_count"]/@content')[0]
    title = d.xpath('//meta[@property="og:title"]/@content')[0]
    likes = d.xpath('//meta[@itemprop="interactionCount" '
                    'and starts-with(@content, "UserLikes:")]/@content')[0].split(":")[1]

    out = (
        "\x02{title}\x02 by "
        "\x02{name}\x02 - "
        "length \x02{duration}\x02 - "
        "{likes}\u2191 - "
        "\x02{plays}\x02 plays - "
        "posted on \x02{published}\x02"
    )

    output = out.format(title=title, duration=duration, likes=likes,
                        plays=plays, name=name, published=published)

    return output + " - " + url


text:
from util import hook, http

@hook.regex(r"(?i)http://(?:www\.)?tinyurl.com/([A-Za-z0-9\-]+)")
def tinyurl(match):
    try:
        return http.open(match.group()).url.strip()
    except http.URLError as e:
        pass


text:
from __future__ import unicode_literals

from builtins import chr
import re
import urllib.parse

from util import hook, http

@hook.api_key("wolframalpha")
@hook.command("who")
@hook.command("what")
@hook.command("where")
@hook.command("when")
@hook.command("why")
@hook.command
def ask(inp, say=None, trigger=None, nick=None, api_key=None):
    ".ask <query> -- ask wolfram a question, get an answer"

    if trigger in ["who", "what", "where", "when", "why"]:
        inp = f"{trigger} {inp}"

    base_url = "http://api.wolframalpha.com/v1/conversation.jsp"
    query = urllib.parse.quote_plus(inp)
    url = f"{base_url}?appid={api_key}&i={query}"

    response = http.get_json(url)
    result = response.get("result")

    if not result:
        say(f"Sorry, I do not know how to answer that {nick}.")

    return result

@hook.api_key("wolframalpha")
@hook.command("wa")
@hook.command
def wolframalpha(inp, api_key=None):
    ".wa/.wolframalpha <query> -- computes <query> using Wolfram Alpha"

    url = "http://api.wolframalpha.com/v2/query?format=plaintext"

    result = http.get_xml(url, input=inp, appid=api_key)

    pod_texts = []
    for pod in result.xpath("//pod"):
        title = "" if pod.attrib["title"] == "Result" else pod.attrib["title"] + ": "
        if pod.attrib["id"] == "Input":
            continue

        results = []
        for subpod in pod.xpath("subpod/plaintext/text()"):
            subpod = subpod.strip().replace("\\n", "; ")
            subpod = re.sub(r"\s+", " ", subpod)
            if subpod:
                results.append(subpod)
        if results:
            pod_texts.append(title + "|".join(results))

    ret = ". ".join(pod_texts)

    if not pod_texts:
        return "no results"

    ret = re.sub(r"\\(.)", r"\1", ret)

    def unicode_sub(match):
        return chr(int(match.group(1), 16))

    ret = re.sub(r"\\:([0-9a-z]{4})", unicode_sub, ret)

    if len(ret) > 430:
        ret = ret[: ret.rfind(" ", 0, 430)]
        ret = re.sub(r"\W+$", "", ret) + "..."

    if not ret:
        return "no results"

    return ret


text:
" seen.py: written by sklnd in about two beers July 2009"

from builtins import object
import time
import unittest

from util import hook, timesince

def db_init(db):
    "check to see that our db has the the seen table and return a connection."
    db.execute(
        "create table if not exists seen(name, time, quote, chan, "
        "primary key(name, chan))"
    )
    db.commit()

@hook.singlethread
@hook.event("PRIVMSG", ignorebots=False)
def seeninput(paraml, input=None, db=None, bot=None):
    db_init(db)
    db.execute(
        "insert or replace into seen(name, time, quote, chan)" "values(?,?,?,?)",
        (input.nick.lower(), time.time(), input.msg, input.chan),
    )
    db.commit()

@hook.command
def seen(inp, nick="", chan="", db=None, input=None):
    ".seen <nick> -- Tell when a nickname was last in active in irc"

    inp = inp.lower()

    if input.conn.nick.lower() == inp:
        # user is looking for us, being a smartass
        return "You need to get your eyes checked."

    if inp == nick.lower():
        return "Have you looked in a mirror lately?"

    db_init(db)

    last_seen = db.execute(
        "select name, time, quote from seen where" " name = ? and chan = ?", (inp, chan)
    ).fetchone()

    if last_seen:
        reltime = timesince.timesince(last_seen[1])
        if last_seen[2][0:1] == "\x01":
            return "%s was last seen %s ago: *%s %s*" % (
                inp,
                reltime,
                inp,
                last_seen[2][8:-1],
            )
        else:
            return "%s was last seen %s ago saying: %s" % (inp, reltime, last_seen[2])
    else:
        return "I've never seen %s" % inp

class SeenTest(unittest.TestCase):
    class Mock(object):
        def __init__(self, **kwargs):
            self.__dict__.update(kwargs)

    def setUp(self):
        import sqlite3

        self.db = sqlite3.connect(":memory:")

    def seeninput(self, nick, msg, chan="#test"):
        seeninput(None, db=self.db, input=self.Mock(nick=nick, msg=msg, chan=chan))

    def seen(self, inp, nick="bob", chan="#test", bot_nick="skybot"):
        return seen(
            inp,
            nick=nick,
            chan=chan,
            db=self.db,
            input=self.Mock(conn=self.Mock(nick=bot_nick)),
        )

    def test_missing(self):
        assert "I've never seen nemo" in self.seen("NEMO")

    def test_seen(self):
        self.seeninput("nemo", "witty banter")
        assert "nemo was last seen" in self.seen("nemo")
        assert "witty banter" in self.seen("nemo")

    def test_seen_missing_channel(self):
        self.seeninput("nemo", "msg", chan="#secret")
        assert "never seen" in self.seen("nemo")

    def test_seen_ctcp(self):
        self.seeninput("nemo", "\x01ACTION test lol\x01")
        assert self.seen("nemo").endswith("ago: *nemo test lol*")

    def test_snark_eyes(self):
        assert "eyes checked" in self.seen("skybot", bot_nick="skybot")

    def test_snark_mirror(self):
        assert "mirror" in self.seen("bob", nick="bob")


text:
import time

from util import hook, http

prompt = [0,
    {"role": "system", "content":
     "You are a terse and helpful assistant named SkyboGPT, answering questions briefly with one or two sentences. Answer as concisely as possible."},
]

@hook.api_key("openai")
@hook.command("gpt")
@hook.command
def chatgpt(inp, api_key=None, reply=None):
    ".gpt/.chatgpt <query> -- asks ChatGPT a question"

    global prompt

    if not api_key:
        return

    if inp == 'clear':
        prompt = prompt[:2]
        return

    if prompt[0] < time.time() - 60 * 10:  # forget after 10 minutes
        prompt = prompt[:2]

    prompt.append({"role": "user", "content": inp})

    url = "https://api.openai.com/v1/chat/completions"
    result = http.get_json(url, headers={'Authorization': 'Bearer ' + api_key}, json_data={
        "model": "gpt-3.5-turbo",
        "messages": prompt[1:]
    })

    if 'error' in result:
        reply(result['error']['message'])

    msg = result['choices'][0]['message']
    prompt[0] = time.time()
    prompt.append(msg)

    for line in prompt[1:]:
        print(line['role'] + ':', line['content'])
    print(result['usage'])

    if result['usage']['total_tokens'] > 2048:
        prompt.pop(2)

    reply(msg['content'].replace('\n', '  '))


text:
"""
TV information, written by Lurchington 2010
modified by rmmh 2010, 2013
"""

from builtins import map
import datetime

from util import hook, http, timesince

base_url = "http://thetvdb.com/api/"
api_key = "469B73127CA0C411"

def get_episodes_for_series(seriesname):
    res = {"error": None, "ended": False, "episodes": None, "name": None}
    # http://thetvdb.com/wiki/index.php/API:GetSeries
    try:
        query = http.get_xml(base_url + "GetSeries.php", seriesname=seriesname)
    except http.URLError:
        res["error"] = "error contacting thetvdb.com"
        return res

    series_id = query.xpath("//seriesid/text()")

    if not series_id:
        res["error"] = "unknown tv series (using www.thetvdb.com)"
        return res

    series_id = series_id[0]

    try:
        series = http.get_xml(
            base_url + "%s/series/%s/all/en.xml" % (api_key, series_id)
        )
    except http.URLError:
        res["error"] = "error contacting thetvdb.com"
        return res

    series_name = series.xpath("//SeriesName/text()")[0]

    if series.xpath("//Status/text()")[0] == "Ended":
        res["ended"] = True

    res["episodes"] = series.xpath("//Episode")
    res["name"] = series_name
    return res

def get_episode_info(episode):
    episode_air_date = episode.findtext("FirstAired")

    try:
        airdate = datetime.date(*list(map(int, episode_air_date.split("-"))))
    except (ValueError, TypeError):
        return None

    episode_num = "S%02dE%02d" % (
        int(episode.findtext("SeasonNumber")),
        int(episode.findtext("EpisodeNumber")),
    )

    episode_name = episode.findtext("EpisodeName")
    # in the event of an unannounced episode title, users either leave the
    # field out (None) or fill it with TBA
    if episode_name == "TBA":
        episode_name = None

    episode_desc = "%s" % episode_num
    if episode_name:
        episode_desc += " - %s" % episode_name
    return (episode_air_date, airdate, episode_desc)

@hook.command
@hook.command("tv")
def tv_next(inp):
    ".tv_next <series> -- get the next episode of <series>"
    episodes = get_episodes_for_series(inp)

    if episodes["error"]:
        return episodes["error"]

    series_name = episodes["name"]
    ended = episodes["ended"]
    episodes = episodes["episodes"]

    if ended:
        return "%s has ended." % series_name

    next_eps = []
    today = datetime.date.today()

    for episode in reversed(episodes):
        ep_info = get_episode_info(episode)

        if ep_info is None:
            continue

        (episode_air_date, airdate, episode_desc) = ep_info

        if airdate > today:
            next_eps = [
                "%s (%s) (%s)"
                % (
                    episode_air_date,
                    timesince.timeuntil(
                        datetime.datetime.strptime(episode_air_date, "%Y-%m-%d")
                    ),
                    episode_desc,
                )
            ]
        elif airdate == today:
            next_eps = ["Today (%s)" % episode_desc] + next_eps
        else:
            # we're iterating in reverse order with newest episodes last
            # so, as soon as we're past today, break out of loop
            break

    if not next_eps:
        return "there are no new episodes scheduled for %s" % series_name

    if len(next_eps) == 1:
        return "the next episode of %s airs %s" % (series_name, next_eps[0])
    else:
        next_eps = ", ".join(next_eps)
        return "the next episodes of %s: %s" % (series_name, next_eps)

@hook.command
@hook.command("tv_prev")
def tv_last(inp):
    ".tv_last <series> -- gets the most recently aired episode of <series>"
    episodes = get_episodes_for_series(inp)

    if episodes["error"]:
        return episodes["error"]

    series_name = episodes["name"]
    ended = episodes["ended"]
    episodes = episodes["episodes"]

    prev_ep = None
    today = datetime.date.today()

    for episode in reversed(episodes):
        ep_info = get_episode_info(episode)

        if ep_info is None:
            continue

        (episode_air_date, airdate, episode_desc) = ep_info

        if airdate < today:
            # iterating in reverse order, so the first episode encountered
            # before today was the most recently aired
            prev_ep = "%s (%s)" % (episode_air_date, episode_desc)
            break

    if not prev_ep:
        return "there are no previously aired episodes for %s" % series_name
    if ended:
        return "%s ended. The last episode aired %s" % (series_name, prev_ep)
    return "the last episode of %s aired %s" % (series_name, prev_ep)


text:
from builtins import map
import re

from util import hook

@hook.sieve
def sieve_suite(bot, input, func, kind, args):
    if (
        input.command == "PRIVMSG"
        and bot.config.get("ignorebots", True)
        and input.nick.lower()[-3:] == "bot"
        and args.get("ignorebots", True)
    ):
        return None

    if kind == "command":
        if input.trigger in bot.config.get("disabled_commands", []):
            return None

        ignored = bot.config.get("ignored", [])
        if input.host in ignored or input.nick in ignored:
            return None

    fn = re.match(r"^plugins.(.+).py$", func._filename)
    disabled = bot.config.get("disabled_plugins", [])
    if fn and fn.group(1).lower() in disabled:
        return None

    acls = bot.config.get("acls", {})
    for acl in [acls.get(func.__name__), acls.get(input.chan), acls.get(input.server)]:
        if acl is None:
            continue
        if "deny-except" in acl:
            allowed_channels = [x.lower() for x in acl["deny-except"]]
            if input.chan.lower() not in allowed_channels:
                return None
        if "allow-except" in acl:
            denied_channels = [x.lower() for x in acl["allow-except"]]
            if input.chan.lower() in denied_channels:
                return None
        if "whitelist" in acl:
            if func.__name__ not in acl["whitelist"]:
                return None
        if "blacklist" in acl:
            if func.__name__ in acl["whitelist"]:
                return None
        if "blacklist-nicks" in acl:
            if input.nick.lower() in acl["blacklist-nicks"]:
                return None

    admins = input.conn.admins
    input.admin = input.host in admins or input.nick in admins

    if args.get("adminonly", False):
        if not input.admin:
            return None

    return input


text:
from __future__ import unicode_literals

import re

from util import hook, http

SA_THREAD_RE = r"(?i)forums\.somethingawful\.com/\S*\?(?:\S+&)?threadid=(\d+)\S*"
SA_PROFILE_RE = (
    r"(?i)forums\.somethingawful\.com/member.php\?\S+(userid|username)=([^&]+)"
)

LOGIN_URL = "https://forums.somethingawful.com/account.php"
THREAD_URL = "https://forums.somethingawful.com/showthread.php"
PROFILE_URL = "https://forums.somethingawful.com/member.php"

MATCH_OBJECT = type(re.match(r"", ""))  # Is there a nicer way to get this??

GENDER_UNICODE = {"male": "\u2642", "female": "\u2640", "porpoise": "\uE520"}

FORUM_ABBREVS = {
    "Serious Hardware / Software Crap": "SHSC",
    "The Cavern of COBOL": "CoC",
    "General Bullshit": "GBS",
    "Haus of Tech Support": "HoTS",
}

def login(user, password):
    """
    Authenticate against SomethingAwful, both storing that authentication in
    the global cookiejar and returning the relevant cookies

    :param user: your awful username for somethingawful dot com
    :param password: your awful password for somethingawful dot com
    :return: the authentication cookies for somethingawful dot com
    """

    get_sa_cookies = lambda jar: [
        c
        for c in jar
        if c.domain.endswith("forums.somethingawful.com")
        and (c.name == "bbuserid" or c.name == "bbpassword")
    ]

    http.clear_expired_cookies()
    sa_cookies = get_sa_cookies(http.get_cookie_jar())

    if len(sa_cookies) == 2:
        return sa_cookies

    post_data = {"action": "login", "username": user, "password": password}
    http.get(LOGIN_URL, cookies=True, post_data=post_data)

    sa_cookies = get_sa_cookies(http.get_cookie_jar())

    if len(sa_cookies) < 2:
        return None

    return sa_cookies

def parse_profile_html(document):
    """
    Parse an LXML document to retrieve the profile data

    :param document: the LXML document to parse
    :return: a dictionary representing the profile
    """
    username_elements = document.xpath("//*[@class='author']")
    registered_elements = document.xpath("//*[@class='registered']")
    avatar_elements = document.xpath("//*[@class='title']//img")
    info_elements = document.xpath("//*[@class='info']")
    userid_elements = document.xpath("//*[@name='userid']")

    profile = {}

    if userid_elements:
        profile["id"] = userid_elements[0].attrib["value"]

    if username_elements:
        profile["username"] = username_elements[0].text_content()

    if registered_elements:
        profile["registered"] = registered_elements[0].text_content()

    if avatar_elements:
        profile["avatar"] = (
            avatar_elements[0].attrib["src"] if avatar_elements[0].attrib["src"] else ""
        )
        profile["is_newbie"] = profile["avatar"].endswith("/images/newbie.gif")

    if info_elements:
        info_text = info_elements[0].text_content()

        post_count = re.search(r"Post Count(\d+)", info_text)
        if post_count:
            profile["post_count"] = post_count.group(1)

        post_rate = re.search(r"Post Rate([\d\.]+)", info_text)
        if post_rate:
            profile["post_rate"] = post_rate.group(1)

        last_post = re.search(r"Last Post(.+)", info_text)
        if last_post:
            profile["last_post"] = last_post.group(1)

        gender = re.search(r"claims to be a ([-a-z0-9 ]+)", info_text)
        if gender:
            profile["gender"] = gender.group(1).lower()

    if "id" in profile:
        profile["profile_link"] = http.prepare_url(
            PROFILE_URL, {"action": "getinfo", "userid": profile["id"]}
        )
    elif "username" in profile:
        profile["profile_link"] = http.prepare_url(
            PROFILE_URL, {"action": "getinfo", "username": profile["username"]}
        )

    return profile

def parse_thread_html(document):
    """
    Parse an LXML document to retrieve the thread data

    :param document: the LXML document to parse
    :return: a dictionary representing the thread
    """
    breadcrumbs_elements = document.xpath("//div[@class='breadcrumbs']//a")
    author_elements = document.xpath("//dt[contains(@class, author)]")
    last_page_elements = document.xpath("//a[@title='Last page']")

    if not breadcrumbs_elements:
        return

    if not author_elements:
        return

    if len(breadcrumbs_elements) < 2:
        return

    thread_id = int(breadcrumbs_elements[-1].attrib["href"].rsplit("=", 2)[1])

    breadcrumbs = [e.text_content() for e in breadcrumbs_elements]

    thread_title = breadcrumbs[-1]
    forum_title = breadcrumbs[-2]

    if author_elements:
        author = author_elements[0].text_content().strip()
    else:
        author = "Unknown Author"

    # Handle GBS / FYAD / E/N / etc
    if ":" in forum_title:
        forum_title = forum_title.split(":")[0].strip()

    if forum_title in FORUM_ABBREVS:
        forum_title = FORUM_ABBREVS[forum_title]

    if last_page_elements:
        post_count = int(last_page_elements[0].text_content().split(" ")[0])
    else:
        post_count = 1

    posts = {
        x.attrib["id"]: (
            x.xpath('.//dt[contains(@class, "author")]')[0].text_content(),
            x.xpath('.//*[@class="postdate"]')[0].text_content().strip("\n #?"),
            x.xpath('.//*[@class="postbody"]')[0].text_content().strip(),
        )
        for x in document.xpath('//table[contains(@class, "post")]')
    }

    return {
        "id": thread_id,
        "breadcrumbs": breadcrumbs,
        "forum_title": forum_title,
        "thread_title": thread_title,
        "author": author,
        "post_count": post_count,
        "posts": posts,
        "thread_link": http.prepare_url(THREAD_URL, {"threadid": thread_id}),
    }

def get_thread_by_id(credentials, id, params=None):
    """
    Get thread data via the ID of the thread

    :param credentials: the credentials to lookup with
    :param id: the ID to look up
    :return: a dictionary representing the thread
    """
    if not login(credentials["user"], credentials["password"]):
        raise ValueError("invalid login")

    thread_document = http.get_html(
        THREAD_URL,
        cookies=True,
        query_params=params or {"noseen": 1, "threadid": id, "perpage": 1},
    )

    return parse_thread_html(thread_document)

def get_profile_by_id(credentials, id):
    """
    Get profile data via the ID of a user

    :param credentials: the credentials to lookup with
    :param id: the ID to look up
    :return: a dictionary representing a profile
    """
    if not login(credentials["user"], credentials["password"]):
        raise ValueError("invalid login")

    profile_document = http.get_html(
        PROFILE_URL, cookies=True, query_params={"action": "getinfo", "userid": id}
    )

    return parse_profile_html(profile_document)

def get_profile_by_username(credentials, username):
    """
    Get profile data via the username of a user

    :param credentials: the credentials to lookup with
    :param username: the username to look up
    :return: a dictionary representing a profile
    """
    if not login(credentials["user"], credentials["password"]):
        raise ValueError("invalid login")

    profile_document = http.get_html(
        PROFILE_URL,
        cookies=True,
        query_params={"action": "getinfo", "username": username},
    )

    return parse_profile_html(profile_document)

def format_profile_response(profile, show_link=False):
    """
    Format a profile object into a response for sending out via IRC reply.

    :param profile: a dictionary with profile data from parse_profile_html
    :param show_link: whether or not to show the link to the profile
    :return: a human-readable string representation of the profile
    """
    if not profile:
        return "profile not found"

    if "gender" in profile and profile["gender"] in GENDER_UNICODE:
        profile["gender_symbol"] = GENDER_UNICODE[profile["gender"]]
    else:
        profile["gender_symbol"] = "?"

    if show_link:
        return (
            "\x02{username}\x02 ({gender_symbol}) - registered \x02{registered}\x02 - "
            "last post \x02{last_post}\x02 - {post_rate} posts per day - {profile_link}"
        ).format(**profile)
    else:
        return (
            "\x02{username}\x02 ({gender_symbol}) - registered \x02{registered}\x02 - "
            "last post \x02{last_post}\x02 - {post_rate} posts per day"
        ).format(**profile)

@hook.api_key("somethingawful")
@hook.command("profile")
def profile_username(inp, api_key=None):
    """.profile <username> - get the SomethingAwful profile for a user via their username"""
    if api_key is None or "user" not in api_key or "password" not in api_key:
        return

    profile = get_profile_by_username(api_key, inp)

    return format_profile_response(profile, show_link=True)

@hook.api_key("somethingawful")
@hook.regex(SA_PROFILE_RE)
def profile_link(inp, api_key=None):
    if api_key is None or "user" not in api_key or "password" not in api_key:
        return

    if not isinstance(inp, MATCH_OBJECT):
        inp = re.search(SA_PROFILE_RE, inp)

    if not inp:
        return

    profile_lookup_type = inp.group(1).lower()
    profile_lookup_value = inp.group(2)

    profile = None

    if profile_lookup_type == "userid":
        profile = get_profile_by_id(api_key, profile_lookup_value)
    elif profile_lookup_type == "username":
        profile = get_profile_by_username(api_key, profile_lookup_value)

    return format_profile_response(profile)

@hook.api_key("somethingawful")
@hook.regex(SA_THREAD_RE)
def thread_link(inp, api_key=None):
    if api_key is None or "user" not in api_key or "password" not in api_key:
        return

    if not isinstance(inp, MATCH_OBJECT):
        inp = re.search(SA_THREAD_RE, inp)

    if not inp:
        return

    post = None
    if "#post" in inp.group(0):
        parsed = http.urlparse(inp.group(0))
        thread = get_thread_by_id(
            api_key, inp.group(1), params=dict(http.parse_qsl(parsed.query))
        )
        post = thread["posts"].get(parsed.fragment)
        print(post)
    else:
        thread = get_thread_by_id(api_key, inp.group(1))

    if not thread:
        return

    if len(thread["thread_title"]) > 100:
        thread["thread_title"] = thread["thread_title"][0:97] + "\u2026"

    thread["post_count_word"] = "posts" if thread["post_count"] > 1 else "post"

    if post:
        author, date, content = post
        content = re.sub(r"\n+", " // ", content)
        if len(content) > 400:
            content = content[:400] + "..."
        return (
            "\x02{forum_title}\x02 > \x02{thread_title}\x02: "
            "\x02{poster}\x02 on {date}: {content}"
        ).format(poster=author, date=date, content=content, **thread)

    return (
        "\x02{forum_title}\x02 > \x02{thread_title}\x02 by \x02{author}\x02, "
        "\x02{post_count}\x02 {post_count_word}"
    ).format(**thread)


text:
from util import hook, http

@hook.regex(r"vimeo.com/([0-9]+)")
def vimeo_url(match):
    info = http.get_json("http://vimeo.com/api/v2/video/%s.json" % match.group(1))

    if info:
        return (
            "\x02%(title)s\x02 - length \x02%(duration)ss\x02 - "
            "\x02%(stats_number_of_likes)s\x02 likes - "
            "\x02%(stats_number_of_plays)s\x02 plays - "
            "\x02%(user_name)s\x02 on \x02%(upload_date)s\x02" % info[0]
        )


text:
import random

from util import hook, http

@hook.api_key("giphy")
@hook.command("gif")
@hook.command
def giphy(inp, api_key=None):
    """.gif/.giphy <query> -- returns first giphy search result"""
    url = "http://api.giphy.com/v1/gifs/search"
    try:
        response = http.get_json(url, q=inp, limit=10, api_key=api_key)
    except http.HTTPError as e:
        return e.msg

    results = response.get("data")
    if results:
        return random.choice(results).get("bitly_gif_url")
    else:
        return "no results found"


text:
import time
import uuid
import json

from util import hook, http

def piston(language, version, content, filename=None):
    api_url = "https://emkc.org/api/v2/piston/execute"

    if not filename:
        filename = str(uuid.uuid1())

    payload = {
        "language": language,
        "version": version,
        "files": [{"name": filename, "content": content}],
        "stdin": "",
        "args": [],
        "compile_timeout": 5000,
        "run_timeout": 5000,
        "compile_memory_limit": -1,
        "run_memory_limit": -1,
    }

    headers = {"Content-Type", "application/json"}

    result = http.get_json(api_url, json_data=payload, get_methods="POST")

    if result["run"]["stderr"] != "":
        return "Error: " + result["run"]["stderr"]

    return result["run"]["output"].replace("\n", " ")

@hook.command
def cpp(inp, nick=None):
    headers = ['algorithm', 'any', 'array', 'atomic', 'bit', 'bitset',
               'cassert', 'ccomplex', 'cctype', 'cerrno', 'cfenv', 'cfloat',
               'charconv', 'chrono', 'cinttypes', 'ciso646', 'climits',
               'clocale', 'cmath', 'codecvt', 'complex', 'complex.h',
               'condition_variable', 'csetjmp', 'csignal', 'cstdalign',
               'cstdarg', 'cstdbool', 'cstddef', 'cstdint', 'cstdio',
               'cstdlib', 'cstring', 'ctgmath', 'ctime', 'cuchar', 'cwchar',
               'cwctype', 'cxxabi.h', 'deque', 'exception', 'execution',
               'fenv.h', 'filesystem', 'forward_list', 'fstream',
               'functional', 'future', 'initializer_list', 'iomanip', 'ios',
               'iosfwd', 'iostream', 'istream', 'iterator', 'limits', 'list',
               'locale', 'map', 'math.h', 'memory', 'memory_resource', 'mutex',
               'new', 'numeric', 'optional', 'ostream', 'queue', 'random',
               'ratio', 'regex', 'scoped_allocator', 'set', 'shared_mutex',
               'sstream', 'stack', 'stdexcept', 'stdlib.h', 'streambuf',
               'string', 'string_view', 'system_error', 'tgmath.h', 'thread',
               'tuple', 'typeindex', 'typeinfo', 'type_traits',
               'unordered_map', 'unordered_set', 'utility', 'valarray',
               'variant', 'vector', 'version']
    header_includes = "\n".join([f"#include <{x}>" for x in headers])
    contents = f"""
        {header_includes}
        using namespace std;
        using namespace std::chrono_literals;
        using namespace std::complex_literals;
        using namespace std::string_literals;
        using namespace std::string_view_literals;
        int main() {{
            {inp};
        }}
    """
    return piston("c++", "10.2.0", contents, filename=nick)

@hook.command
def lua(inp, nick=None):
    return piston("lua", "5.4.2", inp, filename=nick)

@hook.command("cl")
def lisp(inp, nick=None):
    return piston("lisp", "2.1.2", inp, filename=nick)

@hook.command("elisp")
@hook.command("el")
def emacs(inp, nick=None):
    return piston("emacs", "27.1.0", inp, filename=nick)

@hook.command("clj")
def clojure(inp, nick=None):
    return piston("clojure", "1.10.3", inp, filename=nick)

@hook.command("rs")
def rust(inp, nick=None):
    contents = "func main() {" + inp + "}";
    return piston("rust", "1.50.0", contents, filename=nick)

@hook.command("py")
@hook.command("py3")
def python(inp, nick=None):
    return piston("python", "3.10.0", inp, filename=nick)

@hook.command("py2")
def python2(inp, nick=None):
    return piston("python2", "2.7.18", inp, filename=nick)

@hook.command("rb")
def ruby(inp, nick=None):
    return piston("ruby", "3.0.1", inp, filename=nick)

@hook.command("ts")
def typescript(inp, nick=None):
    return piston("typescript", "4.2.3", inp, filename=nick)

@hook.command("js")
@hook.command("node")
def javascript(inp, nick=None):
    return piston("javascript", "16.3.0", inp, filename=nick)

@hook.command
def perl(inp, nick=None):
    return piston("perl", "5.26.1", inp, filename=nick)

@hook.command
def php(inp, nick=None):
    contents = "<?php" + inp + "?>"
    return piston("php", "8.0.2", contents, filename=nick)

@hook.command
def swift(inp, nick=None):
    return piston("swift", "5.3.3", inp, filename=nick)

if __name__ == "__main__":
    result = piston("py", "3.9.4", "print('test')")

    print(result)


text:
from __future__ import division, unicode_literals
from past.utils import old_div
import math
import time

from util import hook, urlnorm, timesince

expiration_period = 60 * 60 * 24  # 1 day

ignored_urls = [urlnorm.normalize("http://google.com")]

def db_init(db):
    db.execute("create table if not exists urlhistory" "(chan, url, nick, time)")
    db.commit()

def insert_history(db, chan, url, nick):
    db.execute(
        "insert into urlhistory(chan, url, nick, time) " "values(?,?,?,?)",
        (chan, url, nick, time.time()),
    )
    db.commit()

def get_history(db, chan, url):
    db.execute(
        "delete from urlhistory where time < ?", (time.time() - expiration_period,)
    )
    return db.execute(
        "select nick, time from urlhistory where "
        "chan=? and url=? order by time desc",
        (chan, url),
    ).fetchall()

def nicklist(nicks):
    nicks.sort(key=lambda n: n.lower())

    if len(nicks) <= 2:
        return " and ".join(nicks)
    else:
        return ", and ".join((", ".join(nicks[:-1]), nicks[-1]))

def format_reply(history):
    if not history:
        return

    last_nick, recent_time = history[0]
    last_time = timesince.timesince(recent_time)

    if len(history) == 1:
        return "%s linked that %s ago." % (last_nick, last_time)

    hour_span = math.ceil(old_div((time.time() - history[-1][1]), 3600))
    hour_span = "%.0f hours" % hour_span if hour_span > 1 else "hour"

    hlen = len(history)
    ordinal = ["once", "twice", "%d times" % hlen][min(hlen, 3) - 1]

    if len(dict(history)) == 1:
        last = "last linked %s ago" % last_time
    else:
        last = "last linked by %s %s ago" % (last_nick, last_time)

    return "that url has been posted %s in the past %s by %s (%s)." % (
        ordinal,
        hour_span,
        nicklist([h[0] for h in history]),
        last,
    )

@hook.regex(r"([a-zA-Z]+://|www\.)[^ ]+")
def urlinput(match, nick="", chan="", db=None, bot=None):
    db_init(db)
    url = urlnorm.normalize(match.group())
    if url not in ignored_urls:
        url = url
        history = get_history(db, chan, url)
        insert_history(db, chan, url, nick)

        inp = match.string.lower()

        for name in dict(history):
            if name.lower() in inp:  # person was probably quoting a line
                return  # that had a link. don't remind them.

        if nick not in dict(history):
            return format_reply(history)


text:
"""
🔮 Spooky fortunes and assistance for witches
"""
from future.standard_library import hooks

with hooks():
    from urllib.error import HTTPError
    from urllib.parse import quote

from util import hook, http

@hook.command
def tarot(inp):
    ".tarot <cardname> -- finds a card by name"
    search = quote(inp)

    try:
        card = http.get_json(f"https://tarot-api.com/find/{search}")
    except HTTPError:
        return "The spirits are displeased."

    return card["name"] + ": " + ", ".join(card["keywords"])

@hook.command(autohelp=False)
def fortune(inp):
    """
    .fortune -- returns one random card and it's fortune
    """

    try:
        cards = http.get_json("https://tarot-api.com/draw/1")
    except HTTPError:
        return "The spirits are displeased."

    card = cards[0]

    return card["name"] + ": " + ", ".join(card["keywords"])


text:
from __future__ import unicode_literals

from util import http, hook

@hook.regex(r"(?i)https://(?:www\.)?news\.ycombinator\.com\S*id=(\d+)")
def hackernews(match):
    base_api = "https://hacker-news.firebaseio.com/v0/item/"
    entry = http.get_json(base_api + match.group(1) + ".json")

    if entry["type"] == "story":
        entry["title"] = http.unescape(entry["title"])
        return "{title} by {by} with {score} points and {descendants} comments ({url})".format(
            **entry
        )

    if entry["type"] == "comment":
        entry["text"] = http.unescape(entry["text"].replace("<p>", " // "))
        return '"{text}" -- {by}'.format(**entry)


text:
import socket
import subprocess
import time

from util import hook, http

socket.setdefaulttimeout(10)  # global setting

def get_version():
    try:
        stdout = subprocess.check_output(["git", "log", "--format=%h"])
    except:
        revnumber = 0
        shorthash = "????"
    else:
        revs = stdout.splitlines()
        revnumber = len(revs)
        shorthash = revs[0]

    shorthash = shorthash.decode("utf-8")

    http.ua_skybot = "Skybot/r%d %s (http://github.com/rmmh/skybot)" % (
        revnumber,
        shorthash,
    )

    return shorthash, revnumber

# autorejoin channels
@hook.event("KICK")
def rejoin(paraml, conn=None):
    if paraml[1] == conn.nick:
        if paraml[0].lower() in conn.channels:
            conn.join(paraml[0])

# join channels when invited
@hook.event("INVITE")
def invite(paraml, conn=None):
    conn.join(paraml[-1])

@hook.event("004")
def onjoin(paraml, conn=None):
    # identify to services
    nickserv_password = conn.nickserv_password
    nickserv_name = conn.nickserv_name
    nickserv_command = conn.nickserv_command
    if nickserv_password:
        conn.msg(nickserv_name, nickserv_command % nickserv_password)
        time.sleep(1)

    # set mode on self
    mode = conn.user_mode
    if mode:
        conn.cmd("MODE", [conn.nick, mode])

    conn.join_channels()

    # set user-agent as a side effect of reading the version
    get_version()

@hook.regex(r"^\x01VERSION\x01$")
def version(inp, notice=None):
    ident, rev = get_version()
    notice(
        "\x01VERSION skybot %s r%d - http://github.com/rmmh/"
        "skybot/\x01" % (ident, rev)
    )


text:
from time import gmtime
import re

from util import hook, http

PRE_DB_SEARCH_URL = "https://pr3.us/search.php"
PRE_DB_RE_NOT_FOUND = re.compile("^Nothing found for: .+$")

def get_predb_release(release_name):
    timestamp = gmtime()

    try:
        # Without accept headers, the underlying
        # site falls apart for some reason
        h = http.get_html(
            PRE_DB_SEARCH_URL,
            search=release_name,
            ts=timestamp,
            pretimezone=0,
            timezone=0,
            headers={"Accept": "*/*", "Accept-Language": "en-US,en;q=0.5"},
        )
    except http.HTTPError:
        return None

    results = h.xpath("//tr")

    if not results:
        return None

    first_result = results[0]

    if PRE_DB_RE_NOT_FOUND.match(first_result.text_content()):
        return None

    section_field, name_field, _, size_field, ts_field = first_result.xpath("./td")[:5]

    date, time = ts_field.text_content().split()

    return {
        "date": date,
        "time": time,
        "section": section_field.text_content(),
        "name": name_field.text_content().strip(),
        "size": size_field.text_content(),
    }

@hook.command
def predb(inp):
    """.predb <query> -- searches scene releases"""

    release = get_predb_release(inp)

    if not release:
        return "zero results"

    if release["size"]:
        release["size"] = " - %s" % release["size"]

    return "{date} - {section} - {name}{size}".format(**release)


text:
"dotnetpad.py: by sklnd, because gobiner wouldn't shut up"

import urllib.request, urllib.parse, urllib.error

from util import hook, http

def dotnetpad(lang, code):
    "Posts a provided snippet of code in a provided langugage to dotnetpad.net"

    code = code.encode("utf8")
    params = {"language": lang, "code": code}

    try:
        result = http.get_json(
            "https://dotnetpad.net/Skybot", post_data=params, get_method="POST"
        )
    except http.HTTPError:
        return "error: unable to connect to dotnetpad"

    if result["Errors"]:
        return "First error: %s" % (result["Errors"][0]["ErrorText"])
    elif result["Output"]:
        return result["Output"].lstrip()
    else:
        return "No output"

@hook.command
def fs(inp):
    ".fs -- post a F# code snippet to dotnetpad.net and print the results"

    return dotnetpad("fsharp", inp)

@hook.command
def cs(snippet):
    ".cs -- post a C# code snippet to dotnetpad.net and print the results"

    file_template = (
        "using System; "
        "using System.Linq; "
        "using System.Collections.Generic; "
        "using System.Text; "
        "%s"
    )

    class_template = "public class Default " "{" "    %s \n" "}"

    main_template = "public static void Main(String[] args) " "{" "    %s \n" "}"

    # There are probably better ways to do the following, but I'm feeling lazy
    # if no main is found in the snippet, use the template with Main in it
    if "public static void Main" not in snippet:
        code = main_template % snippet
        code = class_template % code
        code = file_template % code

    # if Main is found, check for class and see if we need to use the
    # classed template
    elif "class" not in snippet:
        code = class_template % snippet
        code = file_template % code

        return "Error using dotnetpad"
    # if we found class, then use the barebones template
    else:
        code = file_template % snippet

    return dotnetpad("csharp", code)


text:
import re
import random

from util import hook

@hook.command
def choose(inp):
    ".choose <choice1>, <choice2>, ... <choicen> -- makes a decision"

    c = re.findall(r"([^,]+)", inp)
    if len(c) == 1:
        c = re.findall(r"(\S+)", inp)
        if len(c) == 1:
            return "the decision is up to you"

    return random.choice(c).strip()


text:
from __future__ import unicode_literals

from util import hook, http

@hook.command
def calc(inp):
    """.calc <term> -- returns Google Calculator result"""

    h = http.get_html("http://www.google.com/search", q=inp)

    m = h.xpath('//h2[@class="r"]/text()')

    if not m:
        return "could not calculate " + inp

    res = " ".join(m[0].split())

    return res


text:
from __future__ import unicode_literals

import time

from util import hook, timesince

def db_init(db):
    "check to see that our db has the tell table and return a dbection."
    db.execute(
        "create table if not exists tell"
        "(user_to, user_from, message, chan, time,"
        "primary key(user_to, message))"
    )
    db.commit()

    return db

def get_tells(db, user_to):
    return db.execute(
        "select user_from, message, time, chan from tell where"
        " user_to=lower(?) order by time",
        (user_to.lower(),),
    ).fetchall()

@hook.singlethread
@hook.event("PRIVMSG")
def tellinput(paraml, input=None, db=None):
    if "showtells" in input.msg.lower():
        return

    db_init(db)

    tells = get_tells(db, input.nick)

    if tells:
        user_from, message, time, chan = tells[0]
        reltime = timesince.timesince(time)

        reply = "%s said %s ago in %s: %s" % (user_from, reltime, chan, message)
        if len(tells) > 1:
            reply += " (+%d more, .showtells to view)" % (len(tells) - 1)

        db.execute(
            "delete from tell where user_to=lower(?) and message=?",
            (input.nick, message),
        )
        db.commit()
        input.pm(reply)

@hook.command(autohelp=False)
def showtells(inp, nick="", chan="", pm=None, db=None):
    ".showtells -- view all pending tell messages (sent in PM)."

    db_init(db)

    tells = get_tells(db, nick)

    if not tells:
        pm("You have no pending tells.")
        return

    for tell in tells:
        user_from, message, time, chan = tell
        past = timesince.timesince(time)
        pm("%s said %s ago in %s: %s" % (user_from, past, chan, message))

    db.execute("delete from tell where user_to=lower(?)", (nick,))
    db.commit()

@hook.command
def tell(inp, nick="", chan="", db=None, conn=None):
    ".tell <nick> <message> -- relay <message> to <nick> when <nick> is around"

    query = inp.split(" ", 1)

    if len(query) != 2:
        return tell.__doc__

    user_to = query[0].lower()
    message = query[1].strip()
    user_from = nick

    if chan.lower() == user_from.lower():
        chan = "a pm"

    if user_to in (user_from.lower(), conn.nick.lower()):
        return "No."

    db_init(db)

    if (
        db.execute("select count() from tell where user_to=?", (user_to,)).fetchone()[0]
        >= 5
    ):
        return "That person has too many things queued."

    try:
        db.execute(
            "insert into tell(user_to, user_from, message, chan,"
            "time) values(?,?,?,?,?)",
            (user_to, user_from, message, chan, time.time()),
        )
        db.commit()
    except db.IntegrityError:
        return "Message has already been queued."

    return "I'll pass that along."


text:
from util import hook, http

# https://api.esv.org/account/create-application/
@hook.api_key("bible")
@hook.command("god")
@hook.command
def bible(inp, api_key=None):
    ".bible <passage> -- gets <passage> from the Bible (ESV)"

    base_url = (
        "https://api.esv.org/v3/passage/text/?"
        "include-headings=false&"
        "include-passage-horizontal-lines=false&"
        "include-heading-horizontal-lines=false&"
        "include-passage-references=false&"
        "include-short-copyright=false&"
        "include-footnotes=false&"
    )

    text = http.get_json(base_url, q=inp, headers={"Authorization": "Token " + api_key})

    text = " ".join(text["passages"]).strip()

    if len(text) > 400:
        text = text[: text.rfind(" ", 0, 400)] + "..."

    return text

@hook.command("allah")
@hook.command
def koran(inp):  # Koran look-up plugin by Ghetto Wizard
    ".koran <chapter.verse> -- gets <chapter.verse> from the Koran"

    url = "http://quod.lib.umich.edu/cgi/k/koran/koran-idx?type=simple"

    results = http.get_html(url, q1=inp).xpath("//li")

    if not results:
        return "No results for " + inp

    return results[0].text_content()


text:
import re
from functools import wraps
from time import time

from util import hook, http

CDECL_URL = "https://cdecl.org/"
RE_QUERY_ENDPOINT = re.compile('var QUERY_ENDPOINT = "([^"]+)"')
QUERY_URL_MEMO_TTL = 600

def memoize(ttl):
    def _memoizer(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            if hasattr(wrapper, "memo") and wrapper.memo:
                memo_time, result = wrapper.memo
                if time() - memo_time < ttl:
                    return result

            result = f(*args, **kwargs)
            wrapper.memo = (time(), result)
            return result

        return wrapper

    return _memoizer

@memoize(QUERY_URL_MEMO_TTL)
def get_cdecl_query_url():
    print("RUNNING")
    result = http.get(CDECL_URL)

    match = RE_QUERY_ENDPOINT.search(result)

    if not match:
        return None

    return match.group(1)

@hook.command
def cdecl(inp):
    """.cdecl <expr> -- translate between C declarations and English, using cdecl.org"""
    query_url = get_cdecl_query_url()

    if not query_url:
        return "cannot find CDECL query url"

    return http.get(query_url, q=inp)


text:
import random
import re
import time
import unittest

from util import hook

def add_quote(db, chan, nick, add_nick, msg):
    db.execute(
        """insert or fail into quote (chan, nick, add_nick,
                    msg, time) values(?,?,?,?,?)""",
        (chan, nick, add_nick, msg, time.time()),
    )
    db.commit()

def del_quote(db, chan, nick, msg):
    updated = db.execute(
        """update quote set deleted = 1 where
                  chan=? and lower(nick)=lower(?) and msg=?""",
        (chan, nick, msg),
    )
    db.commit()

    if updated.rowcount == 0:
        return False
    else:
        return True

def get_quotes_by_nick(db, chan, nick):
    return db.execute(
        "select time, nick, msg from quote where deleted!=1 "
        "and chan=? and lower(nick)=lower(?) order by time",
        (chan, nick),
    ).fetchall()

def get_quotes_by_chan(db, chan):
    return db.execute(
        "select time, nick, msg from quote where deleted!=1 "
        "and chan=? order by time",
        (chan,),
    ).fetchall()

def get_quote_by_id(db, num):
    return db.execute(
        "select time, nick, msg from quote where deleted!=1 " "and rowid=?", (num,)
    ).fetchall()

def format_quote(q, num, n_quotes):
    ctime, nick, msg = q
    return "[%d/%d] %s <%s> %s" % (
        num,
        n_quotes,
        time.strftime("%Y-%m-%d", time.gmtime(ctime)),
        nick,
        msg,
    )

@hook.command("q")
@hook.command
def quote(inp, nick="", chan="", db=None, admin=False):
    ".q/.quote [#chan] [nick] [#n]/.quote add|delete <nick> <msg> -- gets " "random or [#n]th quote by <nick> or from <#chan>/adds or deletes " "quote"

    db.execute(
        "create table if not exists quote"
        "(chan, nick, add_nick, msg, time real, deleted default 0, "
        "primary key (chan, nick, msg))"
    )
    db.commit()

    add = re.match(r"add[^\w@]+(\S+?)>?\s+(.*)", inp, re.I)
    delete = re.match(r"delete[^\w@]+(\S+?)>?\s+(.*)", inp, re.I)
    retrieve = re.match(r"(\S+)(?:\s+#?(-?\d+))?$", inp)
    retrieve_chan = re.match(r"(#\S+)\s+(\S+)(?:\s+#?(-?\d+))?$", inp)
    retrieve_id = re.match(r"(\d+)$", inp)

    if add:
        quoted_nick, msg = add.groups()
        try:
            add_quote(db, chan, quoted_nick, nick, msg)
            db.commit()
        except db.IntegrityError:
            return "message already stored, doing nothing."
        return "quote added."
    if delete:
        if not admin:
            return "only admins can delete quotes"
        quoted_nick, msg = delete.groups()
        if del_quote(db, chan, quoted_nick, msg):
            return "deleted quote '%s'" % msg
        else:
            return "found no matching quotes to delete"
    elif retrieve_id:
        (quote_id,) = retrieve_id.groups()
        num = 1
        quotes = get_quote_by_id(db, quote_id)
    elif retrieve:
        select, num = retrieve.groups()
        if select.startswith("#"):
            quotes = get_quotes_by_chan(db, select)
        else:
            quotes = get_quotes_by_nick(db, chan, select)
    elif retrieve_chan:
        chan, nick, num = retrieve_chan.groups()

        quotes = get_quotes_by_nick(db, chan, nick)
    else:
        return quote.__doc__

    if num:
        num = int(num)

    n_quotes = len(quotes)

    if not n_quotes:
        return "no quotes found"

    if num:
        if num > n_quotes or (num < 0 and num < -n_quotes):
            return "I only have %d quote%s for %s" % (
                n_quotes,
                ("s", "")[n_quotes == 1],
                select,
            )
        elif num < 0:
            selected_quote = quotes[num]
            num = n_quotes + num + 1
        else:
            selected_quote = quotes[num - 1]
    else:
        num = random.randint(1, n_quotes)
        selected_quote = quotes[num - 1]

    return format_quote(selected_quote, num, n_quotes)

class QuoteTest(unittest.TestCase):
    def setUp(self):
        import sqlite3

        self.db = sqlite3.connect(":memory:")

        quote("", db=self.db)  # init DB

    def quote(self, arg, **kwargs):
        return quote(arg, chan="#test", nick="alice", db=self.db, **kwargs)

    def add_quote(self, msg=""):
        add_quote(
            self.db,
            "#test",
            "Socrates",
            "Plato",
            msg
            or "Education is the kindling of a flame," " not the filling of a vessel.",
        )

    def test_retrieve_chan(self):
        self.add_quote()
        assert "<Socrates> Education" in self.quote("#test")

    def test_retrieve_user(self):
        self.add_quote()
        assert "<Socrates> Education" in self.quote("socrates")

    def test_no_quotes(self):
        assert "no quotes found" in self.quote("#notachan")

    def test_quote_too_high(self):
        self.add_quote()
        assert "I only have 1 quote for #test" in self.quote("#test 4")

    def test_add(self):
        self.quote("add <someone> witty phrase")
        assert "witty" in self.quote("#test")

    def test_add_twice(self):
        self.quote("add <someone> lol")
        assert "already stored" in self.quote("add <someone> lol")

    def test_del_not_admin(self):
        assert "only admins" in self.quote("delete whoever 4")

    def test_del_not_exists(self):
        assert "found no matching" in self.quote("delete whoever 4", admin=True)

    def test_del(self):
        self.add_quote("hi")
        assert "deleted quote 'hi'" in self.quote("delete socrates hi", admin=True)

    def test_retrieve_id(self):
        self.add_quote()
        assert "Education is" in self.quote("1")

    def test_retrieve_chan_user(self):
        self.add_quote()
        assert "Education" in self.quote("#test socrates")
        assert "Education" in self.quote("#test socrates 1")

    def test_nth(self):
        self.add_quote("first quote")
        self.add_quote("second quote")
        self.add_quote("third quote")
        self.add_quote("fourth quote")
        assert "third" in self.quote("socrates -2")
        assert "only have 4" in self.quote("socrates -9")

if __name__ == "__main__":
    unittest.main()


text:
"""Weather, thanks to pirateweather and google geocoding."""
from __future__ import unicode_literals

from util import hook, http

GEOCODING_URL = "https://maps.googleapis.com/maps/api/geocode/json"
PIRATEWEATHER_URL = "https://api.pirateweather.net/forecast/"

def geocode_location(api_key, loc):
    """Get a geocoded location from gooogle's geocoding api."""
    try:
        parsed_json = http.get_json(GEOCODING_URL, address=loc, key=api_key)
    except IOError:
        return None

    return parsed_json

def get_weather_data(api_key, lat, long):
    """Get weather data from pirateweather."""
    query = "{key}/{lat},{long}".format(key=api_key, lat=lat, long=long)
    url = PIRATEWEATHER_URL + query
    try:
        parsed_json = http.get_json(url)
    except IOError:
        return None

    return parsed_json

def f_to_c(temp_f):
    """Convert F to C."""
    return (temp_f - 32) * 5 / 9

def mph_to_kph(mph):
    """Convert mph to kph."""
    return mph * 1.609

@hook.api_key("google", "pirateweather")
@hook.command(autohelp=False)
def weather(inp, chan="", nick="", reply=None, db=None, api_key=None):
    """.weather <location> [dontsave] | @<nick> -- Get weather data."""
    if "google" not in api_key and "pirateweather" not in api_key:
        return None

    # this database is used by other plugins interested in user's locations,
    # like .near in tag.py
    db.execute(
        "create table if not exists "
        "location(chan, nick, loc, lat, lon, primary key(chan, nick))"
    )

    if inp[0:1] == "@":
        nick = inp[1:].strip()
        loc = None
        dontsave = True
    else:
        dontsave = inp.endswith(" dontsave")
        # strip off the " dontsave" text if it exists and set it back to `inp`
        # so we don't report it back to the user incorrectly
        if dontsave:
            inp = inp[:-9].strip().lower()
        loc = inp

    if not loc:  # blank line
        loc = db.execute(
            "select loc, lat, lon from location where chan=? and nick=lower(?)",
            (chan, nick),
        ).fetchone()
        if not loc:
            return weather.__doc__
        addr, lat, lng = loc
    else:
        location = geocode_location(api_key["google"], loc)

        if not location or location.get("status") != "OK":
            reply("Failed to determine location for {}".format(inp))
            return

        geo = location.get("results", [{}])[0].get("geometry", {}).get("location", None)
        if not geo or "lat" not in geo or "lng" not in geo:
            reply("Failed to determine location for {}".format(inp))
            return

        addr = location["results"][0]["formatted_address"]
        lat = geo["lat"]
        lng = geo["lng"]

    parsed_json = get_weather_data(api_key["pirateweather"], lat, lng)
    current = parsed_json.get("currently")

    if not current:
        reply("Failed to get weather data for {}".format(inp))
        return

    forecast = parsed_json["daily"]["data"][0]

    info = {
        "city": addr,
        "t_f": current["temperature"],
        "t_c": f_to_c(current["temperature"]),
        "h_f": forecast["temperatureHigh"],
        "h_c": f_to_c(forecast["temperatureHigh"]),
        "l_f": forecast["temperatureLow"],
        "l_c": f_to_c(forecast["temperatureLow"]),
        "weather": current["summary"],
        "humid": int(current["humidity"] * 100),
        "wind": "Wind: {mph:.1f}mph/{kph:.1f}kph".format(
            mph=current["windSpeed"], kph=mph_to_kph(current["windSpeed"])
        ),
        "forecast": parsed_json.get("hourly", {}).get("summary", ""),
    }
    reply(
        "{city}: {weather}, {t_f:.1f}F/{t_c:.1f}C"
        "(H:{h_f:.1f}F/{h_c:.1f}C L:{l_f:.1f}F/{l_c:.1f}C)"
        ", Humidity: {humid}%, {wind} \x02{forecast}\x02".format(**info)
    )

    if inp and not dontsave:
        db.execute(
            "insert or replace into "
            "location(chan, nick, loc, lat, lon) "
            "values (?, ?, ?, ?, ?)",
            (chan, nick.lower(), addr, lat, lng),
        )
        db.commit()


text:
from __future__ import unicode_literals

from util import http, hook

@hook.command(autohelp=False)
def bitcoin(inp, say=None):
    ".bitcoin -- gets current exchange rate for bitcoins from Bitstamp"
    data = http.get_json("https://www.bitstamp.net/api/ticker/")
    say(
        "USD/BTC: \x0307${:,.2f}\x0f - High: \x0307${:,.2f}\x0f -"
        " Low: \x0307${:,.2f}\x0f - Volume: {:,.2f} BTC".format(
            float(data["last"]),
            float(data["high"]),
            float(data["low"]),
            float(data["volume"]),
        )
    )

@hook.command(autohelp=False)
def ethereum(inp, say=None):
    ".ethereum -- gets current exchange rate for ethereum from Bitstamp"
    data = http.get_json("https://www.bitstamp.net/api/v2/ticker/ethusd")
    say(
        "USD/ETH: \x0307${:,.2f}\x0f - High: \x0307${:,.2f}\x0f -"
        " Low: \x0307${:,.2f}\x0f - Volume: {:,.2f} ETH".format(
            float(data["last"]),
            float(data["high"]),
            float(data["low"]),
            float(data["volume"]),
        )
    )


text:
from __future__ import unicode_literals

import re

from util import hook, http

@hook.command("u")
@hook.command
def urban(inp):
    """.u/.urban <phrase> -- looks up <phrase> on urbandictionary.com"""

    page = http.get_json("https://api.urbandictionary.com/v0/define", term=inp)
    defs = page["list"]

    if not defs:
        return "not found."

    out = defs[0]["word"] + ": " + defs[0]["definition"].replace("\r\n", " ")

    if len(out) > 400:
        out = out[: out.rfind(" ", 0, 400)] + "..."

    return out

# define plugin by GhettoWizard & Scaevolus
@hook.command("dictionary")
@hook.command
def define(inp):
    ".define/.dictionary <word> -- fetches definition of <word>"

    url = "http://ninjawords.com/"

    h = http.get_html(url + http.quote_plus(inp))

    definition = h.xpath(
        '//dd[@class="article"] | '
        '//div[@class="definition"] |'
        '//div[@class="example"]'
    )

    if not definition:
        return "No results for " + inp

    def format_output(show_examples):
        result = "%s: " % h.xpath('//dt[@class="title-word"]/a/text()')[0]

        correction = h.xpath('//span[@class="correct-word"]/text()')
        if correction:
            result = 'definition for "%s": ' % correction[0]

        sections = []
        for section in definition:
            if section.attrib["class"] == "article":
                sections += [[section.text_content() + ": "]]
            elif section.attrib["class"] == "example":
                if show_examples:
                    sections[-1][-1] += " " + section.text_content()
            else:
                sections[-1] += [section.text_content()]

        for article in sections:
            result += article[0]
            if len(article) > 2:
                result += " ".join(
                    "%d. %s" % (n + 1, section) for n, section in enumerate(article[1:])
                )
            else:
                result += article[1] + " "

        synonyms = h.xpath('//dd[@class="synonyms"]')
        if synonyms:
            result += synonyms[0].text_content()

        result = re.sub(r"\s+", " ", result)
        result = re.sub("\xb0", "", result)
        return result

    result = format_output(True)
    if len(result) > 450:
        result = format_output(False)

    if len(result) > 450:
        result = result[: result.rfind(" ", 0, 450)]
        result = re.sub(r"[^A-Za-z]+\.?$", "", result) + " ..."

    return result

@hook.command("e")
@hook.command
def etymology(inp):
    ".e/.etymology <word> -- Retrieves the etymology of chosen word"

    h = http.get_html("https://www.etymonline.com/search", q=inp)

    etym = h.xpath("string(//section[1])")

    if not etym:
        return "No etymology found for " + inp

    etym = etym.replace(inp, "\x02%s\x02" % inp)

    if len(etym) > 400:
        etym = etym[: etym.rfind(" ", 0, 400)] + " ..."

    return etym


text:
from __future__ import unicode_literals

from util import hook, http

MORBOTRON_URL = "https://morbotron.com"
FRINKIAC_URL = "https://frinkiac.com"
MASTEROFALLSCIENCE_URL = "https://masterofallscience.com"

SEARCH_ENDPOINT = "/api/search"
CAPTIONS_ENDPOINT = "/api/caption"
LINK_URL = "/caption"

def get_frames(site_root, query):
    try:
        url = site_root + SEARCH_ENDPOINT
        content = http.get_json(url, q=query)
    except IOError:
        return None

    return content

def get_caption(site_root, episode, timestamp):
    try:
        url = site_root + CAPTIONS_ENDPOINT
        content = http.get_json(url, e=episode, t=timestamp)
    except IOError:
        return None

    return content

def get_response(site_root, input):
    frames = get_frames(site_root, input)

    if frames is None or len(frames) == 0:
        return 'no results'

    frame = frames[0]

    episode = frame['Episode']
    timestamp = frame['Timestamp']

    caption = get_caption(site_root, episode, timestamp)
    url = '{}/{}/{}'.format(site_root + LINK_URL, episode, timestamp)

    subtitles = caption['Subtitles']
    if len(subtitles) == 0:
        return '{} - {}'.format(episode, url)

    subtitle = ' '.join(s['Content'] for s in subtitles)

    return '{} - {} {} '.format(subtitle, episode, url)

@hook.command('simpsons')
@hook.command()
def frinkiac(inp):
    """.frinkiac <query> -- Get a frame from The Simpsons."""
    return get_response(FRINKIAC_URL, inp)

@hook.command('futurama')
@hook.command()
def morbotron(inp):
    """.morbotron <query> -- Get a frame from Futurama."""
    return get_response(MORBOTRON_URL, inp)

@hook.command('morty')
@hook.command('rick')
@hook.command()
def rickandmorty(inp):
    """.rickandmorty <query> -- Get a frame from Rick and Morty."""
    return get_response(MASTEROFALLSCIENCE_URL, inp)

@hook.regex(r"(?i)https://(frinkiac|morbotron|masterofallscience)\.com"
            r"/(caption|img)/S(\d{2})E(\d{2})/(\d+)")
def lookup(match):
    domain = match.group(1)

    season_episode = 'S{}E{}'.format(match.group(3), match.group(4))
    timestamp = match.group(5)

    urls = {
        'frinkiac': FRINKIAC_URL,
        'morbotron': MORBOTRON_URL,
        'masterofallscience': MASTEROFALLSCIENCE_URL
    }

    site_root = urls[domain]
    caption = get_caption(site_root, season_episode, timestamp)

    episode = caption['Episode']
    episode_key = episode['Key']
    title = episode['Title']

    subtitles = caption['Subtitles']
    if len(subtitles) == 0:
        return '{} - {}'.format(episode_key, title)

    subtitle = ' '.join(s['Content'] for s in subtitles)

    return '{} - {}'.format(subtitle, episode_key)


text:
import re

from util import hook

@hook.command(autohelp=False)
def help(inp, bot=None, pm=None):
    ".help [command] -- gives a list of commands/help for a command"

    funcs = {}
    disabled = bot.config.get("disabled_plugins", [])
    disabled_comm = bot.config.get("disabled_commands", [])
    for command, (func, args) in bot.commands.items():
        fn = re.match(r"^plugins.(.+).py$", func._filename)
        if fn.group(1).lower() not in disabled:
            if command not in disabled_comm:
                if func.__doc__ is not None:
                    if func in funcs:
                        if len(funcs[func]) < len(command):
                            funcs[func] = command
                    else:
                        funcs[func] = command

    commands = dict((value, key) for key, value in iter(funcs.items()))

    if not inp:
        pm("available commands: " + " ".join(sorted(commands)))
    else:
        if inp in commands:
            pm(commands[inp].__doc__)


text:
from __future__ import unicode_literals

import json
import re

from util import hook, http

SEARCH_URL = (
    "https://yfrdx308zd-dsn.algolia.net/1/indexes/wp_live_searchable_posts/query"
    "?x-algolia-application-id=YFRDX308ZD&x-algolia-api-key=7da15c5275374261c3a4bdab2ce5d321"
)

@hook.command
def snopes(inp):
    ".snopes <topic> -- searches snopes for an urban legend about <topic>"

    params = {"params": http.urlencode({"query": inp, "hitsPerPage": 5})}
    results = http.get_json(SEARCH_URL, post_data=json.dumps(params).encode("utf8"))[
        "hits"
    ]

    if not results:
        return "no matching pages found"

    post = results[0]
    for post in results:
        if post["post_type"] == "fact_check":
            return fmt(post)

def fmt(post):
    permalink = post["permalink"]

    if "fact_check_claim" in post:
        claim = post["fact_check_claim"]
        if post["taxonomies"].get("fact_check_category") == ["Fake News"]:
            status = "Fake News"
        else:
            status = post["taxonomies"]["fact_check_rating"][0]
    else:
        content = post["content"]
        m = re.search(r"(?:Claim|Glurge|Legend|FACT CHECK): (.*)", content)
        if m:
            claim = m.group(1).strip()
        else:
            claim = content.split("\n")[0]
            print("???", claim)
        if claim == "Claim":
            print("!!!", content)
        m = re.search(
            r"FALSE|TRUE|MIXTURE|UNDETERMINED|CORRECTLY ATTRIBUTED|(?<=Status:).*",
            content,
        )
        if m:
            status = m.group(0)
        else:
            status = "???"

    claim = re.sub(
        r"[\s\xa0]+", " ", http.unescape(claim)
    ).strip()  # compress whitespace
    status = re.sub(r"[\s\xa0]+", " ", http.unescape(status)).title().strip()

    if len(claim) > 300:
        claim = claim[:300] + "..."

    return "Claim: {0} Status: {1} {2}".format(claim, status, permalink)

if __name__ == "__main__":
    a = http.get_json(
        SEARCH_URL,
        post_data=json.dumps(
            {"params": http.urlencode({"query": "people", "hitsPerPage": 1000})}
        ).encode("utf8"),
    )
    print(len(a["hits"]))
    try:
        for x in a["hits"]:
            if x["post_type"] == "fact_check":
                f = fmt(x)
                print(f)
                assert len(f) < 400
    except AttributeError:
        print(x["permalink"])
        print(x["content"])
        raise


text:
"""
The Last.fm API key is retrieved from the bot config file.
"""

from util import hook, http

api_url = "http://ws.audioscrobbler.com/2.0/?format=json"

@hook.api_key("lastfm")
@hook.command(autohelp=False)
def lastfm(inp, chan="", nick="", reply=None, api_key=None, db=None):
    ".lastfm <username> [dontsave] | @<nick> -- gets current or last played " "track from lastfm"

    db.execute(
        "create table if not exists "
        "lastfm(chan, nick, user, primary key(chan, nick))"
    )

    if inp[0:1] == "@":
        nick = inp[1:].strip()
        user = None
        dontsave = True
    else:
        user = inp

        dontsave = user.endswith(" dontsave")
        if dontsave:
            user = user[:-9].strip().lower()

    if not user:
        user = db.execute(
            "select user from lastfm where chan=? and nick=lower(?)", (chan, nick)
        ).fetchone()
        if not user:
            return lastfm.__doc__
        user = user[0]

    response = http.get_json(
        api_url, method="user.getrecenttracks", api_key=api_key, user=user, limit=1
    )

    if "error" in response:
        return "error: %s" % response["message"]

    if (
        not "track" in response["recenttracks"]
        or len(response["recenttracks"]["track"]) == 0
    ):
        return "no recent tracks for user \x02%s\x0F found" % user

    tracks = response["recenttracks"]["track"]

    if type(tracks) == list:
        # if the user is listening to something, the tracks entry is a list
        # the first item is the current track
        track = tracks[0]
        status = "current track"
    elif type(tracks) == dict:
        # otherwise, they aren't listening to anything right now, and
        # the tracks entry is a dict representing the most recent track
        track = tracks
        status = "last track"
    else:
        return "error parsing track listing"

    title = track["name"]
    album = track["album"]["#text"]
    artist = track["artist"]["#text"]

    ret = "\x02%s\x0F's %s - \x02%s\x0f" % (user, status, title)
    if artist:
        ret += " by \x02%s\x0f" % artist
    if album:
        ret += " on \x02%s\x0f" % album

    reply(ret)

    if inp and not dontsave:
        db.execute(
            "insert or replace into lastfm(chan, nick, user) " "values (?, ?, ?)",
            (chan, nick.lower(), inp),
        )
        db.commit()


text:
from __future__ import unicode_literals

from util import http, hook

@hook.command()
def crypto(inp, say=None):
    fsym = inp.split(" ")[0].upper()
    price_response = http.get_json(
        "https://min-api.cryptocompare.com/data/pricemultifull", fsyms=fsym, tsyms="USD"
    )

    if price_response.get("Response") == "Error":
        return "Error: " + price_response.get(
            "Message", "An API Error Has Occured. https://min-api.cryptocompare.com/"
        )

    if not price_response["DISPLAY"].get(fsym):
        return "Error: Unable to lookup pricing for {}".format(fsym)

    values = price_response["DISPLAY"][fsym]["USD"]

    say(
        (
            "USD/{FROMSYMBOL}: \x0307{PRICE}\x0f - High: \x0307{HIGHDAY}\x0f "
            "- Low: \x0307{LOWDAY}\x0f - Volume: {VOLUMEDAY}"
            " ({VOLUMEDAYTO}) - Total Supply: {SUPPLY} - MktCap: {MKTCAP}"
        ).format(**values)
    )