Python bot 模块,Bot() 实例源码

我们从Python开源项目中,提取了以下25个代码示例,用于说明如何使用bot.Bot()

项目:poor-webhook    作者:ThreatResponse    | 项目源码 | 文件源码
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
项目:WineHelper    作者:scorelabio    | 项目源码 | 文件源码
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!')
项目:instagramBot    作者:mcclane    | 项目源码 | 文件源码
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)
项目:instagramBot    作者:mcclane    | 项目源码 | 文件源码
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)
项目:demo.slackbot    作者:SynapseFI    | 项目源码 | 文件源码
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
项目:dev-wine-helper    作者:WineHelperEnseirb    | 项目源码 | 文件源码
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!')
项目:sketal    作者:vk-brain    | 项目源码 | 文件源码
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)
项目:chromecastslack    作者:sh0oki    | 项目源码 | 文件源码
def main():
    m = ChromecastManager(bot.Bot())
    while True:
        m.poll()
        sleep(POLL_INTERVAL)
项目:WineHelper    作者:scorelabio    | 项目源码 | 文件源码
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})
项目:AlphaBot    作者:PokemonAlpha    | 项目源码 | 文件源码
def main():
    logger.info('Alpha Bot v1.1')
    config = init_config()
    setup_logging(config)
    bot = Bot(config)
    bot.start()
项目:CoolesSpiel    作者:AlinaGri    | 项目源码 | 文件源码
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)
项目:CoolesSpiel    作者:AlinaGri    | 项目源码 | 文件源码
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)
项目:instagramBot    作者:mcclane    | 项目源码 | 文件源码
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)
项目:instagramBot    作者:mcclane    | 项目源码 | 文件源码
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)
项目:instagramBot    作者:mcclane    | 项目源码 | 文件源码
def mass_unfollow(username, password, n):
    b = Bot(username, password)
    b.mass_unfollow(n)
项目:stalk-the-DMV    作者:thisisandreeeee    | 项目源码 | 文件源码
def __init__(self):
        self.db = DB()
        self.logger = Logger()
        self.bot = Bot()
项目:Slack-Python-Onboarding-Tutorial    作者:slackapi    | 项目源码 | 文件源码
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)
项目:Slack-Python-Onboarding-Tutorial    作者:slackapi    | 项目源码 | 文件源码
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})
项目:dev-wine-helper    作者:WineHelperEnseirb    | 项目源码 | 文件源码
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})
项目:botfriend    作者:leonardr    | 项目源码 | 文件源码
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)
项目:WineHelper    作者:scorelabio    | 项目源码 | 文件源码
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)
项目:WineHelper    作者:scorelabio    | 项目源码 | 文件源码
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)
项目:Slack-Python-Onboarding-Tutorial    作者:slackapi    | 项目源码 | 文件源码
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})
项目:dev-wine-helper    作者:WineHelperEnseirb    | 项目源码 | 文件源码
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)
项目:dev-wine-helper    作者:WineHelperEnseirb    | 项目源码 | 文件源码
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)