<repo_name>eeue56/slack-today-i-did<file_sep>/run_tests.sh
#! /usr/bin/env bash
set -e

# make sure that there's no syntax errors
python3 -c "import main"

# run flake8 on everything
flake8 main.py slack_today_i_did

# run doctest
python3 -m doctest slack_today_i_did/*.py

python3 -m pytest
<file_sep>/slack_today_i_did/our_repo.py
"""
This file lets you do things with a repo. At the moment, the ElmRepo class
can be used in order to get meta info on Elm files in a repo,
such as the number of 0.16/0.17 files
"""

import os
import glob
from typing import List, Dict
from enum import Enum
from contextlib import contextmanager


class OurRepo(object):
    def __init__(self, folder: str, token: str, org: str, repo: str):
        self.folder = folder
        self.token = token
        self.org = org
        self.repo = repo

    def make_repo_dir(self) -> None:
        os.makedirs(self.folder, exist_ok=True)

    def _git_init(self) -> None:
        url = f'https://{self.token}@github.com/{self.org}/{self.repo}.git'
        os.system(f'git clone --depth 1 {url}')
        os.chdir(self.repo)
        os.system(f'git remote set-url origin {url}')

    def _git_clone(self, branch_name: str = 'master') -> None:
        os.system(f'git remote set-branches origin {branch_name}')
        os.system(f'git fetch --depth 1 origin {branch_name}')
        os.system(f'git checkout origin/{branch_name}')

    def get_ready(self, branch_name: str = 'master') -> None:
        current_dir = os.getcwd()
        try:
            self.make_repo_dir()
            os.chdir(self.folder)
            self._git_init()
        except:
            pass
        finally:
            os.chdir(current_dir)

        os.chdir(self.repo_dir)
        try:
            self._git_clone(branch_name)
        finally:
            os.chdir(current_dir)

    @property
    def repo_dir(self):
        return f'{self.folder}/{self.repo}'


class ElmVersion(Enum):
    unknown = -1
    v_016 = 0
    v_017 = 1


class ElmRepo(OurRepo):
    def __init__(self, *args, **kwargs):
        OurRepo.__init__(self, *args, **kwargs)
        self._known_files = {ElmVersion.v_016: [], ElmVersion.v_017: []}
        self._breakdown_cache = {}
        self._import_cache = {}
        self._caching_lookups = False

    def get_elm_files(self) -> List[str]:
        return glob.glob(f'{self.repo_dir}/**/*.elm', recursive=True)

    @property
    def number_of_017_files(self):
        elm_017_count = 0

        for file in self.get_elm_files():
            if self.what_kinda_file(file) == ElmVersion.v_017:
                elm_017_count += 1

        return elm_017_count

    @property
    def number_of_016_files(self):
        elm_016_count = 0

        for file in self.get_elm_files():
            if self.what_kinda_file(file) == ElmVersion.v_016:
                elm_016_count += 1

        return elm_016_count

    def get_files_for_017(self, pattern: str) -> List[str]:
        pattern = pattern.replace('.', '/')
        all_files = glob.glob(
            f'{self.repo_dir}/**/{pattern}.elm',
            recursive=True
        )

        return [
            filename for filename in all_files
            if self.what_kinda_file(filename) == ElmVersion.v_017
        ]

    def get_matching_filenames(self, pattern: str) -> List[str]:
        pattern = pattern.replace('.', '/')
        all_files = glob.glob(
            f'{self.repo_dir}/**/{pattern}.elm',
            recursive=True
        )
        return all_files

    @contextmanager
    def cached_lookups(self) -> None:
        self.create_cache()
        yield
        self.end_cache()

    def create_cache(self) -> None:
        """ create a cache so they can be used to make some calcs faster """
        self._caching_lookups = True
        self._breakdown_cache = {}
        self._import_cache = {}

    def end_cache(self) -> None:
        """ removes a lock so caches are no longer used """
        self._caching_lookups = False

    def get_017_porting_breakdown(self, pattern: str) -> Dict[str, Dict[str, int]]:  # noqa: E501
        all_files = self.get_matching_filenames(pattern)

        breakdown = {
            filename: self.how_hard_to_port(filename) for filename in all_files
            if self.what_kinda_file(filename) == ElmVersion.v_016
        }

        if len(all_files) == 1:
            imports = self.file_import_list(all_files[0])
            for import_ in imports:
                file_names = self.get_matching_filenames(import_)

                if not any(name in breakdown for name in file_names):
                    import_breakdown = self.get_017_porting_breakdown(import_)
                    breakdown.update(import_breakdown)

        return breakdown

    def file_import_list(self, filename: str) -> List[str]:
        """ returns the list of modules imported by a file """

        if self._caching_lookups:
            if filename in self._import_cache:
                return self._import_cache[filename]

        import_lines = []
        in_comment = False

        with open(filename) as f:
            for line in f:
                if in_comment:
                    if line.strip().endswith('-}'):
                        in_comment = False
                elif line.startswith('import '):
                    just_the_module = 'import '.join(line.split('import ')[1:])
                    import_lines.append(just_the_module)
                elif line.strip().startswith('{-'):
                    if line.strip().endswith('-}'):
                        in_comment = False
                    else:
                        in_comment = True
                elif not (line.startswith('module') or line.startswith(' ')):
                    # we're past the imports
                    break

        if self._caching_lookups:
            self._import_cache[filename] = import_lines

        return import_lines

    def how_hard_to_port(self, filename: str) -> Dict[str, int]:
        """ returns a breakdown of how hard a file is to port 0.16 -> 0.17 """

        if self._caching_lookups:
            if filename in self._breakdown_cache:
                return self._breakdown_cache[filename]

        breakdown = {}

        with open(filename) as f:
            text = f.read()

        if '\nport' in text or 'Signal' in text:
            port_count = text.count('\nport')
            signal_count = text.count('Signal')
            breakdown['Ports and signals'] = (port_count + signal_count) * 3

        if 'import Native' in text:
            native_modules_count = text.count('import Native')
            breakdown['Native modules imported'] = (native_modules_count * 2)

        if ' Html' in text:
            html_count = text.count(' Html')
            breakdown['Html stuff'] = html_count

        if self._caching_lookups:
            self._breakdown_cache[filename] = breakdown

        return breakdown

    def what_kinda_file(self, filename: str) -> ElmVersion:
        """ if a filename is known to be 0.16 or 0.17, return that const
            otherwise, go through line by line to try and find some identifiers
        """
        if filename in self._known_files[ElmVersion.v_016]:
            return ElmVersion.v_016

        if filename in self._known_files[ElmVersion.v_017]:
            return ElmVersion.v_017

        with open(filename) as f:
            for line in f:
                if line.strip():
                    if 'exposing' in line:
                        self._known_files[ElmVersion.v_017].append(filename)
                        return ElmVersion.v_017
                    if 'where' in line:
                        self._known_files[ElmVersion.v_016].append(filename)
                        return ElmVersion.v_016
        return ElmVersion.unknown
<file_sep>/slack_today_i_did/reports.py
import json
from typing import Tuple, Dict, Any
import time
import datetime


class Report(object):
    def __init__(
            self,
            channel: str,
            name: str,
            time_to_run: Tuple[int, int],
            people,
            wait: Tuple[int, int],
            time_run=None,
            reports_dir='reports',
    ):
        self.name = name
        self.people_to_bother = people
        self.channel = channel
        self.responses = {}
        self.time_run = time_run
        self.wait_for = wait
        self.time_to_run = time_to_run
        self.reports_dir = reports_dir
        self.is_ended = False
        self.last_day_run = None

    def bother_people(self):
        if not self.people_to_bother:
            return []

        self.responses = {}
        self.time_run = datetime.datetime.utcnow()
        messages = []
        messages.append((self.channel, 'Starting my report!'))

        for person in self.people_to_bother[:]:
            messages.append((person, 'Sorry to bother you!'))
            self.add_response(person, '')
            self.people_to_bother.remove(person)

        return messages

    def is_time_to_bother_people(self) -> bool:
        if self.is_ended:
            return False

        current_time = time.time()
        time_now_string = time.strftime(
            "%j:%Y %H:%M:",
            time.gmtime(current_time)
        )

        hour_mins = time_now_string.split(' ')[1].split(':')
        (hours, minutes) = hour_mins[0], hour_mins[1]
        time_now = (int(hours), int(minutes))

        if time_now < self.time_to_run:
            return False

        if self.time_run is not None:
            day_year = time_now_string.split(' ')[0].split(':')
            (day_in_year, year) = day_year[0], day_year[1]
            (time_run_day_in_year, time_run_year) = self.time_run.strftime('%j:%Y').split(':')  # noqa: E501

            if int(year) == int(time_run_year) and int(day_in_year) == int(time_run_day_in_year):  # noqa: E501
                return False

        self.is_ended = False
        return True

    def is_time_to_end(self) -> bool:
        if self.is_ended:
            return False

        current_time = time.time()
        time_now = time.strftime("%H:%M:", time.gmtime(current_time))
        time_now = (int(time_now.split(':')[0]), int(time_now.split(':')[1]))

        end_time = (self.time_to_run[0] + self.wait_for[0], self.time_to_run[1] + self.wait_for[1])  # noqa: E501

        if time_now < end_time:
            return False

        self.is_ended = True
        return True

    def is_for_user(self, user):
        return user in self.responses or user in self.people_to_bother

    def add_response(self, user, message):
        if user not in self.responses:
            self.responses[user] = message
        else:
            if len(self.responses[user]) == 0:
                self.responses[user] = message
            else:
                self.responses[user] += '\n' + message
        self.save_responses()

    def save_responses(self):
        with open(f'{self.reports_dir}/report-{self.name}-{self.channel}-{self.time_run}.json', 'w') as f:  # noqa: E501
            json.dump(self.responses, f)

    def save(self):
        with open(f'{self.reports_dir}/report-config-{self.name}-{self.channel}.json', 'w') as f:  # noqa: E501
            json.dump(f, self.as_dict())

    def as_dict(self):
        return {
            'name': self.name,
            'channel': self.channel,
            'responses': self.responses,
            'people_to_bother': self.people_to_bother,
            'time_run': (str(self.time_run) if self.time_run is not None else "")  # noqa: E501
        }


class Sessions(object):
    def __init__(self):
        self.sessions = {}

    def has_running_session(self, person: str) -> bool:
        if person not in self.sessions:
            return False

        return self.sessions[person]['is_running']

    def start_session(self, person: str, channel: str) -> None:
        self.sessions[person] = {
            'is_running': True,
            'messages': [],
            'channel': channel
        }

    def end_session(self, person: str) -> None:
        if not self.has_running_session(person):
            return

        self.sessions[person]['is_running'] = False

    def add_message(self, person: str, message: str) -> None:
        if not self.has_running_session(person):
            return

        self.sessions[person]['messages'].append(message)

    def get_entry(self, person: str) -> Dict[str, Any]:
        if person not in self.sessions:
            return {}

        return self.sessions[person]

    def retire_session(self, person: str, filename: str) -> None:
        session_info = self.sessions.pop(person)

        with open(filename, 'w') as f:
            json.dump(session_info, f)

    def load_from_file(self, filename: str) -> None:
        """ Load people:session from a file """
        try:
            with open(filename) as f:
                as_json = json.load(f)
        except FileNotFoundError:
            return

        for (name, session) in as_json['sessions'].items():
            self.sessions[name] = session

    def save_to_file(self, filename: str) -> None:
        """ save people:sessions to a file """
        with open(filename, 'w') as f:
            json.dump({'sessions': self.sessions}, f)
