def _event_handler(event_type, slack_event):
"""
A helper function that routes events from Slack to our Bot
by event type and subtype.
Parameters
----------
event_type : str
type of event received from Slack
slack_event : dict
JSON response from a Slack reaction event
Returns
----------
obj
Response object with 200 - ok or 500 - No Event Handler error
"""
team_id = slack_event["team_id"]
# ================ Team Join Events =============== #
# When the user first joins a team, the type of event will be team_join
if event_type == "team_join":
pass
else:
pass
python类Bot()的实例源码
def slack_oauth(request):
code = request.GET['code']
params = {
'code': code,
'client_id': settings.SLACK_CLIENT_ID,
"client_secret": settings.SLACK_CLIENT_SECRET
}
url = 'https://slack.com/api/oauth.access'
json_response = requests.get(url, params)
data = json.loads(json_response.text)
Team.objects.get_or_create(
name=data['team_name'],
team_id=data['team_id'],
bot_user_id=data['bot']['bot_user_id'],
bot_access_token=data['bot']['bot_access_token']
)
return HttpResponse('Bot added to your Slack team!')
def clean(username, password, delay, b=None): # works for only me!
if(b==None):
b = Bot(username, password)
with open("mcclanewhitelist.txt", "r") as f:
whitelist = { line.split(" ")[0] : line.split(" ")[1] for line in f.readlines() }
time.sleep(2)
follower_count = b.get_follower_count(b.id)
print("Followers: "+str(follower_count))
following_count = b.get_following_count(b.id)
print("Following: "+str(following_count))
followers_list = b.get_followers(b.id, follower_count)
following = b.get_following(b.id, following_count)
followers = {}
for thing in followers_list:
followers[thing['node']['id']] = 1
for thing in following:
print(thing)
if not thing['node']['id'] in followers:
b.unfollow(thing['node']['id'])
time.sleep(delay)
def WayneWashington(username, password, delay, hashtags, compliments, b=None):
if(b==None):
b = Bot(username, password)
while(True):
try:
#get the media for a random hashtag
tag = hashtags[randint(0,len(hashtags)-1)].strip()
media = b.get_media_from_hashtag(tag)
#iterate through, like all and comment on a few
for i in range(len(media)):
b.like(media[i]['id'], tag)
time.sleep(1)
if(randint(0,9) == 0):
b.comment(compliments[randint(0, len(compliments)-1)], media[i]['id'], tag)
time.sleep(1)
time.sleep(randint(delay-5,delay+5))
except Exception as e:
print(str(e))
time.sleep(delay)
def start_slack_event_loop():
"""Main event loop for program."""
bot = Bot(slack_client, os.environ['SLACKBOT_ID'])
# second delay between reading from Slack RTM firehose
READ_WEBSOCKET_DELAY = 1
if slack_client.rtm_connect():
print('Bot connected and running!')
while True:
stream = bot.slack_client.rtm_read()
print(stream)
if stream and len(stream) > 0:
bot.parse_slack_output(stream)
time.sleep(READ_WEBSOCKET_DELAY)
else:
print('Connection failed.')
# start connection to Slack streaming API
def slack_oauth(request):
code = request.GET['code']
params = {
'code': code,
'client_id': settings.SLACK_CLIENT_ID,
"client_secret": settings.SLACK_CLIENT_SECRET
}
url = 'https://slack.com/api/oauth.access'
json_response = requests.get(url, params)
data = json.loads(json_response.text)
Team.objects.get_or_create(
name=data['team_name'],
team_id=data['team_id'],
bot_user_id=data['bot']['bot_user_id'],
bot_access_token=data['bot']['bot_access_token']
)
return HttpResponse('Bot added to your Slack team!')
def test_loading(self):
with self.assertLogs(self.bot.logger, level='INFO') as cm:
self.bot = Bot(BotSettings, logger=self.bot.logger)
self.assertIn(f'INFO:{self.bot.logger.name}:Initializing bot', cm.output)
self.assertIn(f'INFO:{self.bot.logger.name}:Initializing vk clients', cm.output)
self.assertIn(f'INFO:{self.bot.logger.name}:Loading plugins', cm.output)
def main():
m = ChromecastManager(bot.Bot())
while True:
m.poll()
sleep(POLL_INTERVAL)
def pre_install(request):
#"""This route renders the installation page with 'Add to Slack' button."""
# Since we've set the client ID and scope on our Bot object, we can change
# them more easily while we're developing our app.
client_id = pyBot.oauth["client_id"]
scope = pyBot.oauth["scope"]
# Our template is using the Jinja templating language to dynamically pass
# our client id and scope
#return render_template("install.html", client_id=client_id, scope=scope)
return render(request, 'install.html', {'client_id': client_id, 'scope':scope})
def main():
logger.info('Alpha Bot v1.1')
config = init_config()
setup_logging(config)
bot = Bot(config)
bot.start()
def createBot(self, name, coord = [30,30]):
'''
bekommt eine Position eines Spielers und zeichnet diesen
Parameter: coord, bsp ("034", "100")
return values: -
'''
bot = Bot()
bot.nick = name
bot.rect.centerx = coord[0]
bot.rect.centery = coord[1]
self.allBots.add(bot)
def createBot(self, name, coord = [30,30]):
'''
bekommt eine Position eines Spielers und zeichnet diesen
Parameter: coord, bsp ("034", "100")
return values: -
'''
bot = Bot()
bot.nick = name
bot.rect.centerx = coord[0]
bot.rect.centery = coord[1]
bot.room = self.name
self.allBots.add(bot)
def run_bot(username, password, delay, pipe_depth, number_of_tags, compliments, hashtags):
# Create the bot
b = Bot(username, password)
time.sleep(2)
followed_list = []
for i in range(number_of_tags):
try:
#get the media for a random hashtag
tag = hashtags[randint(0,len(hashtags)-1)].strip()
media = b.get_media_from_hashtag(tag)
#iterate through, like all and comment on a few
for i in range(len(media)):
b.like(media[i]['id'], tag)
time.sleep(1)
if(randint(0,9) == 0):
b.comment(compliments[randint(0, len(compliments)-1)], media[i]['id'], tag)
time.sleep(1)
b.follow(media[i]['owner']['id'], tag)
time.sleep(1)
followed_list.append(media[i]['owner']['id'])
rand_delay = randint(delay-5,delay+5)/2
time.sleep(rand_delay)
if(len(followed_list) > pipe_depth):
b.unfollow(followed_list[0])
followed_list.pop(0)
time.sleep(rand_delay)
except Exception as e:
print(str(e))
time.sleep(delay)
#clean up
for user in followed_list:
b.unfollow(user)
def run_cycle(username, password, delay, cycle_length, compliments, hashtags):
# Create the bot
b = Bot(username, password)
time.sleep(2)
followed_list = []
#follow people
while(len(followed_list) < cycle_length):
try:
#get the media for a random hashtag
tag = hashtags[randint(0,len(hashtags)-1)].strip()
media = b.get_media_from_hashtag(tag)
#iterate through, like all and comment on a few
for i in range(len(media)):
b.like(media[i]['id'])
time.sleep(1)
if(randint(0,9) == 0):
b.comment(compliments[randint(0, len(compliments)-1)], media[i]['id'], tag)
time.sleep(1)
b.follow(media[i]['owner']['id'])
time.sleep(1)
followed_list.append(media[i]['owner']['id'])
time.sleep(randint(delay-5,delay+5))
if(len(followed_list) > cycle_length):
break
if(len(followed_list) > cycle_length):
break
except Exception as e:
print(str(e))
time.sleep(delay)
#clean up
clean(username, password, delay, b)
def mass_unfollow(username, password, n):
b = Bot(username, password)
b.mass_unfollow(n)
def __init__(self):
self.db = DB()
self.logger = Logger()
self.bot = Bot()
def pre_install():
"""This route renders the installation page with 'Add to Slack' button."""
# Since we've set the client ID and scope on our Bot object, we can change
# them more easily while we're developing our app.
client_id = pyBot.oauth["client_id"]
scope = pyBot.oauth["scope"]
# Our template is using the Jinja templating language to dynamically pass
# our client id and scope
return render_template("install.html", client_id=client_id, scope=scope)
def hears():
"""
This route listens for incoming events from Slack and uses the event
handler helper function to route events to our Bot.
"""
slack_event = json.loads(request.data)
# ============= Slack URL Verification ============ #
# In order to verify the url of our endpoint, Slack will send a challenge
# token in a request and check for this token in the response our endpoint
# sends back.
# For more info: https://api.slack.com/events/url_verification
if "challenge" in slack_event:
return make_response(slack_event["challenge"], 200, {"content_type":
"application/json"
})
# ============ Slack Token Verification =========== #
# We can verify the request is coming from Slack by checking that the
# verification token in the request matches our app's settings
if pyBot.verification != slack_event.get("token"):
message = "Invalid Slack verification token: %s \npyBot has: \
%s\n\n" % (slack_event["token"], pyBot.verification)
# By adding "X-Slack-No-Retry" : 1 to our response headers, we turn off
# Slack's automatic retries during development.
make_response(message, 403, {"X-Slack-No-Retry": 1})
# ====== Process Incoming Events from Slack ======= #
# If the incoming request is an Event we've subcribed to
if "event" in slack_event:
event_type = slack_event["event"]["type"]
# Then handle the event by event_type and have your bot respond
return _event_handler(event_type, slack_event)
# If our bot hears things that are not events we've subscribed to,
# send a quirky but helpful error response
return make_response("[NO EVENT IN SLACK REQUEST] These are not the droids\
you're looking for.", 404, {"X-Slack-No-Retry": 1})
def pre_install(request):
#"""This route renders the installation page with 'Add to Slack' button."""
# Since we've set the client ID and scope on our Bot object, we can change
# them more easily while we're developing our app.
client_id = pyBot.oauth["client_id"]
scope = pyBot.oauth["scope"]
# Our template is using the Jinja templating language to dynamically pass
# our client id and scope
#return render_template("install.html", client_id=client_id, scope=scope)
return render(request, 'install.html', {'client_id': client_id, 'scope':scope})
def for_external_key(cls, bot, key):
"""Find or create the Post with the given external key.
"""
from bot import Bot
if isinstance(bot, Bot):
bot = bot.model
_db = Session.object_session(bot)
return get_one_or_create(_db, Post, bot=bot, external_key=key)
def _event_handler(event_type, slack_event):
"""
A helper function that routes events from Slack to our Bot
by event type and subtype.
Parameters
----------
event_type : str
type of event recieved from Slack
slack_event : dict
JSON response from a Slack reaction event
Returns
----------
obj
Response object with 200 - ok or 500 - No Event Handler error"""
team_id = slack_event["team_id"]
pyBot.find_team(team_id)
if event_type == "message":
sender_id = None
if "user" in slack_event["event"]:
sender_id = slack_event["event"]["user"]
adapted_message = sr.adapt_message_to_wit(sender_id, slack_event["event"]["text"].encode('utf-8'))
message = wit.treatment(adapted_message, sender_id)
channel = slack_event["event"]["channel"]
print "SLACK DEBUG \n"
print message
pyBot.send_message(sender_id, channel, message)
return HttpResponse("OK", 200)
# ============= Event Type Not Found! ============= #
# If the event_type does not have a handler
#message = "You have not added an event handler for the %s" % event_type
# Return a helpful error message
#channel = slack_event["event"]["channel"]
#if "user" in slack_event["event"]:
# pyBot.send_message(channel, message)
return HttpResponse("OK", 200)
def hears(request):
"""
This route listens for incoming events from Slack and uses the event
handler helper function to route events to our Bot.
"""
#Wit makes our responses timeout, so we ignore Slack retries
if "HTTP_X_SLACK_RETRY_NUM" in request.META:
return HttpResponse("OK", 200)
slack_event = json.loads(request.body)
# ============= Slack URL Verification ============ #
# In order to verify the url of our endpoint, Slack will send a challenge
# token in a request and check for this token in the response our endpoint
# sends back.
# For more info: https://api.slack.com/events/url_verification
if "challenge" in slack_event:
return HttpResponse(slack_event["challenge"], 200)
#removed {"content_type":"application/json"} from flask response
# ============ Slack Token Verification =========== #
# We can verify the request is coming from Slack by checking that the
# verification token in the request matches our app's settings
if pyBot.verification != slack_event.get("token"):
print "Invalid Slack verification token: %s \npyBot has: \
%s\n\n" % (slack_event["token"], pyBot.verification)
# By adding "X-Slack-No-Retry" : 1 to our response headers, we turn off
# Slack's automatic retries during development.
return HttpResponse(message, 403)
# ====== Process Incoming Events from Slack ======= #
# If the incoming request is an Event we've subcribed to
if "event" in slack_event:
event_type = slack_event["event"]["type"]
# Then handle the event by event_type and have your bot respond
return _event_handler(event_type, slack_event)
# If our bot hears things that are not events we've subscribed to,
# send a quirky but helpful error response
return HttpResponse("[NO EVENT IN SLACK REQUEST] These are not the droids\
you're looking for.", 404)
def _event_handler(event_type, slack_event):
"""
A helper function that routes events from Slack to our Bot
by event type and subtype.
Parameters
----------
event_type : str
type of event recieved from Slack
slack_event : dict
JSON response from a Slack reaction event
Returns
----------
obj
Response object with 200 - ok or 500 - No Event Handler error
"""
team_id = slack_event["team_id"]
# ================ Team Join Events =============== #
# When the user first joins a team, the type of event will be team_join
if event_type == "team_join":
user_id = slack_event["event"]["user"]["id"]
# Send the onboarding message
pyBot.onboarding_message(team_id, user_id)
return make_response("Welcome Message Sent", 200,)
# ============== Share Message Events ============= #
# If the user has shared the onboarding message, the event type will be
# message. We'll also need to check that this is a message that has been
# shared by looking into the attachments for "is_shared".
elif event_type == "message" and slack_event["event"].get("attachments"):
user_id = slack_event["event"].get("user")
if slack_event["event"]["attachments"][0].get("is_share"):
# Update the onboarding message and check off "Share this Message"
pyBot.update_share(team_id, user_id)
return make_response("Welcome message updates with shared message",
200,)
# ============= Reaction Added Events ============= #
# If the user has added an emoji reaction to the onboarding message
elif event_type == "reaction_added":
user_id = slack_event["event"]["user"]
# Update the onboarding message
pyBot.update_emoji(team_id, user_id)
return make_response("Welcome message updates with reactji", 200,)
# =============== Pin Added Events ================ #
# If the user has added an emoji reaction to the onboarding message
elif event_type == "pin_added":
user_id = slack_event["event"]["user"]
# Update the onboarding message
pyBot.update_pin(team_id, user_id)
return make_response("Welcome message updates with pin", 200,)
# ============= Event Type Not Found! ============= #
# If the event_type does not have a handler
message = "You have not added an event handler for the %s" % event_type
# Return a helpful error message
return make_response(message, 200, {"X-Slack-No-Retry": 1})
def _event_handler(event_type, slack_event):
"""
A helper function that routes events from Slack to our Bot
by event type and subtype.
Parameters
----------
event_type : str
type of event recieved from Slack
slack_event : dict
JSON response from a Slack reaction event
Returns
----------
obj
Response object with 200 - ok or 500 - No Event Handler error"""
team_id = slack_event["team_id"]
pyBot.find_team(team_id)
if event_type == "message":
sender_id = None
received_message = None
if "user" in slack_event["event"]:
sender_id = slack_event["event"]["user"]
received_message = slack_event["event"]["text"].encode('utf-8')
if received_message == 'Recommencer':
sr.reset_search(sender_id)
adapted_message = sr.adapt_message_to_wit(sender_id, slack_event["event"]["text"].encode('utf-8'))
message = wit.treatment(adapted_message, sender_id)
channel = slack_event["event"]["channel"]
print "SLACK DEBUG \n"
print message
pyBot.send_message(sender_id, channel, message)
return HttpResponse("OK", 200)
# ============= Event Type Not Found! ============= #
# If the event_type does not have a handler
#message = "You have not added an event handler for the %s" % event_type
# Return a helpful error message
#channel = slack_event["event"]["channel"]
#if "user" in slack_event["event"]:
# pyBot.send_message(channel, message)
return HttpResponse("OK", 200)
def hears(request):
"""
This route listens for incoming events from Slack and uses the event
handler helper function to route events to our Bot.
"""
#Wit makes our responses timeout, so we ignore Slack retries
if "HTTP_X_SLACK_RETRY_NUM" in request.META:
return HttpResponse("OK", 200)
slack_event = json.loads(request.body)
# ============= Slack URL Verification ============ #
# In order to verify the url of our endpoint, Slack will send a challenge
# token in a request and check for this token in the response our endpoint
# sends back.
# For more info: https://api.slack.com/events/url_verification
if "challenge" in slack_event:
return HttpResponse(slack_event["challenge"], 200)
#removed {"content_type":"application/json"} from flask response
# ============ Slack Token Verification =========== #
# We can verify the request is coming from Slack by checking that the
# verification token in the request matches our app's settings
if pyBot.verification != slack_event.get("token"):
print "Invalid Slack verification token: %s \npyBot has: \
%s\n\n" % (slack_event["token"], pyBot.verification)
# By adding "X-Slack-No-Retry" : 1 to our response headers, we turn off
# Slack's automatic retries during development.
return HttpResponse(message, 403)
# ====== Process Incoming Events from Slack ======= #
# If the incoming request is an Event we've subcribed to
if "event" in slack_event:
event_type = slack_event["event"]["type"]
# Then handle the event by event_type and have your bot respond
return _event_handler(event_type, slack_event)
# If our bot hears things that are not events we've subscribed to,
# send a quirky but helpful error response
return HttpResponse("[NO EVENT IN SLACK REQUEST] These are not the droids\
you're looking for.", 404)
