我们从Python开源项目中,提取了以下25个代码示例,用于说明如何使用bot.Bot()。
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
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 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 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)