<file_sep>/slack_today_i_did/generic_bot.py
"""
This file contains the bot itself

To add a new function:
    - add an entry in `known_functions`. The key is the command the bot
        will know, the value is the command to run
    - You must add type annotitions for the bot to match args up correctly

"""

import html

from typing import List, Union, NamedTuple

from slack_today_i_did.better_slack import BetterSlack
from slack_today_i_did.command_history import CommandHistory

import slack_today_i_did.self_aware as self_aware

import slack_today_i_did.parser as parser
import slack_today_i_did.text_tools as text_tools


ChannelMessage = NamedTuple('ChannelMessage', [('channel', str), ('text', str)])
ChannelMessages = Union[ChannelMessage, List[ChannelMessage]]


class GenericSlackBot(BetterSlack):
    _user_id = None
    _last_sender = None

    def __init__(self, *args, **kwargs):
        BetterSlack.__init__(self, *args, **kwargs)
        self.name = 'generic-slack-bot'

        self.command_history = CommandHistory()

    def is_direct_message(self, channel):
        """ Direct messages start with `D`
        """
        return channel.startswith('D')

    def was_directed_at_me(self, text):
        return text.startswith(f'<@{self.user_id}>')

    def parse_direct_message(self, message):
        self._actually_parse_message(message)

    async def main_loop(self):
        await BetterSlack.main_loop(
            self,
            parser=self.parse_messages,
            on_tick=self.on_tick
        )

    def known_tokens(self) -> List[str]:
        return list(self.known_functions().keys())

    def known_functions(self):
        return {**self.known_user_functions(), **self.known_statements()}

    def known_user_functions(self):
        return {
            'func-that-return': self.functions_that_return,
            'error-help': self.error_help,
            'help': self.help,
            'possible-funcs': self.possible_funcs,
            'list': self.list,
            'reload-funcs': self.reload_functions,
        }

    def known_statements(self):
        return {
            '!!': self.last_command_statement
        }

    def on_tick(self):
        pass

    @property
    def user_id(self):
        if self._user_id is None:
            data = self.connected_user(self.name)
            self._user_id = data

        return self._user_id

    def parse(self, text, channel):
        """ Take text, a default channel,
        """

        # we only tokenize those that talk to me
        if self.was_directed_at_me(text):
            user_id_string = f'<@{self.user_id}>'

            if text.startswith(user_id_string):
                text = text[len(user_id_string):].strip()
        else:
            return None

        tokens = parser.tokenize(text, self.known_tokens())
        return parser.parse(tokens, self.known_functions())

    def _actually_parse_message(self, message):
        channel = message['channel']
        text = message['text']

        stuff = self.parse(text, channel)

        if stuff is None:
            return

        func_call = stuff.func_call
        evaluate = stuff.evaluate

        # we always give the channel as the first arg
        default_args = [parser.Constant(channel, str)]
        try:
            evaluation = evaluate(func_call, default_args)

            # deal with exceptions running the command
            if len(evaluation.errors) > 0:
                self.send_channel_message(channel, '\n\n'.join(evaluation.errors))
                return

            if func_call.return_type == ChannelMessages:
                if isinstance(evaluation.result, ChannelMessage):
                    messages = [evaluation.result]
                else:
                    messages = evaluation.result

                for message in messages:
                    self.send_channel_message(message.channel, message.text)

            if evaluation.action != self.known_statements()['!!']:
                self.command_history.add_command(channel, evaluation.action, evaluation.args)
        except Exception as e:
            self.send_channel_message(channel, f'We got an error {e}!')

    def parse_message(self, message):
        # if we don't have any of the useful data, return early

        if 'type' not in message or 'text' not in message:
            return None

        if message['type'] != 'message':
            return None

        self._last_sender = message.get('user', None)

        # if it's a direct message, parse it differently
        if self.is_direct_message(message['channel']):
            return self.parse_direct_message(message)

        return self._actually_parse_message(message)

    def parse_messages(self, messages):
        for message in messages:
            self.parse_message(message)

    def error_help(self, channel: str, problem: str) -> ChannelMessages:
        """ present an error help message """
        if problem == 'NO_TOKENS':
            return ChannelMessage(channel, 'I\'m sorry, I couldn\'t find any tokens. Try using `help` or `list`')  # noqa: E501
        else:
            return ChannelMessage(channel, f'Some problem: {problem}')

    def last_command_statement(self, channel: str) -> ChannelMessages:
        """ run the last command again """

        stuff = self.command_history.last_command(channel)

        if stuff is None:
            self.send_channel_message(channel, 'No commands have been run yet!')
            return

        action = stuff['action']
        args = stuff['args']

        return action(*args)

    def list(self, channel: str) -> ChannelMessages:
        """ list known statements and functions """

        message = 'Main functions:\n'
        message += '\n'.join(
            f'`{func}`'for func in self.known_user_functions()
        )

        message += '\nStatements:\n'
        message += '\n'.join(
            f'`{func}`'for func in self.known_statements()
        )

        return ChannelMessage(channel, message)

    def possible_funcs(self, channel: str, name: str) -> ChannelMessages:
        """ give me a name and I'll tell you funcs which are close """

        known_functions = self.known_functions()

        # default the length of the name
        acceptable_score = len(name)

        # but for long names, we want to cap it a bit
        if acceptable_score > 10:
            acceptable_score = 10
        elif acceptable_score > 5:
            acceptable_score = 4

        possibles = possible_functions(known_functions, name, acceptable_score=acceptable_score)

        if len(possibles) == 0:
            return ChannelMessage(channel, "I don't know what you mean and have no suggestions")

        message = f'I don\'t know about `{name}`. But I did find the following functions with similiar names:\n'
        message += ' | '.join(possibles[:5])

        return ChannelMessage(channel, message)

    # TODO: args should be annotated with `List[parser.FuncArg]` once
    # https://github.com/python/typing/issues/306 is resolved.
    @parser.metafunc
    def help(self, channel: str, args) -> ChannelMessages:
        """ given a function name, I'll tell you about it
        """
        if not len(args):
            return self.list(channel)

        if isinstance(args[0], parser.Constant):
            func_name = args[0].value
        else:
            func_name = args[0].func_name
        known_functions = self.known_functions()

        if func_name not in known_functions:
            return self.possible_funcs(channel, func_name)

        func = known_functions[func_name]
        docs = ' '.join(line.strip() for line in func.__doc__.split('\n'))

        message = f'`{func_name}` has the help message:\n{docs}\n'

        if not parser.is_metafunc(func):
            message += 'And the type info:\n```\n'

            type_info = '\n'.join(
                f'- {arg_name} : {arg_type}'
                for (arg_name, arg_type) in func.__annotations__.items()
            )

            message += f'{type_info}\n```'

        return ChannelMessage(channel, message)

    def reload_functions(self, channel: str) -> ChannelMessages:
        """ reload the functions a bot knows """
        self_aware.restart_program()
        return []

    def functions_that_return(self, channel: str, text: str) -> ChannelMessages:
        """ give a type, return functions that return things of that type
        """
        func_names = []
        text = text.strip()
        text = html.unescape(text)

        for (name, func) in self.known_functions().items():
            if str(func.__annotations__.get('return', None)) == text:
                func_names.append((name, func.__annotations__))

        message = f"The following functions return `{text}`:\n"
        message += '```\n'
        message += '\n'.join(name for (name, type) in func_names)
        message += '\n```'

        return ChannelMessage(channel, message)


def possible_functions(known_functions, name, acceptable_score=5):
    possibles = [
        (text_tools.token_based_levenshtein(func_name, name), func_name) for func_name in known_functions
    ]

    possibles = [x for x in possibles if x[0] < acceptable_score]
    possibles = sorted(possibles, key=lambda x: x[0])
    possibles = [x[1] for x in possibles]

    return possibles


class BotExtension(GenericSlackBot):
    def __init__(self, *args, **kwargs):
        pass
<file_sep>/setup.py
#!/usr/bin/env python

from distutils.core import setup

setup(
    name='slack_today_i_did',
    version='1.0',
    description='A slack bot for NoRedInk',
    author='<NAME>',
    author_email='<EMAIL>',
    url='https://www.github.com/NoRedInk/slack-today-i-did',
    packages=['slack_today_i_did'],
)
<file_sep>/requirements.txt
json-tricks==3.2.0
mypy-lang==0.4.4
requests==2.11.1
slackclient==1.0.2
typed-ast==0.5.6
websockets==3.2
psutil==4.3.1
prompt-toolkit==1.0.8
<file_sep>/slack_today_i_did/notify.py
import json
from typing import List
import re


class Notification(object):
    """ Allows you to register multiple regex patterns with a person
        This is designed to let you notify users based on an incoming pattern
    """

    def __init__(self):
        self.patterns = {}

    def add_pattern(self, person: str, pattern: str) -> None:
        """ register a pattern to notify a given person
        """

        if person not in self.patterns:
            self.patterns[person] = []

        self.patterns[person].append(pattern)

    def forget_pattern(self, person: str, pattern: str) -> None:
        """ stop notifying a person for a given pattern
        """

        if person not in self.patterns:
            return

        if pattern in self.patterns[person]:
            self.patterns[person].remove(pattern)

    def who_wants_it(self, text: str) -> None:
        """ returns a list of people that want to be notified by
            a message that matches any of the registered patterns
        """
        who_wants_it = []

        for (person, patterns) in self.patterns.items():
            for pattern in patterns:
                if re.search(pattern, text, re.MULTILINE) is not None:
                    who_wants_it.append(person)
                    break

        return who_wants_it

    def get_patterns(self, person: str) -> List[str]:
        """ get a list of patterns for a person
        """
        if person in self.patterns:
            return self.patterns[person]
        return []

    def load_from_file(self, filename: str) -> None:
        """ Load people:patterns from a file """
        try:
            with open(filename) as f:
                as_json = json.load(f)
        except FileNotFoundError:
            return

        for (name, patterns) in as_json['patterns'].items():
            self.patterns[name] = patterns

    def save_to_file(self, filename: str) -> None:
        """ save people:patterns to a file """
        with open(filename, 'w') as f:
            json.dump({'patterns': self.patterns}, f)
<file_sep>/slack_today_i_did/type_aware/json.py
"""
Support dumping and loading of typing types.
"""
import typing
import functools

from json_tricks import nonp


def encode_type(obj):
    if isinstance(obj, type) or isinstance(obj.__class__, typing.TypingMeta):
        return {'__type_repr__': repr(obj)}
    return obj


class TypeHook(object):
    def __init__(self, types):
        self.type_lookup_map = {repr(type_obj): type_obj for type_obj in types}

    def __call__(self, dct):
        if isinstance(dct, dict) and '__type_repr__' in dct:
            return self.type_lookup_map.get(dct['__type_repr__'], dct)
        return dct


def _append_to_key(kwargs, key, obj):
    objs = []
    if key in kwargs:
        objs = list(kwargs[key])
    objs.append(obj)
    kwargs[key] = objs


