secretary
/
waitonbot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Big checkin pre-testing. Made types static in most files

This commit is contained in:
Jacob Henry 2018-11-03 22:49:42 -04:00
parent ce3f979114
commit 855a43910f
8 changed files with 217 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
channel_util.py
View File

@ -1,6 +1,8 @@
import slack_util
from typing import Match
DB_NAME = "channel_priveleges"
from slackclient import SlackClient
import slack_util
# Useful channels
GENERAL = "C0CFHPNEM"
@ -11,10 +13,8 @@ HOUSEJOBS = "CDWDDTAT0"
# Callback for telling what channel we in
def channel_check_callback(slack, msg, match):
def channel_check_callback(slack: SlackClient, msg: dict, match: Match) -> None:
# Sets the users scroll
# with shelve.open(DB_NAME) as db:
rest_of_msg = match.group(1).strip()
rest_of_msg = rest_of_msg.replace("<", "lcaret")
rest_of_msg = rest_of_msg.replace(">", "rcaret")

38
identifier.py
View File

@ -3,6 +3,9 @@ Allows users to register their user account as a specific scroll
"""
import shelve
from typing import Optional, List, Match
from slackclient import SlackClient
import slack_util
import scroll_util
@ -17,7 +20,9 @@ NON_REG_MSG = ("You currently have no scroll registered. To register, type\n"
def identify_callback(slack, msg, match):
# Sets the users scroll
"""
Sets the users scroll
"""
with shelve.open(DB_NAME) as db:
# Get the query
query = match.group(1).strip()
@ -34,8 +39,10 @@ def identify_callback(slack, msg, match):
slack_util.reply(slack, msg, result)
def identify_other_callback(slack, msg, match):
# Sets the users scroll
def identify_other_callback(slack: SlackClient, msg: dict, match: Match):
"""
Sets another users scroll
"""
with shelve.open(DB_NAME) as db:
# Get the query
user = match.group(1).strip()
@ -56,7 +63,10 @@ def identify_other_callback(slack, msg, match):
# noinspection PyUnusedLocal
def check_callback(slack, msg, match):
def check_callback(slack: SlackClient, msg: dict, match: Match):
"""
Replies with the users current scroll assignment
"""
# Tells the user their current scroll
with shelve.open(DB_NAME) as db:
try:
@ -69,13 +79,15 @@ def check_callback(slack, msg, match):
# noinspection PyUnusedLocal
def name_callback(slack, msg, match):
# Tells the user what slack thinks their name is
"""
Tells the user what it thinks the calling users name is.
"""
with shelve.open(DB_NAME) as db:
try:
scroll = db[msg.get("user")]
brother = scroll_util.find_by_scroll(scroll)
if brother:
result = "The bot thinks your name is {}".format(brother["name"])
result = "The bot thinks your name is {}".format(brother.name)
else:
result = "The bot couldn't find a name for scroll {}".format(scroll)
except (KeyError, ValueError):
@ -85,18 +97,19 @@ def name_callback(slack, msg, match):
slack_util.reply(slack, msg, result)
def lookup_msg_brother(msg):
def lookup_msg_brother(msg: dict) -> Optional[scroll_util.Brother]:
"""
Finds the real-world name of whoever posted msg.
Utilizes their bound-scroll.
:return: brother dict or None
"""
return lookup_slackid_brother(msg.get("user"))
def lookup_slackid_brother(slack_id):
def lookup_slackid_brother(slack_id: str) -> Optional[scroll_util.Brother]:
"""
Gets whatever brother the userid is registered to
:return: Brother dict or None
:return: Brother object or None
"""
with shelve.open(DB_NAME) as db:
try:
@ -106,17 +119,18 @@ def lookup_slackid_brother(slack_id):
return None
def lookup_brother_userids(brother):
def lookup_brother_userids(brother: scroll_util.Brother) -> List[str]:
"""
Returns a list of all userids associated with the given brother.
:param brother: Std brother dict as specc'd in scroll_util
:param brother: Brother to lookup scrolls for
:return: List of user id strings (may be empty)
"""
with shelve.open(DB_NAME) as db:
keys = db.keys()
result = []
for user_id in keys:
if db[user_id] == brother["scroll"]:
if db[user_id] == brother.scroll:
result.append(user_id)
return result

46
job_nagger.py
View File

@ -6,24 +6,29 @@ import channel_util
SHEET_ID = "1lPj9GjB00BuIq9GelOWh5GmiGsheLlowPnHLnWBvMOM"
# Note: These ranges use named range feature of google sheets. To edit range of jobs, edit the named range in Data -> Named Ranges
# Note: These ranges use named range feature of google sheets.
# To edit range of jobs, edit the named range in Data -> Named Ranges
eight_job_range = "EightJobs" # Format: Job Day Bro
fiftythree_job_range = "FiftyThreeJobs"
class Job(object):
def __init__(self, house, job_name, day, brother_name):
"""
Object representing a house job
"""
def __init__(self, house: str, job_name: str, day: str, brother_name: str):
self.house, self.job_name, self.day, self.brother_name = house, job_name, day, brother_name
self.day = self.day.lower().strip()
def lookup_brother_slack_id(self):
brother_dict = scroll_util.find_by_name(self.brother_name)
return identifier.lookup_brother_userids(brother_dict)
def get_jobs(day=None):
"""
Retrieves the house jobs for a given day.
If no day is provided, returns all house jobs for the week.
def nag_callback(slack, msg, match):
# Get the day
day = match.group(1).lower().strip()
:param day: Day to compare the day column to. If equal, keep. If Not, discard. If day==None, ignore this filter
:return: list of Job objects
"""
# Get the spreadsheet section
eight_jobs = google_api.get_sheet_range(SHEET_ID, eight_job_range)
@ -41,20 +46,35 @@ def nag_callback(slack, msg, match):
jobs = eight_jobs + ft_jobs
# Filter to day
jobs = [j for j in jobs if j.day == day]
if day:
jobs = [j for j in jobs if j.day == day]
return jobs
def nag_callback(slack, msg, match):
# Get the day
day = match.group(1).lower().strip()
jobs = get_jobs(day)
# If no jobs found, somethings up. Probably mispelled day.
if not jobs:
slack_util.reply(slack, msg, "No jobs found. Check that the day is spelled correctly, with no extra symbols", in_thread=True)
slack_util.reply(slack, msg, "No jobs found. Check that the day is spelled correctly, with no extra symbols",
in_thread=True)
return
# Nag each
response = "Do yer jerbs! They are as follows:\n"
for job in jobs:
# Make the row template
response += "({}) {} -- ".format(job.house, job.job_name)
ids = job.lookup_brother_slack_id()
if ids:
for slack_id in ids:
# Find the people to @
brother = scroll_util.find_by_name(job.brother_name)
brother_slack_ids = identifier.lookup_brother_userids(brother)
if brother_slack_ids:
for slack_id in brother_slack_ids:
response += "<@{}> ".format(slack_id)
else:
response += "{} (scroll missing. Please register for @ pings!)".format(job.brother_name)

108
job_signoff.py
View File

@ -1,8 +1,12 @@
from typing import List, Any, Match, Tuple
from fuzzywuzzy import process
from slackclient import SlackClient
import channel_util
import google_api
import identifier
import job_nagger
import slack_util
import scroll_util
@ -10,40 +14,41 @@ SHEET_ID = "1lPj9GjB00BuIq9GelOWh5GmiGsheLlowPnHLnWBvMOM"
# Note: These ranges use named range feature of google sheets.
# To edit range of jobs, edit the named range in Data -> Named Ranges in the Goole Sheets page
output_range = "JobScoreTracker"
output_range = "JobScoreTrackerV2"
MIN_RATIO = 0.9
SIGNOFF_REWARD = 0.1
SAFETY_DELAY = 5
# Used to track a sufficiently shitty typed name
class BadName(Exception):
def __init__(self, name, score):
self.name = name
self.score = score
def as_response(self):
return "Unable to perform operation. Best name match {} had a match score of {}, falling short of match ratio " \
"{}. Please type name better.".format(self.name, self.score, MIN_RATIO)
def get_curr_points():
def get_curr_points() -> List[Tuple[str, float, str]]:
"""
:return: The current contents of the output range
"""
# Get the current stuff
curr_output = google_api.get_sheet_range(SHEET_ID, output_range)
all_jobs = job_nagger.get_jobs()
# Each element: force length of 2.
def row_fixer(r):
# Each element: force length of 3. Fmt name, score, job
def row_fixer(r: List[Any]) -> Tuple[str, float, str]:
if len(r) == 0:
return ["", 0]
elif len(r) == 1:
return [r[0], 0]
return "", 0, ""
else:
try:
v = float(r[1])
except ValueError:
v = 0
return [r[0], v]
# Get the appropriate score
if len(r) > 1:
try:
curr_score = float(r[1])
except ValueError:
curr_score = 0
else:
curr_score = 0
# Find the current job
job = ""
for j in all_jobs:
if j.brother_name == r[0]:
job = "{} ({} - {})".format(j.job_name, j.day, j.house)
break
return r[0], curr_score, job
# Fix each row
curr_output = [row_fixer(x) for x in curr_output]
@ -52,11 +57,30 @@ def get_curr_points():
return curr_output
def put_points(vals):
def put_points(vals: List[Tuple[str, float, str]]) -> None:
# Turn the tuples to lists
vals = [list(row) for row in vals]
google_api.set_sheet_range(SHEET_ID, output_range, vals)
def signoff_callback(slack, msg, match):
def alert_user(slack: SlackClient, name: str, saywhat: str) -> None:
"""
DM a brother saying something
"""
brother_dict = scroll_util.find_by_name(name)
# We do this as a for loop just in case multiple people reg. to same scroll for some reason (e.g. dup accounts)
for slack_id in identifier.lookup_brother_userids(brother_dict):
dm_id = slack_util.im_channel_for_id(slack, slack_id)
if dm_id:
slack_util.reply(slack, None, saywhat, to_channel=dm_id, in_thread=False)
else:
print("Warning: unable to find dm for brother {}".format(brother_dict))
def signoff_callback(slack: SlackClient, msg: dict, match: Match) -> None:
"""
Callback to signoff a user.
"""
# Find the index of our person.
name = match.group(1)
@ -71,26 +95,12 @@ def signoff_callback(slack, msg, match):
"You ({}) were credited with the signoff".format(bro_name, bro_total, ass_name))
alert_user(slack, bro_name,
"You, who we believe to be {}, just had your house job signed off by {}!".format(bro_name, ass_name))
except BadName as e:
except scroll_util.BadName as e:
# We didn't find a name - no action was performed.
slack_util.reply(slack, msg, e.as_response())
def alert_user(slack, name, saywhat):
"""
DM a brother saying something
"""
brother_dict = scroll_util.find_by_name(name)
# We do this as a for loop just in case multiple people reg. to same scroll for some reason (e.g. dup accounts)
for slack_id in identifier.lookup_brother_userids(brother_dict):
dm_id = slack_util.im_channel_for_id(slack, slack_id)
if dm_id:
slack_util.reply(slack, None, saywhat, to_channel=dm_id, in_thread=False)
else:
print("Warning: unable to find dm for brother {}".format(brother_dict))
def punish_callback(slack, msg, match):
def punish_callback(slack: SlackClient, msg: dict, match: Match) -> None:
# Find the index of our person.
name = match.group(2)
@ -112,12 +122,12 @@ def punish_callback(slack, msg, match):
"Perhaps the asshoman made a mistake when they first signed you off.\n"
"If you believe that they undid the signoff accidentally, go talk to them".format(bro_name, signer))
except BadName as e:
except scroll_util.BadName as e:
# We didn't find a name - no action was performed.
slack_util.reply(slack, msg, e.as_response())
def adjust_scores(*name_delta_tuples):
def adjust_scores(*name_delta_tuples: Tuple[str, float]) -> List[Tuple[str, float, str]]:
# Get the current stuff
points = get_curr_points()
names = [p[0] for p in points]
@ -130,16 +140,17 @@ def adjust_scores(*name_delta_tuples):
# If bad ratio, error
if ratio < MIN_RATIO:
raise BadName(target_name, ratio)
raise scroll_util.BadName(target_name, ratio, MIN_RATIO)
# Where is he in the list?
target_index = names.index(target_name)
# Get his current score
curr_score = points[target_index][1]
curr_job = points[target_index][2]
# target should be in the form index, (name, score)
target_new = [target_name, curr_score + delta]
target_new = target_name, curr_score + delta, curr_job
# Put it back
points[target_index] = target_new
@ -154,8 +165,9 @@ def adjust_scores(*name_delta_tuples):
return [points[i] for i in modified_user_indexes]
def reset_callback(slack, msg, match):
pass
# noinspection PyUnusedLocal
def reset_callback(slack: SlackClient, msg: dict, match: Match) -> None:
raise NotImplementedError()
# reset_callback()

8
management_commands.py
View File

@ -1,8 +1,12 @@
from typing import Match, List
from slackclient import SlackClient
import channel_util
import slack_util
def list_hooks_callback_gen(hooks):
def list_hooks_callback_gen(hooks: List[slack_util.Hook]):
# noinspection PyUnusedLocal
def callback(slack, msg, match):
slack_util.reply(slack, msg, "\n".join(hook.pattern for hook in hooks))
@ -12,7 +16,7 @@ def list_hooks_callback_gen(hooks):
# Gracefully reboot to reload code changes
# noinspection PyUnusedLocal
def reboot_callback(slack, msg, match):
def reboot_callback(slack: SlackClient, msg: dict, match: Match) -> None:
response = "Ok. Rebooting..."
slack_util.reply(slack, msg, response)
exit(0)

69
scroll_util.py
View File

@ -4,25 +4,37 @@ Only really kept separate for neatness sake.
"""
import re
from typing import List, Optional, Match
from fuzzywuzzy import process
from slackclient import SlackClient
import slack_util
class Brother(object):
"""
Represents a brother.
"""
def __init__(self, name: str, scroll: int):
self.name = name
self.scroll = scroll
# load the family tree
familyfile = open("sortedfamilytree.txt", 'r')
# Parse out
brother_match = re.compile(r"([0-9]*)~(.*)")
brothers = [brother_match.match(line) for line in familyfile]
brothers = [m for m in brothers if m]
brothers = [{
"scroll": int(m.group(1)),
"name": m.group(2)
} for m in brothers]
brothers_matches = [brother_match.match(line) for line in familyfile]
brothers_matches = [m for m in brothers_matches if m]
brothers: List[Brother] = [Brother(m.group(2), int(m.group(1))) for m in brothers_matches]
def scroll_callback(slack, msg, match):
def scroll_callback(slack: SlackClient, msg: dict, match: Match) -> None:
"""
Finds the scroll of a brother, or the brother of a scroll, based on msg text.
"""
# Get the query
query = match.group(1).strip()
@ -33,7 +45,7 @@ def scroll_callback(slack, msg, match):
except ValueError:
result = find_by_name(query)
if result:
result = "Brother {} has scroll {}".format(result["name"], result["scroll"])
result = "Brother {} has scroll {}".format(result.name, result.scroll)
else:
result = "Couldn't find brother {}".format(query)
@ -41,19 +53,48 @@ def scroll_callback(slack, msg, match):
slack_util.reply(slack, msg, result)
def find_by_scroll(scroll):
def find_by_scroll(scroll: int) -> Optional[Brother]:
"""
Lookups a brother in the family list, using their scroll.
:param scroll: The integer scroll to look up
:return: The brother, or None
"""
for b in brothers:
if b["scroll"] == scroll:
if b.scroll == scroll:
return b
return None
def find_by_name(name):
# coerce name into dict form
name = {"name": name}
# Used to track a sufficiently shitty typed name
class BadName(Exception):
def __init__(self, name: str, score: float, threshold: float):
self.name = name
self.score = score
self.threshold = threshold
def as_response(self) -> str:
return "Unable to perform operation. Best name match {} had a match score of {}, falling short of minimum " \
"match ratio {}. Please type name better.".format(self.name, self.score, self.threshold)
def find_by_name(name: str, threshold: Optional[float] = None) -> Brother:
"""
Looks up a brother by name. Raises exception if threshold provided and not met.
:param threshold: Minimum match ratio to accept. Can be none.
:param name: The name to look up, with a fuzzy search
:return: The best-match brother
"""
# Really quikly mog name into Brother, so the processor function works fine
name_bro = Brother(name, -1)
# Do fuzzy match
return process.extractOne(name, brothers, processor=lambda b: b["name"])[0]
found = process.extractOne(name_bro, brothers, processor=lambda b: b.name)
if (not threshold) or found[1] > threshold:
return found[0]
else:
raise BadName(name, found[1], threshold)
scroll_hook = slack_util.Hook(scroll_callback, pattern=r"scroll\s+(.*)")