def _wrap_dumper(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        _append_to_key(kwargs, 'extra_obj_encoders', encode_type)
        return fn(*args, **kwargs)
    return wrapper


def _wrap_loader(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        known_types = kwargs.pop('known_types', tuple())
        _append_to_key(kwargs, 'extra_obj_pairs_hooks', TypeHook(known_types))
        return fn(*args, **kwargs)
    return wrapper


dump = _wrap_dumper(nonp.dump)
dumps = _wrap_dumper(nonp.dumps)
load = _wrap_loader(nonp.load)
loads = _wrap_loader(nonp.loads)
<file_sep>/main.py
import json
import asyncio
from slack_today_i_did.bot_file import TodayIDidBot
from slack_today_i_did.bot_repl import ReplBot
from slack_today_i_did.our_repo import ElmRepo
import os
import argparse


def setup():
    config_found = True

    try:
        with open('priv.json') as f:
            data = json.load(f)
    except FileNotFoundError:
        print('not using config file..')
        print('some features may be disabled!')
        config_found = False

    repo = None

    os.makedirs('reports', exist_ok=True)

    if config_found:
        os.makedirs('repos', exist_ok=True)
        github_data = data['github']
        repo = ElmRepo(github_data['folder'], github_data['token'], github_data['org'], github_data['repo'])
    else:
        data = {}

    return (data, repo)


def setup_slack(data, repo):
    return TodayIDidBot(
        data.get('token', ''),
        rollbar_token=data.get('rollbar-token', None),
        elm_repo=repo
    )


def setup_cli(data, repo):
    return ReplBot(
        data.get('token', ''),
        rollbar_token=data.get('rollbar-token', None),
        elm_repo=repo
    )


def main():
    parser = argparse.ArgumentParser(description='Start the slack-today-i-did-bot')

    parser.add_argument(
        '--repl',
        '-r',
        action='store_true',
        help='run the repl',
        default=False
    )
    parser.add_argument(
        '--slack',
        '-s',
        action='store_true',
        help='run the slack bot',
        default=False
    )

    args = parser.parse_args()

    if args.repl and args.slack:
        print('Please only start the repl or the slack bot!')
        exit(-1)

    (data, repo) = setup()

    if args.repl:
        print('starting repl..')
        client = setup_cli(data, repo)
    elif args.slack:
        print('starting slack client..')
        client = setup_slack(data, repo)
    else:
        print('starting slack client..')
        client = setup_slack(data, repo)

    loop = asyncio.get_event_loop()
    loop.run_until_complete(client.main_loop())


if __name__ == '__main__':
    main()
<file_sep>/slack_today_i_did/bot_file.py
"""
This file contains the bot itself

To add a new function:
    - add an entry in `known_functions`.
        The key is the command the bot will know, the value is the command to run
    - You must add type annotitions for the bot to match args up correctly

"""

from typing import Dict, Any

from slack_today_i_did.rollbar import Rollbar

from slack_today_i_did.extensions import (
    BasicStatements, KnownNamesExtensions,
    NotifyExtensions, ReportExtensions,
    SessionExtensions, RollbarExtensions,
    ElmExtensions, ExtensionExtensions
)

from slack_today_i_did.generic_bot import GenericSlackBot, ChannelMessage, ChannelMessages
import slack_today_i_did.self_aware as self_aware


class Extensions(
    BasicStatements, KnownNamesExtensions,
    NotifyExtensions, ReportExtensions,
    SessionExtensions, RollbarExtensions,
    ElmExtensions, ExtensionExtensions
):
    pass


class TodayIDidBot(Extensions, GenericSlackBot):

    def __init__(self, *args, **kwargs):
        kwargs = self._setup_from_kwargs_and_remove_fields(**kwargs)

        GenericSlackBot.__init__(self, *args, **kwargs)
        self.reports = {}
        self.name = 'today-i-did'

        self._setup_known_names()
        self._setup_notify()
        self._setup_sessions()
        self._setup_command_history()
        self._setup_enabled_tokens()

    def _setup_from_kwargs_and_remove_fields(self, **kwargs: Dict[str, Any]) -> Dict[str, Any]:
        rollbar_token = kwargs.pop('rollbar_token', None)
        if rollbar_token is None:
            self.rollbar = None
        else:
            self.rollbar = Rollbar(rollbar_token)

        self.repo = kwargs.pop('elm_repo', None)
        self.reports_dir = kwargs.pop('reports_dir', 'reports')
        self.known_names_file = kwargs.pop('known_names_file', 'names.json')
        self.notify_file = kwargs.pop('notify_file', 'notify.json')
        self.session_file = kwargs.pop('session_file', 'sessions.json')
        self.command_history_file = kwargs.pop('command_history_file', 'command_history.json')
        return kwargs

    def _setup_command_history(self) -> None:
        known_functions = {action.__name__: action for action in self.known_functions().values()}
        self.command_history.load_from_file(known_functions, (ChannelMessages,), self.command_history_file)

    @property
    def features_enabled(self):
        return {
            "repo": self.repo is not None,
            "rollbar": self.rollbar is not None
        }

    def _actually_parse_message(self, message):
        GenericSlackBot._actually_parse_message(self, message)

        strings = []
        for attachment in message.get('attachments', []):
            strings.extend(self.attachment_strings(attachment))

        if 'text' in message:
            strings.append(message['text'])

        people_who_want_notification = []

        for string in strings:
            people_who_want_notification.extend(self.notify.who_wants_it(string))

        people_who_want_notification = set(people_who_want_notification)

        for person in people_who_want_notification:
            self.ping_person(message['channel'], person)

        if self.command_history.needs_save:
            self.command_history.save_to_file(self.command_history_file)

    def parse_direct_message(self, message):
        user = message['user']
        text = message['text']

        if self.was_directed_at_me(text):
            return self._actually_parse_message(message)

        name = self.user_name_from_id(user)

        for (channel_name, reports) in self.reports.items():
            for report in reports.values():
                if report.is_for_user(name):
                    report.add_response(name, text)
                    self.send_message(name, 'Thanks!')

        if self.sessions.has_running_session(user):
            self.sessions.add_message(user, text)
            self.sessions.save_to_file(self.session_file)

    def on_tick(self):
        for (channel, reports) in self.reports.items():
            for report in reports.values():
                if report.is_time_to_bother_people():
                    report.bother_people(self)
                elif report.is_time_to_end():
                    self.single_report_responses(channel, report)

    def known_statements(self):
        return {
            'FOR': self.for_statement,
            'AT': self.at_statement,
            'WAIT': self.wait_statement,
            'NOW': self.now_statement,
            'NUM': self.num_statement,
            '!!': self.last_command_statement
        }

    def known_user_functions(self):
        return {
            'bother': self.bother,
            'bother-all-now': self.bother_all_now,
            'report-responses': self.report_responses,
            'responses': self.responses,

            'func-that-return': self.functions_that_return,
            'error-help': self.error_help,
            'help': self.help,
            'list': self.list,

            'reload-funcs': self.reload_functions,
            'reload': self.reload_branch,
            'status': self.status,

            'house-party': self.party,

            'rollbar-item': self.rollbar_item,

            'elm-progress': self.elm_progress,
            'elm-progress-on': self.elm_progress_on,
            'find-017-matches': self.find_elm_017_matches,
            'how-hard-to-port': self.how_hard_to_port,

            'who-do-you-know': self.get_known_names,
            'know-me': self.add_known_name,

            'when-you-hear': self.when_you_hear,
            'forget': self.stop_listening,

            'start-session': self.start_session,
            'end-session': self.end_session,

            'tokens-status': self.tokens_status,
            'disable-token': self.disable_token,
            'enable-token': self.enable_token,

            'known-ext': self.known_extensions,
            'disable-ext': self.disable_extension,
            'enable-ext': self.enable_extension,
            'load-ext': self.load_extension
        }

    def reload_branch(self, channel: str, branch: str = None) -> ChannelMessages:
        """ reload a branch and trigger a restart """

        if branch is None:
            branch = self_aware.git_current_version()
            if branch.startswith('HEAD DETACHED'):
                return []

            on_branch_message = 'On branch'

            if branch.startswith(on_branch_message):
                branch = branch[len(on_branch_message):]

        self_aware.git_checkout(branch)
        self_aware.restart_program()

        return []

    def status(self, channel: str, show_all: str = None) -> ChannelMessages:
        """ provides meta information about the bot """

        current_version = self_aware.git_current_version()

        message = f"I am running on {current_version}\n"

        if show_all is not None:
            message += '-------------\n'
            message += '\n'.join(
                f'{feature} is {"enabled" if is_enabled else "disabled"}'
                for (feature, is_enabled) in self.features_enabled.items()
            )
            message += '\n-------------\n'

            message += f'Python version: {self_aware.python_version()}\n'
            message += f'Ruby version: {self_aware.ruby_version()}\n'

        return ChannelMessage(channel, message)

    def party(self, channel: str) -> ChannelMessages:
        """ TADA """
        return ChannelMessage(channel, ':tada:')
<file_sep>/tests/test_command_history.py
import pytest
from typing import Union
import os
import slack_today_i_did.command_history as command_history

MOCK_CHANNEL = 'dave'
MOCK_FUNCTION_NAME = 'MOCK_FUNCTION'
MOCK_TEST_FILE = '.testdata_command_history'


def MOCK_FUNCTION():
    pass


MOCK_KNOWN_TOKENS = {
    'MOCK_FUNCTION': MOCK_FUNCTION
}
MOCK_UNION = Union[int, str]
MOCK_KNOWN_TYPES = (MOCK_UNION,)


def test_add_pattern():
    commands = command_history.CommandHistory()

    assert commands.last_command(MOCK_CHANNEL) is None

    commands.add_command(MOCK_CHANNEL, MOCK_FUNCTION, [])

    last_command = commands.last_command(MOCK_CHANNEL)

    assert last_command['action'] == MOCK_FUNCTION
    assert last_command['args'] == []
    assert commands.needs_save == True

    assert len(commands.history) == 1
    assert len(commands.history[MOCK_CHANNEL]) == 1


def test_saving_and_loading(tmpdir):
    history_file = str(tmpdir.join(MOCK_TEST_FILE))
    commands = command_history.CommandHistory()
    commands.load_from_file(MOCK_KNOWN_TOKENS, MOCK_KNOWN_TYPES, history_file)
    assert commands.last_command(MOCK_CHANNEL) is None
    assert commands.needs_save == False

    commands.add_command(MOCK_CHANNEL, MOCK_FUNCTION, [MOCK_UNION])
    commands.save_to_file(history_file)

    assert commands.last_command(MOCK_CHANNEL) is not None

    new_commands = command_history.CommandHistory()
    new_commands.load_from_file(MOCK_KNOWN_TOKENS, MOCK_KNOWN_TYPES, history_file)

    assert new_commands.last_command(MOCK_CHANNEL) is not None
    assert commands == new_commands
<file_sep>/slack_today_i_did/bot_repl.py
from slack_today_i_did.bot_file import TodayIDidBot

import prompt_toolkit.layout.lexers
import prompt_toolkit.auto_suggest
from prompt_toolkit.interface import CommandLineInterface
from prompt_toolkit.shortcuts import (
    create_prompt_application,
    create_asyncio_eventloop
)
from prompt_toolkit.history import FileHistory
from prompt_toolkit.contrib.completers import WordCompleter


class ReplBot(TodayIDidBot):
    def __init__(self, *args, **kwargs):
        TodayIDidBot.__init__(self, *args, **kwargs)
        self._setup_cli_history()

    def _setup_cli_history(self):
        self.cli_history = FileHistory('.cli_history')

    async def __aenter__(self):
        self.eventloop = create_asyncio_eventloop()
        self.completer = WordCompleter(self.known_tokens())

        config = {
            'complete_while_typing': True,
            'enable_history_search': True,
            'history': self.cli_history,
            'auto_suggest': Suggestions(),
            'completer': self.completer,
            'lexer': prompt_toolkit.layout.lexers.SimpleLexer()
        }

        self.cli = CommandLineInterface(
            application=create_prompt_application(f'{self.name} >>> ', **config),
            eventloop=self.eventloop
        )
        return self

    async def __aexit__(self, *args, **kwargs):
        return self

    async def get_message(self):
        incoming = await self.cli.run_async()
        incoming = f'<@{self.name}> {incoming.text}'

        data = [{
            "text": incoming,
            "type": "message",
            "channel": "CLI",
            "user": "CLI"
        }]

        return data

    def ping(self):
        return self.send_to_websocket({"type": "ping"})

    def send_to_websocket(self, data):
        if 'type' not in data:
            return

        if data['type'] == 'message':
            print(data['text'])

    def set_known_users(self):
        return

    def user_name_from_id(self, my_id):
        for (name, id) in self.known_users.items():
            if id == my_id:
                return name
        return None

    def open_chat(self, name: str) -> str:
        return name

    def send_message(self, name: str, message: str) -> None:
        json = {"type": "message", "channel": id, "text": message}
        self.send_to_websocket(json)

    def send_channel_message(self, channel: str, message: str) -> None:
        json = {"type": "message", "channel": channel, "text": message}
        self.send_to_websocket(json)

    def connected_user(self, username: str) -> str:
        return username

    def attachment_strings(self, attachment):
        strings = []

        for (k, v) in attachment.items():
            if isinstance(v, str):
                strings.append(v)

        for field in attachment.get('fields', []):
            strings.append(field['title'])
            strings.append(field['value'])

        return strings


class Suggestions(prompt_toolkit.auto_suggest.AutoSuggestFromHistory):
    def get_suggestion(self, cli, buffer, document):
        return prompt_toolkit.auto_suggest.AutoSuggestFromHistory.get_suggestion(
            self,
            cli,
            buffer,
            document
        )
<file_sep>/tests/test_notify.py
import pytest
import os
import slack_today_i_did.notify as notify

MOCK_PERSON = 'dave'
MOCK_VALID_PATTERN = '[a]'
MOCK_INVALID_PATTERN = '['
MOCK_TEXT_WITH_PATTERN = 'a'
MOCK_TEXT_WITHOUT_PATTERN = "b"
MOCK_TEST_FILE = '.testdata_notify'


def test_add_pattern():
    notification = notify.Notification()

    notification.add_pattern(MOCK_PERSON, MOCK_VALID_PATTERN)

    assert MOCK_PERSON in notification.patterns
    assert notification.patterns[MOCK_PERSON] == [MOCK_VALID_PATTERN]


def test_forget_pattern():
    notification = notify.Notification()

    notification.add_pattern(MOCK_PERSON, MOCK_VALID_PATTERN)
    notification.forget_pattern(MOCK_PERSON, MOCK_VALID_PATTERN)
    assert MOCK_PERSON in notification.patterns
    assert len(notification.patterns[MOCK_PERSON]) == 0


    notification.add_pattern(MOCK_PERSON, MOCK_VALID_PATTERN)
    notification.forget_pattern(MOCK_PERSON, MOCK_INVALID_PATTERN)
    assert MOCK_PERSON in notification.patterns
    assert len(notification.patterns[MOCK_PERSON]) == 1


def test_who_wants_it():
    notification = notify.Notification()

    notification.add_pattern(MOCK_PERSON, MOCK_VALID_PATTERN)

    who_wants_it = notification.who_wants_it(MOCK_TEXT_WITHOUT_PATTERN)
    assert len(who_wants_it) == 0

    who_wants_it = notification.who_wants_it(MOCK_TEXT_WITH_PATTERN)
    assert len(who_wants_it) == 1
    assert who_wants_it[0] == MOCK_PERSON


def test_saving_and_loading(tmpdir):


    notification = notify.Notification()
    notification.load_from_file(MOCK_TEST_FILE)
    assert len(notification.patterns) == 0

    notification.add_pattern(MOCK_PERSON, MOCK_VALID_PATTERN)
    notification.save_to_file(MOCK_TEST_FILE)

    assert len(notification.patterns) == 1

    new_notification = notify.Notification()
    new_notification.load_from_file(MOCK_TEST_FILE)
    assert len(new_notification.patterns) == 1
    assert notification.patterns == new_notification.patterns

    os.remove(MOCK_TEST_FILE)
<file_sep>/README.md
# slack-today-i-did

A chat-ops bot designed around the principle of ensuring that types for functions are correct.


## Features

- slack and repl support
- live-reload of functions
- live-reload of the entire project
- type safe interactions
- useful error hints
- enable/disable features at runtime
- command history

## Setup

- You must have Python 3.6 installed
- Use virtualenv
- To install deps run `pip install -r requirements.txt`

- In order to use this bot, you must create a file called `priv.json` which looks like this:
```
{
    "token": "<SLACK_BOT_TOKEN>",
    "rollbar-token" : "<ROLLBAR_READ_TOKEN>",
    "github" : {
        "token" : "<GITHUB_OAUTH_TOKEN",
        "repo" : "NoRedInk",
        "org" : "NoRedInk",
        "folder" : "repos"
    }
}

```

## Running

- `python main.py` starts up the slack bot by default
- `python main.py --repl` starts up the local repl


<file_sep>/tests/test_parser.py
import pytest
import slack_today_i_did.parser as parser

MOCK_TOKENS = {
    'hello': lambda x:x,
    'NOW': lambda x:x
}

TEXT_WITHOUT_TOKENS = 'dave fish'
TEXT_WITH_TOKEN = "hello dave"
TEXT_WITH_TOKEN_MISPELLED = "hfllo"
TEXT_WITH_TOKENS = "hello dave NOW"


def test_tokens_with_index_on_no_tokens():
    tokens = parser.tokens_with_index(MOCK_TOKENS, TEXT_WITHOUT_TOKENS)
    assert len(tokens) == 0


def test_tokens_with_index_on_single_token():
    tokens = parser.tokens_with_index(MOCK_TOKENS, TEXT_WITH_TOKEN)
    assert len(tokens) == 1

    first_token = tokens[0]

    assert first_token[0] == 0
    assert first_token[1] == 'hello'


def test_tokens_with_index_on_tokens():
    tokens = parser.tokens_with_index(MOCK_TOKENS, TEXT_WITH_TOKENS)
    assert len(tokens) == 2

    first_token = tokens[0]
    second_token = tokens[1]

    assert first_token[0] == 0
    assert first_token[1] == 'hello'
    assert second_token[0] == 11
    assert second_token[1] == 'NOW'


def test_tokenize_on_no_tokens():
    tokens = parser.tokenize(TEXT_WITHOUT_TOKENS, MOCK_TOKENS)
    assert len(tokens) == 0


def test_tokenize_on_one_token():
    tokens = parser.tokenize(TEXT_WITH_TOKEN, MOCK_TOKENS)
    assert len(tokens) == 1
    first_token = tokens[0]

    assert first_token[0] == 0
    assert first_token[1] == 'hello'
    assert first_token[2] == 'dave'


def test_tokenize_on_tokens():
    tokens = parser.tokenize(TEXT_WITH_TOKENS, MOCK_TOKENS)
    assert len(tokens) == 2
    first_token = tokens[0]

    assert first_token[0] == 0
    assert first_token[1] == 'hello'
    assert first_token[2] == 'dave '

    second_token = tokens[1]

    assert second_token[0] == 11
    assert second_token[1] == 'NOW'
    assert second_token[2] == ''


def test_parse_no_tokens():
    known_funcs = {'error-help': lambda x:x}
    stuff = parser.parse([], known_funcs)

    assert stuff.func_call.func_name == 'error-help'
    assert stuff.func_call.args == [parser.Constant('NO_TOKENS', str)]


def test_parse_first_func_not_in_known_funcs():
    # `eval` trusts tokenize to have extracted
    # only known function names into tokens
    with pytest.raises(KeyError):
        stuff = parser.parse([(0, 'help', ''), (0, 'hello', 'dave')], MOCK_TOKENS)


def test_eval_help_with_arg():
    funcs_with_help = dict(MOCK_TOKENS, help=lambda x:x)
    tokens = parser.tokenize('help ' + TEXT_WITH_TOKENS, funcs_with_help)
    stuff = parser.parse(tokens, funcs_with_help)

    assert stuff.func_call.func_name == 'help'
    assert stuff.func_call.args[0] == parser.FuncCall('hello', [parser.Constant('dave ', str)], None)
    assert stuff.func_call.args[1] == parser.FuncCall('NOW', [], None)


def test_eval_help_with_badly_spelled_function():
    funcs_with_help = dict(MOCK_TOKENS, help=lambda x:x)
    tokens = parser.tokenize('help ' + TEXT_WITH_TOKEN_MISPELLED, funcs_with_help)
    stuff = parser.parse(tokens, funcs_with_help)

    assert stuff.func_call.func_name == 'help'
    assert stuff.func_call.args == [parser.Constant('hfllo', str)]

    result = stuff.evaluate(stuff.func_call)
    assert result.result == 'hfllo'
    assert result.errors == []


def test_eval_first_func_with_one_arg():
    known_funcs = {'hello': lambda x:x}
    stuff = parser.parse([(0, 'hello', 'world')], known_funcs)

    assert stuff.func_call.func_name == 'hello'
    assert stuff.func_call.args == [parser.Constant('world', str)]

    result = stuff.evaluate(stuff.func_call)
    assert result.result == 'world'
    assert result.errors == []


def test_eval_second_func_with_one_arg():
    known_funcs = {'cocoa': lambda x:x + 'cocoa', 'double': lambda x:x * 2}
    known_funcs['double'].__annotations__['return'] = str

    stuff = parser.parse([(0, 'cocoa', ''), (0, 'double', 'cream')], known_funcs)

    assert stuff.func_call.func_name == 'cocoa'
    assert stuff.func_call.args == [parser.FuncCall('double', [parser.Constant('cream', str)], str)]

    result = stuff.evaluate(stuff.func_call)
    assert result.result == 'creamcreamcocoa'
    assert result.errors == []


def test_eval_second_func_with_error():
    known_funcs = {'cocoa': lambda x:x + 'cocoa', 'koan': lambda x: 1 / 0}

    stuff = parser.parse([(0, 'cocoa', ''), (0, 'koan', 'x')], known_funcs)

    assert stuff.func_call.func_name == 'cocoa'
    assert stuff.func_call.args == [parser.FuncCall('koan', [parser.Constant('x', str)], None)]

    result = stuff.evaluate(stuff.func_call)
    assert result.result == None
    assert 'koan threw division by zero' in result.errors[0]


def test_eval_type_mismtach():
    def koan(x: int) -> None:
        return 1 / 0
    known_funcs = {'koan': koan}

    stuff = parser.parse([(0, 'koan', 'x')], known_funcs)

    assert stuff.func_call.func_name == 'koan'
    assert stuff.func_call.args == [parser.Constant('x', str)]

    result = stuff.evaluate(stuff.func_call)
    assert 'You tried to give me' in result.errors[0]
    assert 'but I wanted a `<class \'int\'>`' in result.errors[0]


def test_eval_arg_mismtach():
    def more(a: str, b: str) -> str:
        return a + b + ' more'
    known_funcs = {'more': more}

    stuff = parser.parse([(0, 'more', 'x')], known_funcs)

    assert stuff.func_call.func_name == 'more'
    assert stuff.func_call.args == [parser.Constant('x', str)]

    result = stuff.evaluate(stuff.func_call)
    assert 'I wanted things to look like' in result.errors[0]
    assert 'Need some more' in result.errors[0]
<file_sep>/tests/test_sessions.py
import pytest
import os
import slack_today_i_did.reports as reports

MOCK_PERSON = 'dave'
MOCK_CHANNEL = '#durp'
MOCK_TEXT = 'abcdflk'
MOCK_TEST_FILE = '.testdata_sessions'


def test_start_session():
    sessions = reports.Sessions()

    sessions.start_session(MOCK_PERSON, MOCK_CHANNEL)

    assert MOCK_PERSON in sessions.sessions
    assert sessions.has_running_session(MOCK_PERSON)
    assert sessions.get_entry(MOCK_PERSON).get('channel') == MOCK_CHANNEL
    assert len(sessions.get_entry(MOCK_PERSON).get('messages')) == 0

    sessions.add_message(MOCK_PERSON, MOCK_TEXT)
    assert sessions.get_entry(MOCK_PERSON).get('channel') == MOCK_CHANNEL
    assert len(sessions.get_entry(MOCK_PERSON).get('messages')) == 1
    assert sessions.get_entry(MOCK_PERSON).get('messages') == [MOCK_TEXT]

def test_end_session():
    sessions = reports.Sessions()

    sessions.start_session(MOCK_PERSON, MOCK_CHANNEL)
    sessions.add_message(MOCK_PERSON, MOCK_TEXT)
    sessions.end_session(MOCK_PERSON)

    assert sessions.get_entry(MOCK_PERSON).get('channel') == MOCK_CHANNEL
    assert len(sessions.get_entry(MOCK_PERSON).get('messages')) == 1
    assert sessions.get_entry(MOCK_PERSON).get('messages') == [MOCK_TEXT]
    assert not sessions.has_running_session(MOCK_PERSON)


def test_session_saving_and_loading(tmpdir):
    sessions = reports.Sessions()
    sessions.load_from_file(MOCK_TEST_FILE)
    assert len(sessions.sessions) == 0

    sessions.start_session(MOCK_PERSON, MOCK_CHANNEL)
    sessions.add_message(MOCK_PERSON, MOCK_TEXT)
    sessions.save_to_file(MOCK_TEST_FILE)

    new_sessions = reports.Sessions()
    new_sessions.load_from_file(MOCK_TEST_FILE)
    assert len(new_sessions.sessions) == 1
    assert sessions.sessions == new_sessions.sessions

    os.remove(MOCK_TEST_FILE)


def test_session_retiring():
    sessions = reports.Sessions()
    sessions.start_session(MOCK_PERSON, MOCK_CHANNEL)
    sessions.add_message(MOCK_PERSON, MOCK_TEXT)
    sessions.retire_session(MOCK_PERSON, MOCK_TEST_FILE)
    assert len(sessions.sessions) == 0

    with open(MOCK_TEST_FILE) as f:
        assert MOCK_TEXT in f.read()

    os.remove(MOCK_TEST_FILE)
<file_sep>/slack_today_i_did/text_tools.py
def levenshtein(current_word: str, next_word: str) -> int:
    ''' Returns how similar a word is to the next word as a number

        no changes return 0
        >>> levenshtein('a', 'a')
        0

        Add a character adds 1 for each character added
        >>> levenshtein('a', 'ab')
        1
        >>> levenshtein('a', 'abc')
        2

        Removing a character adds 1 for each character
        >>> levenshtein('a', '')
        1
        >>> levenshtein('abc', '')
        3

        Replacing a character adds 1 for each character
        >>> levenshtein('abcdf', 'zxcvb')
        4
    '''

    if current_word == '':
        return len(next_word)

    if len(current_word) < len(next_word):
        current_word, next_word = next_word, current_word

    previous_row = list(range(len(next_word) + 1))

    for i, current_character in enumerate(current_word):
        current_row = [i + 1]

        for j, next_character in enumerate(next_word):
            insertions = previous_row[j + 1] + 1
            deletions = current_row[j] + 1
            substitutions = previous_row[j] + (current_character != next_character)
            current_row.append(min(insertions, deletions, substitutions))

        previous_row = current_row[:]
    return previous_row[-1]


def token_based_levenshtein(current_words: str, next_words: str) -> int:
    ''' Returns how similar a word is to the next word as a number

        no changes return 0
        >>> token_based_levenshtein('a', 'a')
        0

        Add a character adds 1 for each character added
        >>> token_based_levenshtein('a', 'ab')
        1
        >>> token_based_levenshtein('a', 'abc')
        2

        Removing a character adds 1 for each character
        >>> token_based_levenshtein('a', '')
        1
        >>> token_based_levenshtein('abc', '')
        3

        Replacing a character adds 1 for each character
        >>> token_based_levenshtein('abcdf', 'zxcvb')
        4

        Adding a word adds 1 char for each word
        >>> token_based_levenshtein('a', 'a-b')
        1

        >>> token_based_levenshtein('a', 'a-be-c')
        3

        >>> token_based_levenshtein('a-b', 'a-be-c')
        2

        >>> token_based_levenshtein('c-b', 'a-be-c')
        3
    '''

    words = current_words.split('-')
    next_words = next_words.split('-')

    if len(words) == 0:
        return len(next_words)

    if len(words) < len(next_words):
        words, next_words = next_words, words

    previous_row = list(range(len(next_words) + 1))

    for i, current_word in enumerate(words):
        current_row = [i + 1]

        for j, coming_word in enumerate(next_words):
            insertions = previous_row[j + 1] + 1
            deletions = current_row[j] + 1
            substitutions = previous_row[j] + (current_word != coming_word)
            word_diff = levenshtein(current_word, coming_word)

            if word_diff > 1:
                current_row.append(word_diff)
            else:
                current_row.append(min(insertions, deletions, substitutions))

        previous_row = current_row[:]
    return previous_row[-1]
<file_sep>/tests/conftest.py
import contextlib

import pytest


@pytest.fixture
def message_context(mocker):
    '''Sets up the specified `bot`'s state as if it just received
    a message from `sender` and about to parse it.

        with message_context(bot, MOCK_PERSON):
             do_stuff_with_the_bot(bot)
    '''
    @contextlib.contextmanager
    def wrapper(bot, sender):
        bot_qualname = f'{bot.__class__.__module__}.{bot.__class__.__qualname__}'
        mocker.patch(
            f'{bot_qualname}._last_sender',
            new_callable=mocker.PropertyMock, return_value=sender)
        mocker.patch.object(
            bot, 'user_name_from_id',
            return_value=sender)
        mocker.patch.object(
            bot, 'connected_user',
            return_value=bot.__class__.__name__)
        mocker.patch.object(
            bot, 'was_directed_at_me',
            return_value=True)

        yield

        mocker.stopall()

    return wrapper
<file_sep>/slack_today_i_did/extensions.py
import datetime
from typing import List
import json
import re
from collections import defaultdict
import importlib
import types

from slack_today_i_did.reports import Report
from slack_today_i_did.generic_bot import BotExtension, ChannelMessage, ChannelMessages
from slack_today_i_did.reports import Sessions
from slack_today_i_did.known_names import KnownNames
from slack_today_i_did.notify import Notification
import slack_today_i_did.parser as parser
import slack_today_i_did.text_tools as text_tools


class BasicStatements(BotExtension):
    def for_statement(self, text: str) -> List[str]:
        """ enter usernames seperated by commas """
        return [blob.strip() for blob in text.split(',')]

    def at_statement(self, text: str) -> datetime.datetime:
        """ enter a time in the format HH:MM """
        return datetime.datetime.strptime(text.strip(), '%H:%M')

    def wait_statement(self, text: str) -> datetime.datetime:
        """ enter how much time to wait in the format HH:MM """
        return datetime.datetime.strptime(text.strip(), '%H:%M')

    def now_statement(self) -> datetime.datetime:
        """ return the current time """
        return datetime.datetime.utcnow()

    def num_statement(self, text: str) -> int:
        """ return a number value """
        try:
            return int(text)
        except:
            return 0


class ExtensionExtensions(BotExtension):
    def _setup_enabled_tokens(self):
        """ Store disabled tokens in a dict of token: user
            Store disabled extensions in just a list of class names
        """
        self._disabled_tokens = {}
        self._disabled_extensions = []

    def _disabled_message(self, who: str, channel: str) -> ChannelMessages:
        # TODO: this function is currently evaluated in the wrong way by the evaluator
        # so we send the message by hand
        self.send_channel_message(channel, f'This function has been disabled by {who}.')
        return []

    def _flatten_bases(self, cls):
        """ Get all the extensions applied to a bot instance
        """
        bases = cls.__bases__
        known_bases = []

        for base in bases:
            if base.__name__ == 'object':
                continue

            current_bases = self._flatten_bases(base)

            known_bases.append(base)

            if BotExtension in current_bases:
                known_bases.extend(current_bases)

        return known_bases

    def known_functions(self):
        known_functions = BotExtension.known_functions(self)

        try:
            if len(self._disabled_tokens) == 0:
                return known_functions
        except:
            return known_functions

        wrapped_functions = {
            k: v for (k, v) in known_functions.items()
            if k not in self._disabled_tokens
        }

        wrapped_functions.update({
            token: lambda *args, **kwargs: self._disabled_message(who, *args, **kwargs)
            for (token, who) in self._disabled_tokens.items()
        })

        return wrapped_functions

    def known_extensions(self, channel: str) -> ChannelMessages:
        """ List all extensions used by the current bot """
        known_bases = list(set(self._flatten_bases(self.__class__)))

        message = 'Currently known extensions:\n'
        message += '\n'.join(base.__name__ for base in known_bases)

        return ChannelMessage(channel, message)

    def tokens_status(self, channel: str) -> ChannelMessages:
        """ Display all the known tokens and if they are enabled """
        known_tokens = [
            (token, token not in self._disabled_tokens) for token in self.known_tokens()
        ]

        message = '\n'.join(f'{token}: {is_enabled}' for (token, is_enabled) in known_tokens)

        return ChannelMessage(channel, message)

    def _manage_extension(self, extension_name: str, is_to_enable: bool, disabler: str) -> None:
        """ Disable or enable an extension, setting who disabled it """
        known_bases = list(set(self._flatten_bases(self.__class__)))
        flipped_tokens = {
            func.__name__: func_alias for (func_alias, func) in self.known_functions().items()
        }

        extensions = [base for base in known_bases if base.__name__ == extension_name]

        for extension in extensions:
            for func in extension.__dict__:
                if func not in flipped_tokens:
                    continue

                if is_to_enable:
                    self._disabled_tokens.pop(flipped_tokens[func], None)
                else:
                    self._disabled_tokens[flipped_tokens[func]] = disabler

            if is_to_enable:
                self._disabled_extensions.remove(extension.__name__)
            else:
                self._disabled_extensions.append(extension.__name__)

    def enable_extension(self, channel: str, extension_name: str) -> ChannelMessages:
        """ enable an extension and all it's exposed tokens by name """
        self._manage_extension(extension_name, is_to_enable=True, disabler=self._last_sender)
        return []

    def disable_extension(self, channel: str, extension_name: str) -> ChannelMessages:
        """ disable an extension and all it's exposed tokens by name """
        self._manage_extension(extension_name, is_to_enable=False, disabler=self._last_sender)
        return []

    def load_extension(self, channel: str, extension_name: str = None) -> ChannelMessages:
        """ Load extensions. By default, load everything.
            Otherwise, load a particular extension
        """
        known_bases = list(set(self._flatten_bases(self.__class__)))
        known_bases_as_str = [base.__name__ for base in known_bases]
        func_names = [func.__name__ for func in self.known_functions().values()]
        meta_funcs = [
            func.__name__ for func in self.known_functions().values() if parser.is_metafunc(func)
        ]

        # make sure to pick up new changes
        importlib.invalidate_caches()

        # import and reload ourselves
        extensions = importlib.import_module(__name__)
        importlib.reload(extensions)

        extension_names = dir(extensions)
        if extension_name is not None and extension_name.strip() != "":
            if extension_name not in extension_names:
                suggestions = [
                    (text_tools.levenshtein(extension_name, name), name) for name in extension_names
                ]

                message = 'No such extension! Maybe you meant one of these:\n'
                message += ' | '.join(name for (_, name) in sorted(suggestions)[:5])
                return ChannelMessage(channel, message)
            else:
                extension_names = [extension_name]

        for extension in extension_names:
            # skip if the extension is not a superclass
            if extension not in known_bases_as_str:
                continue

            extension_class = getattr(extensions, extension)

            for (func_name, func) in extension_class.__dict__.items():
                # we only care about reloading things in our tokens
                if func_name not in func_names:
                    continue

                # ensure that meta_funcs remain so
                if func_name in meta_funcs:
                    func = parser.metafunc(func)

                setattr(self, func_name, types.MethodType(func, self))

        return []

    @parser.metafunc
    def enable_token(self, channel: str, tokens) -> ChannelMessages:
        """ enable tokens """
        for token in tokens:
            if token.func_name in self._disabled_tokens:
                self._disabled_tokens.pop(token.func_name, None)
        return []

    @parser.metafunc
    def disable_token(self, channel: str, tokens) -> ChannelMessages:
        """ disable tokens """
        for token in tokens:
            func_name = token.func_name
            self._disabled_tokens[func_name] = self._last_sender

        return []


class KnownNamesExtensions(BotExtension):
    def _setup_known_names(self) -> None:
        self.known_names = KnownNames()
        self.known_names.load_from_file(self.known_names_file)

    def get_known_names(self, channel: str) -> ChannelMessages:
        """ Grabs the known names to this bot! """
        message = []

        for (person, names) in self.known_names.people.items():
            message.append(f'<@{person}> goes by the names {" | ".join(names)}')

        if len(message) == '':
            message = "I don't know nuffin or no one"

        return ChannelMessage(channel, '\n'.join(message))

    def add_known_name(self, channel: str, name: str) -> ChannelMessages:
        """ adds a known name to the collection for the current_user """
        person = self._last_sender

        self.known_names.add_name(person, name)
        self.known_names.save_to_file(self.known_names_file)

        return []


class NotifyExtensions(BotExtension):
    def _setup_notify(self) -> None:
        self.notify = Notification()
        self.notify.load_from_file(self.notify_file)

    def when_you_hear(self, channel: str, pattern: str) -> ChannelMessages:
        """ notify the user when you see a pattern """
        person = self._last_sender

        try:
            re.compile(pattern)
        except Exception as e:
            return ChannelMessage(channel, f'Invalid regex due to {e.msg}')

        self.notify.add_pattern(person, pattern)
        self.notify.save_to_file(self.notify_file)
        return ChannelMessage(
            channel,
            f'Thanks! You be notified when I hear that pattern. Use `forget` to stop me notifying you!'
        )

    def stop_listening(self, channel: str, pattern: str) -> ChannelMessages:
        """ stop notify the user when you see a pattern """
        person = self._last_sender

        self.notify.forget_pattern(person, pattern)
        self.notify.save_to_file(self.notify_file)

        return []

    def ping_person(self, channel: str, person: str) -> ChannelMessages:
        """ notify a person about a message. Ignore direct messages """
        if self.is_direct_message(channel):
            return

        return ChannelMessage(channel, f"<@{person}> ^")


class ReportExtensions(BotExtension):
    def responses(self, channel: str) -> ChannelMessages:
        """ list the last report responses for the current channel """

        if channel not in self.reports:
            return ChannelMessage(
                channel,
                f'No reports found for channel {channel}'
            )

        message = ""
        for report in self.reports[channel].values():
            message += f'for the report: {report.name}'
            message += '\n\n'.join(
                f'User {user} responded with:\n{response}' for (user, response) in report.responses.items()  # noqa: E501
            )

        return ChannelMessage(channel, message)

    def report_responses(self, channel: str, name: str) -> ChannelMessages:
        """ list the last report responses for the current channel """

        name = name.strip()

        if channel not in self.reports or name not in self.reports[channel]:
            return ChannelMessage(
                channel,
                f'No reports found for channel {channel}'
            )

        return self.single_report_responses(channel, self.reports[channel][name])

    def single_report_responses(self, channel: str, report) -> ChannelMessages:
        """ Send info on a single response """
        message = '\n\n'.join(
            f'User {user} responded with:\n{response}' for (user, response) in report.responses.items()  # noqa: E501
        )

        if message == '':
            message = f'Nobody replied to the report {report.name}'
        else:
            message = f'For the report: {report.name}\n' + message
        return ChannelMessage(channel, message)

    def bother(self, channel: str, name: str, users: List[str], at: datetime.datetime, wait: datetime.datetime) -> ChannelMessages:  # noqa: E501
        """ add a report for the given users at a given time,
            reporting back in the channel requested
        """
        time_to_run = (at.hour, at.minute)
        wait_for = (wait.hour, wait.minute)
        self.add_report(Report(channel, name, time_to_run, users, wait_for, reports_dir=self.reports_dir))
        return []

    def bother_all_now(self, channel: str) -> ChannelMessages:
        """ run all the reports for a channel
        """
        messages = []
        for report in self.reports.get(channel, {}).values():
            bothers = [ChannelMessage(*bother) for bother in report.bother_people()]
            messages.extend(bothers)
        return messages

    def add_report(self, report):
        if report.channel not in self.reports:
            self.reports[report.channel] = {}
        self.reports[report.channel][report.name] = report


class SessionExtensions(BotExtension):
    def _setup_sessions(self) -> None:
        self.sessions = Sessions()
        self.sessions.load_from_file(self.session_file)

    def start_session(self, channel: str) -> ChannelMessages:
        """ starts a session for a user """
        person = self._last_sender
        self.sessions.start_session(person, channel)
        self.sessions.save_to_file(self.session_file)

        message = """
Started a session for you. Send a DMs to me with what you're working on throughout the day.
Tell me `end-session` to finish the session and post it here!
"""
        return ChannelMessage(channel, message.strip())

    def end_session(self, channel: str) -> ChannelMessages:
        """ ends a session for a user """

        person = self._last_sender
        if not self.sessions.has_running_session(person):
            return ChannelMessage(channel, 'No session running.')

        self.sessions.end_session(person)
        self.sessions.save_to_file(self.session_file)

        entry = self.sessions.get_entry(person)

        message = f'Ended a session for the user <@{person}>. They said the following:\n'
        message += '\n'.join(entry['messages'])
        return ChannelMessage(entry['channel'], message)


class RollbarExtensions(BotExtension):
    def rollbar_item(self, channel: str, field: str, counter: int) -> ChannelMessages:
        """ takes a counter, gets the rollbar info for that counter """

        rollbar_info = self.rollbar.get_item_by_counter(counter)

        if field == '' or field == 'all':
            pretty = json.dumps(rollbar_info, indent=4)
        else:
            pretty = rollbar_info.get(
                field,
                f'Could not find the field {field}'
            )

        return ChannelMessage(channel, f'{pretty}')


class ElmExtensions(BotExtension):
    def elm_progress(self, channel: str, version: str) -> ChannelMessages:
        """ give a version of elm to get me to tell you how many number files are on master """

        version = version.strip()
        self.repo.get_ready()
        message = ""

        if version == '0.17':
            message += f"There are {self.repo.number_of_016_files}"
        elif version == '0.16':
            message += f"There are {self.repo.number_of_017_files}"
        else:
            num_016 = self.repo.number_of_016_files
            num_017 = self.repo.number_of_017_files
            message += f"There are {num_016} 0.16 files."
            message += f"\nThere are {num_017} 0.17 files."
            message += f"\nThat puts us at a total of {num_017 + num_016} Elm files."

        return ChannelMessage(channel, message)

    def elm_progress_on(self, channel: str, branch_name: str) -> ChannelMessages:
        """ give a version of elm to get me to tell you how many number files are on master """

        self.repo.get_ready(branch_name)
        message = ""

        num_016 = self.repo.number_of_016_files
        num_017 = self.repo.number_of_017_files
        message += f"There are {num_016} 0.16 files."
        message += f"\nThere are {num_017} 0.17 files."
        message += f"\nThat puts us at a total of {num_017 + num_016} Elm files."  # noqa: E501

        return ChannelMessage(channel, message)

    def find_elm_017_matches(self, channel: str, filename_pattern: str) -> ChannelMessages:  # noqa: E501
        """ give a filename of elm to get me to tell you how it looks on master """  # noqa: E501

        self.repo.get_ready()
        message = "We have found the following filenames:\n"

        filenames = self.repo.get_files_for_017(filename_pattern)
        message += " | ".join(filenames)

        return ChannelMessage(channel, message)

    def how_hard_to_port(self, channel: str, filename_pattern: str) -> ChannelMessages:
        """ give a filename of elm to get me to tell you how hard it is to port
            Things are hard if: contains ports, signals, native or html.
            Ports and signals are hardest, then native, then html.
        """

        self.repo.get_ready()
        message = "We have found the following filenames:\n"

        with self.repo.cached_lookups():
            files = self.repo.get_017_porting_breakdown(filename_pattern)

        message += f'Here\'s the breakdown for the:'

        total_breakdowns = defaultdict(int)

        for (filename, breakdown) in files.items():
            total_hardness = sum(breakdown.values())
            message += f'\nfile {filename}: total hardness {total_hardness}\n'
            message += ' | '.join(
                f'{name} : {value}' for (name, value) in breakdown.items()
            )

            for (name, value) in breakdown.items():
                total_breakdowns[name] += value

        message += '\n---------------\n'
        message += 'For a total of:\n'
        message += ' | '.join(
            f'{name} : {value}' for (name, value) in total_breakdowns.items()
        )

        return ChannelMessage(channel, message)
<file_sep>/slack_today_i_did/parser.py
from typing import Any, TypeVar, Callable, Dict, List, Tuple, NamedTuple, Union
import copy
import functools

# tokenizer types
Token = Tuple[int, str]
TokenAndRest = Tuple[int, str, str]

# parser types
FuncArg = Union['Constant', 'FuncCall']
Constant = NamedTuple(
    'Constant',
    [('value', Any), ('return_type', type)])
FuncCall = NamedTuple(
    'FuncCall',
    [('func_name', str), ('args', List[FuncArg]), ('return_type', type)])
FuncCallBinding = NamedTuple(
    'FuncCallBinding',
    [('func_call', FuncCall), ('evaluate', Callable[[FuncCall, List[FuncArg]], 'FuncResult'])])
FuncResult = NamedTuple(
    'FuncResult',
    [('result', Any), ('return_type', type), ('action', Callable), ('args', List[Any]), ('errors', List[str])])
ArgsResult = NamedTuple(
    'ArgsResult',
    [('result', List[Any]), ('return_types', List[type]), ('errors', List[str])])
FunctionMap = Dict[str, Callable]


def metafunc(fn):
    fn.is_metafunc = True
    return fn


def is_metafunc(fn):
    return getattr(fn, 'is_metafunc', False)


def fill_in_the_gaps(message: str, tokens: List[Token]) -> List[TokenAndRest]:
    """
        take things that look like [(12, FOR)] turn into [(12, FOR, noah)]
    """

    if len(tokens) < 1:
        return []

    if len(tokens) == 1:
        start_index = tokens[0][0]
        token = tokens[0][1]
        bits_after_token = message[start_index + len(token) + 1:]

        return [(start_index, token, bits_after_token)]

    builds = []

    for (i, (start_index, token)) in enumerate(tokens):
        if i == len(tokens) - 1:
            bits_after_token = message[start_index + len(token) + 1:]
            builds.append((start_index, token, bits_after_token))
            continue

        end_index = tokens[i + 1][0]
        bits_after_token = message[start_index + len(token) + 1: end_index]
        builds.append((start_index, token, bits_after_token))

    return builds


def tokens_with_index(known_tokens: List[str], message: str) -> List[Token]:
    """ get the tokens out of a message, in order, along with
        the index it was found at
    """

    build = []

    start_index = 0
    end_index = 0

    for word in message.split(' '):
        if word in known_tokens:
            token = word
            start_index = end_index + message[end_index:].index(token)
            end_index = start_index + len(token)
            build.append((start_index, token))

    return sorted(build, key=lambda x: x[0])


def tokenize(text: str, known_tokens: List[str]) -> List[TokenAndRest]:
    """ Take text and known tokens
    """

    text = text.strip()
    tokens = fill_in_the_gaps(text, tokens_with_index(known_tokens, text))

    return tokens


def parse(tokens: List[TokenAndRest], known_functions: FunctionMap) -> FuncCallBinding:
    args = []

    # when we can't find anything
    if len(tokens) == 0:
        first_function_name = 'error-help'
        args.append(Constant('NO_TOKENS', str))
    # when we have stuff to work with!
    else:
        first_function_name = tokens[0][1]
        first_arg = tokens[0][2].strip()

        if len(first_arg) > 0:
            args.append(Constant(first_arg, str))

        if len(tokens) > 1:
            for (start_index, function_name, arg_to_function) in tokens[1:]:
                func = known_functions[function_name]
                actual_arg = [Constant(arg_to_function, str)] if len(arg_to_function) > 0 else []
                return_type = func.__annotations__.get('return', None)
                args.append(FuncCall(function_name, actual_arg, return_type))

    action = known_functions[first_function_name]
    return_type = action.__annotations__.get('return', None)
    evaluator = functools.partial(evaluate_func_call, known_functions)

    return FuncCallBinding(
        FuncCall(first_function_name, args, return_type),
        evaluator)


def evaluate_func_call(
        known_functions: FunctionMap,
        func_call: FuncCall,
        default_args: List[FuncArg] = []) -> FuncResult:
    """ Evaluate `func_call` in the context of `known_functions`
        after prepending `default_args` to `func_call`'s arguments
    """
    action = known_functions[func_call.func_name]

    if is_metafunc(action):
        args = default_args + [Constant(func_call.args, List[FuncArg])]
    else:
        args = default_args + func_call.args

    args_result = evaluate_args(known_functions, args)

    if len(args_result.errors) > 0:
        return FuncResult(None, None, None, [], args_result.errors)

    argument_errors = []

    # TODO: simply copy.deepcopy and pop('return', None) after
    # https://github.com/python/typing/issues/306 is resolved.
    # `-> ChannelMessages` blows up if you try to copy.deepcopy the annotations.
    annotations = dict(action.__annotations__)
    return_type = annotations.pop('return', None)
    annotations = copy.deepcopy(annotations)

    # check arity mismatch
    num_keyword_args = len(action.__defaults__) if action.__defaults__ else 0
    num_positional_args = len(annotations) - num_keyword_args
    if num_positional_args > len(args_result.result):
        argument_errors.append(
            mismatching_args_messages(
                action,
                annotations,
                args_result.result,
                args_result.return_types
            )
        )

    mismatching_types = mismatching_types_messages(
        action,
        annotations,
        args_result.result,
        args_result.return_types
    )

    if len(mismatching_types) > 0:
        argument_errors.append(mismatching_types)

    if len(argument_errors) > 0:
        return FuncResult(None, None, None, [], argument_errors)

    try:
        return FuncResult(action(*args_result.result), return_type, action, args_result.result, [])
    except Exception as e:
        error_message = exception_error_messages([(func_call.func_name, e)])
        return FuncResult(None, None, None, [], [error_message])


def evaluate_args(
        known_functions: FunctionMap,
        args: List[FuncArg]) -> ArgsResult:
    result = []
    return_types = []
    all_errors = []

    for arg in args:
        if isinstance(arg, Constant):
            result.append(arg.value)
            return_types.append(arg.return_type)
        elif isinstance(arg, FuncCall):
            func_result = evaluate_func_call(known_functions, arg)
            result.append(func_result.result)
            return_types.append(func_result.return_type)

            if len(func_result.errors) > 0:
                all_errors.extend(func_result.errors)

    return ArgsResult(result, return_types, all_errors)


def exception_error_messages(errors) -> str:
    message = f'I got the following errors:\n'
    message += '```\n'

    message += '\n'.join(
        f'- {func_name} threw {error}' for (func_name, error) in errors
    )

    message += '\n```'

    return message


def mismatching_args_messages(action, annotations, arg_values, arg_types) -> str:
    message = f'I wanted things to look like for function `{action.__name__}`:\n'  # noqa: E501
    message += '```\n'

    message += '\n'.join(
        f'- {arg_name} : {arg_type}' for (arg_name, arg_type) in annotations.items()
    )

    message += '\n```'

    message += "\nBut you gave me:\n"
    message += '```\n'
    message += '\n'.join(f'- {arg_value} : {arg_type}'
                         for (arg_value, arg_type) in zip(arg_values, arg_types))
    message += '\n```'

    if len(annotations) < len(arg_values):
        return f'Too many arguments!\n{message}'
    else:
        return f'Need some more arguments!\n{message}'


def mismatching_types_messages(action, annotations, arg_values, arg_types) -> str:
    messages = []

    for (arg_value, arg_type, (arg_name, annotation)) in zip(arg_values, arg_types, annotations.items()):
        if annotation == Any or isinstance(annotation, TypeVar):
            continue

        if arg_type != annotation:
            messages.append(f'Type mistmach for function `{action.__name__}`')
            messages.append(
                f'You tried to give me a `{arg_type}` but I wanted a `{annotation}` for the arg `{arg_name}`!'  # noqa: E501
            )

    return '\n'.join(messages)
<file_sep>/slack_today_i_did/self_aware.py
import os
import subprocess
import sys
import psutil
import logging


def git_checkout(branch):
    os.system('git pull')
    os.system(f'git checkout {branch}')


def git_current_version():
    byte_text = subprocess.check_output(["git", "status"], stderr=subprocess.STDOUT)
    text = byte_text.decode()

    first_line = text.split('\n')[0]

    return first_line


def python_version():
    info = sys.version_info
    version = f'{info[0]}.{info[1]}.{info[2]}'

    return version


def ruby_version():
    byte_text = subprocess.check_output(["ruby", "--version"], stderr=subprocess.STDOUT)
    text = byte_text.decode()

    first_line = text.split('\n')[0]

    return first_line


def restart_program():
    """Restarts the current program, with file objects and descriptors
       cleanup
    """

    try:
        p = psutil.Process(os.getpid())
        for handler in p.get_open_files() + p.connections():
            os.close(handler.fd)
    except Exception as e:
        logging.error(e)

    python = sys.executable
    os.execl(python, python, *sys.argv)
<file_sep>/slack_today_i_did/rollbar.py
"""
A file for dealing with rollbar related things
"""
import requests


class Rollbar(object):
    def __init__(self, token):
        self.token = token
        self.base_url = "https://api.rollbar.com"

    def request(self, url):
        actual_url = self.base_url + url + f"?access_token={self.token}"
        return requests.get(actual_url)

    def get_item_by_id(self, id):
        response = self.request(f'/api/1/item/{id}')

        json = response.json()

        return json

    def get_item_by_counter(self, counter):
        response = self.request(f'/api/1/item_by_counter/{counter}')

        json = response.json()

        return json['result']
<file_sep>/tests/test_today_i_did_bot.py
import pytest
import os
import inspect
import typing
import functools
import datetime
from slack_today_i_did import parser
from slack_today_i_did.bot_file import TodayIDidBot

MOCK_PERSON = 'dave'
MOCK_CHANNEL = '#durp'
MOCK_START_SESSION_TEXT = 'start-session'
MOCK_TEST_FILE = '.testdata_sessions'
MOCK_COMMAND_HISTORY = '.testdata_test_today_i_did_bot_command_history'


@pytest.fixture
def bot(tmpdir):
    reports_dir = tmpdir.mkdir("reports")
    return TodayIDidBot(
        '',
        rollbar_token='',
        elm_repo=None,
        reports_dir=str(reports_dir),
        command_history_file=str(tmpdir.join('command_history.json')))


def test_start_session(mocker, bot, message_context):
    mocked_channel_message = mocker.patch.object(bot, 'send_channel_message')

    with message_context(bot, sender=MOCK_PERSON):
        bot.parse_direct_message({
            'user': MOCK_PERSON,
            'channel': MOCK_CHANNEL,
            'text': MOCK_START_SESSION_TEXT
        })

    assert MOCK_CHANNEL == mocked_channel_message.call_args[0][0]
    assert 'Started a session for you' in mocked_channel_message.call_args[0][1]
    assert mocked_channel_message.call_count == 1


def test_save_and_load_known_user_func_history(mocker, bot, message_context):
    dangerous_commands = ('reload', 'reload-funcs')
    default_args = ('channel',)
    sample_args = {
        int: 'NUM 1',
        str: 'orange',
        datetime.datetime: 'NOW',
        typing.List[str]: 'FOR blurb,blabi',
        inspect._empty: 'help', # guessing it's help's `args` param
    }

    expected_func_names = []

    for command, func in bot.known_user_functions().items():
        if command in dangerous_commands:
            continue

        signature = inspect.signature(func)
        args = (sample_args[param.annotation]
                for (name, param) in signature.parameters.items()
                if name not in default_args)
        message_text = f'{command} {" ".join(args)}'.strip()

        with message_context(bot, sender=MOCK_PERSON):
            spy = mocker.spy(bot, func.__name__)
            mocker.patch.object(bot.rollbar, 'get_item_by_counter', return_value={})
            mocker.patch.object(bot, 'repo')

            # preserve important attributes on the spy
            functools.update_wrapper(spy, func)
            if parser.is_metafunc(func):
                parser.metafunc(spy)

            bot.parse_direct_message({
                'user': MOCK_PERSON,
                'channel': MOCK_CHANNEL,
                'text': message_text
            })

            assert spy.call_count == 1
            expected_func_names.append(func.__name__)

        second_bot = TodayIDidBot('', command_history_file=bot.command_history_file, reports_dir=bot.reports_dir)
        saved_func_names = [command['action'].__name__
                          for command in second_bot.command_history.history[MOCK_CHANNEL]]
        assert saved_func_names == expected_func_names
<file_sep>/slack_today_i_did/better_slack.py
"""
A slack client with much better async support
"""

from slackclient import SlackClient
import websockets
import asyncio
import ssl
import json

ssl_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)

# os x is dumb so this fixes the openssl cert import
try:
    ssl_context.load_verify_locations('/usr/local/etc/openssl/cert.pem')
except:
    pass


# we need to redefine these because the slackclient library is bad
class SlackLoginError(Exception):
    pass


class SlackConnectionError(Exception):
    pass


class BetterSlack(SlackClient):
    """ a better slack client with async/await support """

    def __init__(self, *args, **kwargs):
        SlackClient.__init__(self, *args, **kwargs)
        self.known_users = {}
        self._conn = None
        self.message_queue = []
        self._should_reconnect = False
        self._in_count = 0

    async def __aenter__(self):
        reply = self.server.api_requester.do(self.token, "rtm.start")

        if reply.status_code != 200:
            raise SlackConnectionError
        else:
            login_data = reply.json()

            if login_data["ok"]:
                self.ws_url = login_data['url']
                if not self._should_reconnect:
                    self.server.parse_slack_login_data(login_data)
                self._conn = websockets.connect(self.ws_url, ssl=ssl_context)
            else:
                raise SlackLoginError

        self.websocket = await self._conn.__aenter__()
        return self

    async def __aexit__(self, *args, **kwargs):
        await self._conn.__aexit__(*args, **kwargs)

    async def main_loop(self, parser=None, on_tick=None):
        async with self as self:
            while True:
                while len(self.message_queue) > 0:
                    await self.websocket.send(self.message_queue.pop(0))

                if parser is not None:
                    incoming = await self.get_message()
                    try:
                        parser(incoming)
                    except Exception as e:
                        print(f'Error: {e}')
                if on_tick() is not None:
                    on_tick()
                self._in_count += 1

                if self._in_count > (0.5 * 60 * 3):
                    self.ping()
                    self._in_count = 0

                asyncio.sleep(0.5)

    async def get_message(self):
        incoming = await self.websocket.recv()
        json_data = ""
        json_data += "{0}\n".format(incoming)
        json_data = json_data.rstrip()

        data = []

        if json_data != '':
            for d in json_data.split('\n'):
                data.append(json.loads(d))

        for item in data:
            self.process_changes(item)

        return data

    def ping(self):
        return self.send_to_websocket({"type": "ping"})

    def send_to_websocket(self, data):
        """
        Send a JSON message directly to the websocket. See
        `RTM documentation <https://api.slack.com/rtm` for allowed types.

        :Args:
            data (dict) the key/values to send the websocket.

        """
        data = json.dumps(data)
        self.message_queue.append(data)

    def set_known_users(self):
        response = self.api_call('users.list')

        if not response['ok']:
            return

        for member in response['members']:
            self.known_users[member['name']] = member['id']

    def user_name_from_id(self, my_id):
        for (name, id) in self.known_users.items():
            if id == my_id:
                return name
        return None

    def open_chat(self, name: str) -> str:
        if name not in self.known_users:
            self.set_known_users()

        person = self.known_users[name]
        response = self.api_call('im.open', user=person)

        return response['channel']['id']

    def send_message(self, name: str, message: str) -> None:
        id = self.open_chat(name)

        json = {"type": "message", "channel": id, "text": message}
        self.send_to_websocket(json)

    def send_channel_message(self, channel: str, message: str) -> None:
        json = {"type": "message", "channel": channel, "text": message}
        self.send_to_websocket(json)

    def connected_user(self, username: str) -> str:
        if username not in self.known_users:
            self.set_known_users()

        return self.known_users[username]

    def attachment_strings(self, attachment):
        strings = []

        for (k, v) in attachment.items():
            if isinstance(v, str):
                strings.append(v)

        for field in attachment.get('fields', []):
            strings.append(field['title'])
            strings.append(field['value'])

        return strings
<file_sep>/slack_today_i_did/known_names.py
import json
from typing import List


class KnownNames(object):
    """ Alias a person to a group of names, with saving and loading from disk
    """
    def __init__(self):
        self.people = {}

    def add_name(self, person: str, name: str) -> None:
        if person not in self.people:
            self.people[person] = []

        self.people[person].append(name)

    def get_names(self, person: str) -> List[str]:
        return self.people.get(person, [])

    def load_from_file(self, filename: str) -> None:
        try:
            with open(filename) as f:
                as_json = json.load(f)
        except FileNotFoundError:
            return

        for (person, names) in as_json['people'].items():
            self.people[person] = names

    def save_to_file(self, filename: str) -> None:
        with open(filename, 'w') as f:
            json.dump({'people': self.people}, f)
<file_sep>/tests/test_generic_bot.py
import pytest

from slack_today_i_did.generic_bot import GenericSlackBot

MOCK_PERSON = 'dave'
MOCK_CHANNEL = '#general'


@pytest.fixture
def bot():
    return GenericSlackBot('')


def test_func_that_return(mocker, bot, message_context):
    mocked_channel_message = mocker.patch.object(bot, 'send_channel_message')

    with message_context(bot, sender=MOCK_PERSON):
        bot.parse_direct_message({
            'user': MOCK_PERSON,
            'channel': MOCK_CHANNEL,
            'text': 'func-that-return None'
        })

    assert MOCK_CHANNEL == mocked_channel_message.call_args[0][0]
    assert "The following functions return" in mocked_channel_message.call_args[0][1]
    assert mocked_channel_message.call_count == 1


def test_help_without_arg(mocker, bot, message_context):
    mocked_channel_message = mocker.patch.object(bot, 'send_channel_message')

    with message_context(bot, sender=MOCK_PERSON):
        bot.parse_direct_message({
            'user': MOCK_PERSON,
            'channel': MOCK_CHANNEL,
            'text': 'help'
        })

    assert MOCK_CHANNEL == mocked_channel_message.call_args[0][0]
    assert "Main functions:" in mocked_channel_message.call_args[0][1]
    assert mocked_channel_message.call_count == 1


def test_help_on_metafunc(mocker, bot, message_context):
    mocked_channel_message = mocker.patch.object(bot, 'send_channel_message')

    with message_context(bot, sender=MOCK_PERSON):
        bot.parse_direct_message({
            'user': MOCK_PERSON,
            'channel': MOCK_CHANNEL,
            'text': 'help help'
        })

    assert MOCK_CHANNEL == mocked_channel_message.call_args[0][0]
    assert "I'll tell you about it" in mocked_channel_message.call_args[0][1]
    assert "FuncArg" not in mocked_channel_message.call_args[0][1]
    assert mocked_channel_message.call_count == 1


def test_help_on_func(mocker, bot, message_context):
    mocked_channel_message = mocker.patch.object(bot, 'send_channel_message')

    with message_context(bot, sender=MOCK_PERSON):
        bot.parse_direct_message({
            'user': MOCK_PERSON,
            'channel': MOCK_CHANNEL,
            'text': 'help func-that-return'
        })

    assert MOCK_CHANNEL == mocked_channel_message.call_args[0][0]
    assert "functions that return things" in mocked_channel_message.call_args[0][1]
    assert "text : <class 'str'>" in mocked_channel_message.call_args[0][1]
    assert mocked_channel_message.call_count == 1


def test_help_on_mispelled_func(mocker, bot, message_context):
    mocked_channel_message = mocker.patch.object(bot, 'send_channel_message')

    with message_context(bot, sender=MOCK_PERSON):
        bot.parse_direct_message({
            'user': MOCK_PERSON,
            'channel': MOCK_CHANNEL,
            'text': 'help helc'
        })

    assert MOCK_CHANNEL == mocked_channel_message.call_args[0][0]
    assert "I did find the following functions" in mocked_channel_message.call_args[0][1]
    assert mocked_channel_message.call_count == 1


def test_help_on_mispelled_something(mocker, bot, message_context):
    mocked_channel_message = mocker.patch.object(bot, 'send_channel_message')

    with message_context(bot, sender=MOCK_PERSON):
        bot.parse_direct_message({
            'user': MOCK_PERSON,
            'channel': MOCK_CHANNEL,
            'text': 'help hdebf8u3ijr9jndsiix'
        })

    assert MOCK_CHANNEL == mocked_channel_message.call_args[0][0]
    assert "I don't know what you mean and have no suggestions" in mocked_channel_message.call_args[0][1]
    assert mocked_channel_message.call_count == 1
<file_sep>/slack_today_i_did/command_history.py
from slack_today_i_did.type_aware import json


def makes_state_change(f):
    def wrapper(self, *args, **kwargs):
        f(self, *args, **kwargs)
        self.needs_save = True
    return wrapper


def saves_state(f):
    def wrapper(self, *args, **kwargs):
        f(self, *args, **kwargs)
        self.needs_save = False
    return wrapper


class CommandHistory(object):

    @saves_state
    def __init__(self):
        self.history = {}

    @makes_state_change
    def add_command(self, channel, command, args):
        if channel not in self.history:
            self.history[channel] = []

        self.history[channel].append({'action': command, 'args': args})

    def last_command(self, channel):
        if channel not in self.history:
            return None
        return self.history[channel][-1]

    def __eq__(self, other):
        if not isinstance(other, self.__class__):
            return False

        for channel in self.history:
            if channel not in other.history:
                return False

            if self.history[channel] != other.history[channel]:
                return False

        return True

    @saves_state
    def load_from_file(self, known_tokens, known_types, filename: str) -> None:
        """ Load command history from a file """
        try:
            with open(filename) as f:
                as_json = json.load(f, known_types=known_types)
        except FileNotFoundError:
            return
        except:
            # sometimes bad things happen to good people
            return

        for (channel, commands) in as_json['channels'].items():
            for command_entry in commands:
                command_entry = self._command_entry_from_json(known_tokens, command_entry)
                self.add_command(channel, command_entry['action'], command_entry['args'])

    @saves_state
    def save_to_file(self, filename: str) -> None:
        """ save command history to a file """
        channels = {
            'channels': {
                channel: [self._command_entry_to_json(command) for command in commands]
                for (channel, commands) in self.history.items()
            }
        }

        # if we can't save it, exit early
        try:
            channel_json = json.dumps(channels)
            with open(filename, 'w') as f:
                f.write(channel_json)
        except:
            return None

        self.needs_save = False

    def _command_entry_to_json(self, command_entry):
        action = command_entry['action'].__name__

        return {
            'action': action,
            'args': command_entry['args']
        }

    def _no_longer_exists(self, *args, **kwargs) -> str:
        return 'this thing no longer exists!'

    def _command_entry_from_json(self, known_tokens, json):
        return {
            'action': known_tokens.get(json['action'], self._no_longer_exists),
            'args': json['args']
        }
<file_sep>/.flake8
[flake8]
max-line-length: 120
doctests = True