29
slack_util.py
View File

@ -1,14 +1,17 @@
from time import sleep
import re
from slackclient import SlackClient
import channel_util
from typing import Any
from typing import Any, Optional, Generator, Match, Callable, List
"""
Slack helpers. Separated for compartmentalization
"""
def reply(slack, msg, text, in_thread=True, to_channel=None):
def reply(slack: SlackClient, msg: dict, text: str, in_thread: bool = True, to_channel: str = None) -> None:
"""
Sends message with "text" as its content to the channel that message came from
"""
@ -19,14 +22,13 @@ def reply(slack, msg, text, in_thread=True, to_channel=None):
# Send in a thread by default
if in_thread:
thread = (msg.get("thread_ts") # In-thread case - get parent ts
or msg.get("ts")) # Not in-thread case - get msg itself ts
result = slack.rtm_send_message(channel=to_channel, message=text, thread=thread)
or msg.get("ts")) # Not in-thread case - get msg itself ts
slack.rtm_send_message(channel=to_channel, message=text, thread=thread)
else:
result = slack.rtm_send_message(channel=to_channel, message=text)
return result
slack.rtm_send_message(channel=to_channel, message=text)
def im_channel_for_id(slack, user_id):
def im_channel_for_id(slack: SlackClient, user_id: str) -> Optional[str]:
conversations = slack.api_call("conversations.list", types="im")
if conversations["ok"]:
channels = conversations["channels"]
@ -37,7 +39,7 @@ def im_channel_for_id(slack, user_id):
class SlackDebugCondom(object):
def __init__(self, actual_slack):
def __init__(self, actual_slack: SlackClient):
self.actual_slack = actual_slack
def __getattribute__(self, name: str) -> Any:
@ -49,6 +51,7 @@ class SlackDebugCondom(object):
kwargs["channel"] = channel_util.BOTZONE
kwargs["thread"] = None
self.actual_slack.rtm_send_message(*args, **kwargs)
return override_send_message
else:
# Default behaviour. Try to give the self first, elsewise give the child
@ -58,7 +61,7 @@ class SlackDebugCondom(object):
return self.actual_slack.__getattribute__(name)
def message_stream(slack):
def message_stream(slack) -> Generator[dict]:
"""
Generator that yields messages from slack.
Messages are in standard api format, look it up.
@ -80,7 +83,11 @@ def message_stream(slack):
class Hook(object):
def __init__(self, callback, pattern=None, channel_whitelist=None, channel_blacklist=None):
def __init__(self,
callback: Callable[SlackClient, dict, Match],
pattern: str = None,
channel_whitelist: Optional[List[str]] = None,
channel_blacklist: Optional[List[str]] = None):
# Save all
self.pattern = pattern
self.channel_whitelist = channel_whitelist
@ -99,7 +106,7 @@ class Hook(object):
else:
raise Exception("Cannot whitelist and blacklist")
def check(self, slack, msg):
def check(self, slack: SlackClient, msg: dict) -> bool:
# Fail if pattern invalid
match = re.match(self.pattern, msg['text'], flags=re.IGNORECASE)
if match is None:

22
slavestothemachine.py
View File

@ -1,3 +1,7 @@
from typing import Match
from slackclient import SlackClient
import channel_util
import slack_util
import identifier
@ -10,11 +14,12 @@ lookup_format = "{}\s+(\d+)"
DB_NAME = "towels_rolled"
def fmt_work_dict(work_dict):
def fmt_work_dict(work_dict: dict) -> str:
return ",\n".join(["{} × {}".format(job, count) for job, count in sorted(work_dict.items())])
def count_work_callback(slack, msg, match):
# noinspection PyUnusedLocal
def count_work_callback(slack: SlackClient, msg: dict, match: Match) -> None:
with shelve.open(DB_NAME) as db:
text = msg["text"].lower().strip()
@ -58,7 +63,8 @@ def count_work_callback(slack, msg, match):
slack_util.reply(slack, msg, congrats)
def dump_work_callback(slack, msg, match):
# noinspection PyUnusedLocal
def dump_work_callback(slack: SlackClient, msg: dict, match: Match) -> None:
with shelve.open(DB_NAME) as db:
# Dump out each user
keys = db.keys()
@ -69,13 +75,13 @@ def dump_work_callback(slack, msg, match):
del db[user_id]
# Get the name
brother_name = identifier.lookup_slackid_brother(user_id)
if brother_name is None:
brother_name = user_id
brother = identifier.lookup_slackid_brother(user_id)
if brother is None:
brother = user_id
else:
brother_name = brother_name["name"]
brother = brother.name
result.append("{} has done:\n{}".format(brother_name, fmt_work_dict(work)))
result.append("{} has done:\n{}".format(brother, fmt_work_dict(work)))
result.append("Database wiped. Next dump will show new work since the time of this message")
# Send it back