Python telepot 模块，Bot() 实例源码

def __init__(self, api_key, mods_, whitelist, overseer):
        self.api_key = api_key
        self.bot = telepot.Bot(self.api_key)
        self.overseer = overseer
        logging.debug(self.bot.getMe())
        self.reactiontime = 10 # messages older than this won't be answered

        self.mods=[]
        for m in mods_:
            exec("import mods.{}".format(m)) # totally insecure
            logging.debug("exec mod=mods.{0}.{0}(self)".format(m))
            exec("mod=mods.{0}.{0}(self)".format(m))
            self.mods.append(locals()['mod'])
            # if feedback from the mod is needed
            # start a thread here, e.g.
            # thread.start_new_thread(self.modHandling,(mod,))
            # to use this, use queue_out in each mod
        import ast #json would require strings as keys
        self.whitelist=ast.literal_eval(whitelist)


        self.bot.message_loop(self.recv)

def __init__(self, token, **kwargs):
        '''
        Create a new TelegramNotifier

        :param token: Your Bot's authentication token
        :param recipients: The chatIDs or usernames that should
                           be contacted by this Notifier

        :type recipients: Iterable of recipients or dict mapping categories to recipients
        :param categories: The message categories this Notifier should relay
        :type categories: Iterable
        :param level: The minimum level for messages to be relayed
        :type level: int
        '''
        if _has_telepot is False:
            raise ImportError('The telepot package is required for this Notifier')
        self.bot = telepot.Bot(token)
        super().__init__(**kwargs)

def start(self):
        try:
            self.torrent_client = transmissionrpc.Client(
                self.config.get('transmission', 'rpc_host'),
                port=self.config.get('transmission', 'rpc_port'),
                user=self.config['transmission']['rpc_user'],
                password=self.config['transmission']['rpc_password'])
        except transmissionrpc.error.TransmissionError:
            print("ERROR: Couldn't connect to Transmission. Check rpc configuration.")
            sys.exit()

        try:
            self.bot = telepot.Bot(self.config.get('telegram', 'bot_token'))
            self.bot.setWebhook('') # Remove existent webhook
            self.bot.message_loop(
                self.handle_mesasge,
                relax=self.config.getint('telegram', 'polling_interval'))
        except telepot.exception.TelegramError:
            print("ERROR: Couldn't start Telegram bot. Check telegram bot configuration.")
            sys.exit()

def help(self):
        bot.sendMessage(self._userid, "??? kC Online Judge Bot?\n"
                                     "???? KCOJ Bot?????? [" + NAME + "]\n"
                                     "?????????????????????? Online Judge\n"
                                     "??[OJ ???](" + URL + ")\n"
                                     "????????\n\n"
                                     "???????????????\n"
                                     "1. ??????????????????????????\n"
                                     "2. ???????????????????????\n"
                                     "3. ????????????????????? *??????*\n"
                                     "4. *??????* ?????? *??????* ??\n"
                                     "5. ???????????????\n"
                                     "6. ??? Online Judge ???????????????????\n"
                                     "7. ?????????????????????????\n\n"
                                     "????? *MIT License*\n"
                                     "?????? @PinLin\n"
                                     "??????? GitHub??????????????????> </\n"
                                     "?????\n"
                                     "[https://github.com/PinLin/kcoj_bot]\n\n",
                                     parse_mode='Markdown')

def main():

    global bot

    startupThread = threading.Thread(target=startUpWork, args=())
    startupThread.daemon = True
    startupThread.start()

    if internetOn():
        try:
            bot = telepot.Bot(access_token)
            bot.message_loop(handle)
            bot.sendMessage(CHAT_ID, str(current_system_time.strftime("%d|%m|%Y-%H|%M|%S")) + ': '+ USER + ' is online ^_^')
            print ('[*] Listening ...')
            try:
                while 1:
                    time.sleep(5)
            except KeyboardInterrupt:
                print '[*] Eclipse completed...'
        except Exception as e:
            pass

def test_katbot_main_file():
    """Test argument parsing and calling."""
    from katcr.bot import main

    from unittest.mock import patch
    import os
    import tempfile

    with tempfile.NamedTemporaryFile(mode='w', delete=False) as tmpfile:
        tmpfile.write('foobar')

    opts = {'--token': None, '--shortener': None,
            '--token-file': tmpfile.name}
    with patch('katcr.bot.docopt', side_effect=(opts,)):
        with patch('katcr.bot.MessageLoop'):
            with patch('katcr.bot.telepot.Bot.__init__') as mock:
                main()
                mock.assert_called_with('foobar')
    os.unlink(tmpfile.name)

def __init__(self, seed_tuple, timeout, is_main=True, config=None):
        super(PocketBot, self).__init__(seed_tuple, timeout)
        self.visit_time = 0
        self.find_url = re.compile('(https?://[^\s]+)')
        self.find_tag = re.compile("#[\w']+")
        self.mongo = db.PockebotDBSlave(config)
        self.is_main = is_main
        self.messages = list()
        self.pocket_client = pocket.PocketbotPocketSlave(config['pocket_token'])
        # For forwarding /feedback messages
        self.stuff_bot = telepot.Bot(config['support_bot_token'])
        self.dev_group_chat = config['developers_group']
        # User state
        self.lang = 'en'
        self.tags_promt = True
        self.set_pockebot_tag = True
        self.session_cnt = 0
        self.audio = False
        self.request_chat_id = None
        self._edit_msg_ident = None
        self._editor = None
        self.waiting_for_menu_action = False
        self.reading_time_tag = True

def main(config, file_with_msg):
    with open(file_with_msg) as f:
        text_to_send = f.read()
    bot = telepot.Bot(config['telegram_token'])
    mongo = pymongo.MongoClient(host=config['mongo']['host'])
    people = mongo[config['mongo']['db']]['people']
    already_sent = mongo[config['mongo']['db']]['people_sent']
    cursor = people.find({})
    count = 0
    for record in cursor:
        print('>> {}'.format(count))
        print(record)
        if already_sent.find_one({'id': record['id']}) is None:
            already_sent.insert_one({'id': record['id']})
            send(bot, record['id'], text_to_send)
        count += 1

def do_settings():
    global radius_by_default,radius_step,radius_max,messages, POKEMONS, POKEMON_NUM,about,bot,log_queue,log_to_file,ignored_default,log_notifications,max_notis_per_user_and_cycle,time_between_cycles
    with open('{}/res/telebotsettings.json'.format(workdir)) as f:
        telecfg = json.load(f)

    language = telecfg["language"]
    radius_by_default = int(telecfg["radius_by_default"])
    radius_step = telecfg["radius_step"]
    radius_max = telecfg["radius_max"]
    max_notis_per_user_and_cycle = telecfg["max_notis_per_user_and_cycle"]
    time_between_cycles = telecfg["time_between_cycles"]
    log_notifications = telecfg["log_notifications"]

    log_to_file = telecfg["log_to_file"]

    bot = telepot.Bot(str(telecfg["TELEGRAM_BOT_TOKEN"]))
    about = telecfg["info_about"]

    ignored_default = telecfg["ignored_by_default"]
    if type(ignored_default) == unicode:
        print(time.strftime('[%H:%M:%S] ' + '[!] Warning, the ignored_by_default setting in telebotsettings.json should be now a array like [1,2,3] instead of a string like "1,2,3"'))
        sys.exit()

    with open('{}/res/languages/{}.json'.format(workdir, language)) as f:
        messages = json.load(f)

    with open('{}/webres/static/{}.json'.format(workdir, language)) as f:
        POKEMONS = json.load(f)

def send_message(chat_id, text, disable_notification=False, reply_markup=None, disable_web_page_preview=False):
    try:
        if reply_markup is None:
            bot.sendMessage(chat_id, text, parse_mode='HTML', disable_notification=disable_notification, disable_web_page_preview=disable_web_page_preview)
        else:
            bot.sendMessage(chat_id, text, parse_mode='HTML', disable_notification=disable_notification, disable_web_page_preview=disable_web_page_preview, reply_markup=reply_markup)
    except BotWasBlockedError:
        print_log("[!] Bot was blocked. Couldn't send message.")
    except TelegramError as e:
        print_log("[!] An error happened while sending message " + str(e.json))
    except Exception as e:
        print_log("[!] An unkown error happened while sending message, error: {}".format(e))

def __init__(self, config_file):

        print('[*] Getting config...')
        self.conf = dict()
        self.get_conf(config_file)
        print('[+] Done!\n')

        # Check if CEF_Syslog is enabled
        if self.conf['cef_syslog_enable']:
            print '[+] Syslog Enabled'
            self.syslog = syslog.Syslog(self.conf['cef_syslog_server'])

        # Check if Telegram is enabled
        if self.conf['telegram_enable']:
            print '[+] Telegram Enabled'
            self.bot = telepot.Bot(self.conf['api'])

        # Check if Zabbix is enabled
        if self.conf['zabbix_enable']:
            print '[+] Zabbix Enabled'
            print 'Notifications ',self.conf['notifications']
            print 'Advanced ',self.conf['zabbix_advantage_keys']
        print 

        print('[*] Getting rules...')
        self.get_file_rules()
        print('[+] Done!\n')

        self.rules = json.loads(open(self.conf['rules']).read())

        # List of all senders, enabled or not
        self.senders = [self.send_zabbix, self.send_cef_syslog, self.send_telegram]


        print('[*] A.R.T.L.A.S Started!\n')

def connect(self):
        self.__client = telepot.Bot(self.__bot_token)

    # Sends a start up message on Telegram

def message_handler_thread_work(self):
        self.bot_handle=telepot.Bot(self.token)
        bot_bind_ok=False
        activation_fail_announce=False
        while bot_bind_ok==False:
            try:
                self.bot_handle.getMe()[u"username"]
                bot_bind_ok=True
            except:
                if activation_fail_announce==False:
                    report("w","<"+self.allowed_user+"> "+"Message handler activation error. Will keep trying...")
                    activation_fail_announce=True
                time.sleep(LISTENER_SERVICE_THREAD_HEARTBEAT_SECONDS)

        self.listener.consume_user_messages(self.allowed_user)
        report("w","<"+self.allowed_user+"> "+"Message handler for user \""+self.allowed_user+"\" is now active.")

        while self.keep_running.is_set()==True:
            time.sleep(USER_MESSAGE_HANDLER_THREAD_HEARTBEAT_SECONDS)
            self.check_tasks()
            if self.listen_flag.is_set()==True:
                new_messages=self.listener.consume_user_messages(self.allowed_user)
                total_new_messages=len(new_messages)
                if total_new_messages>0:
                    report("w","<"+self.allowed_user+"> "+str(total_new_messages)+" new message(s) received.")
                    self.process_messages(new_messages)

        report("w","<"+self.allowed_user+"> "+"Message handler exited.")
        self.bot_has_exited.set()
        return

def __get_telepot_instance():
        """
        Gets an instance of Telepot bot.
        :return:
        """
        try:
            return telepot.Bot(BOT_TOKEN)
        except Exception as ex:
            ExceptionHandler.handle_exception(ex, True)

def main():
    # LOG setup
    os.chdir(os.path.join(os.path.abspath(os.sep), 'home', OS_USER, 'SmartSlam', 'ManagementTelegramBOT'))
    log_dir = os.path.join(os.getcwd(), 'LOG')
    if not os.path.exists(log_dir):
        os.makedirs(log_dir)
    logging.basicConfig(filename=os.path.join(log_dir, 'INFO.log'),
                        level=logging.INFO,
                        format='%(asctime)-15s '
                               '%(levelname)s '
                               '--%(filename)s-- '
                               '%(message)s')

    # BOT
    logging.info('Setting up Bot...')

    wake_up_msg = wake_up()
    for client in data['known_clients']:
        bot.sendMessage(client, wake_up_msg)

    for sig in [signal.SIGTERM, signal.SIGINT, signal.SIGHUP, signal.SIGQUIT]:
        signal.signal(sig, signal_handler)

    # STARTING WAITING CYCLE
    logging.info('STARTING LISTENING loop')

    MessageLoop(bot, {'chat': handle,
                      'callback_query': on_callback}).run_as_thread()
    while 1:
        time.sleep(10)

def send_message(self, id, text):
        """
            Send the message text to recipient with id recipient.
        """
        print ('Sending Mssg',text)
        if sys.version_info >= (3, 0):
            message = text
        else:
            message = text.decode('unicode_escape')
        token = self.bot_token
        bot = telepot.Bot(token)
        r = bot.sendMessage(id,text)
        print (r)

def __init__(self, t_channel, config):
        super(Reddit2TelegramSender, self).__init__()
        self.config = config
        self.telepot_bot = telepot.Bot(self.config['telegram_token'])
        self.t_channel = t_channel
        self._make_mongo_connections()
        self._store_stats()

def telegram_status(status, image_path=None):
    try:
        bot = telepot.Bot(settings.TELEGRAM_TOKEN)
        if image_path:
            bot.sendPhoto(settings.TELEGRAM_CHANNEL, open(image_path, "rb"))
        bot.sendMessage(settings.TELEGRAM_CHANNEL, status)
    except telepot.exception.TelepotException as ex:
        raise MeteoSangueException(ex)

def connect(self):
        self.client = telepot.Bot(self.bot_token) 

    #Set the appropriate settings for each alert

def __init__(self):
        global args
        self.config_file = args.config

        with open(self.config_file) as config:
            self._bot_config = json.loads(config.read(), encoding="utf-8")
            self._telebot_token = self._bot_config["token"]
            self._admin = self._bot_config["admin"]
            self._tuling_token = self._bot_config["tuling"]
        self._bot = telepot.Bot(self._telebot_token)
        self._me = self._bot.getMe()
        del self._bot # Won't use it in the future

        self.switch = defaultdict(lambda: True)

def text_message(msg_recieved, sender_id , msg):
    print sender_id, msg_recieved
    msg_list = msg_recieved.split(' ')

    send_msg = "Bot is Under Maintenance ! Try After Some Time. \n Contact Botmaster At http://telegram.me/dhruvramdev"
    bot.sendMessage(sender_id, send_msg)

def test_katbot_isinstance():
    """Test katbot is a bot instance."""
    from telepot import Bot
    from katcr.bot import KATBot
    assert issubclass(KATBot, Bot)
    opts = {'--token': "foo", '--shortener': ''}
    assert isinstance(KATBot(opts), Bot)

def main(api_key):
    global bot
    bot = telepot.Bot(api_key)
    bot.message_loop(handle)
    print('Listening ...')
    while True:
        time.sleep(10)

def listen_thread_work(self):
        self.bot_handle=telepot.Bot(self.token)
        last_check_status=False
        bot_bind_ok=False
        activation_fail_announce=False

        while bot_bind_ok==False:
            try:
                self.name=self.bot_handle.getMe()[u"username"]
                bot_bind_ok=True
            except:
                if activation_fail_announce==False:
                    report("l","Listener service activation error. Will keep trying...")
                    activation_fail_announce=True
                time.sleep(LISTENER_SERVICE_THREAD_HEARTBEAT_SECONDS)

        self.catch_up_IDs()
        os.system("@title FileBot v"+__version__+" - "+self.name)
        set_general_data("bot_name",self.name)
        report("l","Listener service for bot \""+self.name+"\" is now active.")
        self.is_active.set()

        while self.keep_running.is_set()==True:
            time.sleep(LISTENER_SERVICE_THREAD_HEARTBEAT_SECONDS)
            response=[]

            try:
                response=self.bot_handle.getUpdates(offset=self.last_ID_checked+1)
                check_status=True
            except:
                check_status=False

            if check_status!=last_check_status:
                last_check_status=check_status
                if check_status==True:
                    set_general_data("online","yes")
                    report("l","Message retrieval is now online.")
                else:
                    set_general_data("online","no")
                    report("l","Stopped being able to retrieve messages.")

            self.organize_messages(response)

        self.is_active.clear()
        report("l","Listener has exited.")
        self.has_exited.set()
        return

def screen_init(self):
        global COLOR_FRAME_TX
        global REPORT_X
        global REPORT_Y
        global REPORT_WIDTH
        global REPORT_HEIGHT
        global COLOR_WINDOW_BG
        global COLOR_REPORT_TX
        global COLOR_REPORT_BG
        global WINDOW_COLS
        global COMMAND_BOX_TITLE
        global COLOR_COMMAND_TITLE_TX
        global REPORT_WIDTH_RANGE
        global TITLEBAR_ONLINESTATUS_X
        global TITLEBAR_TIMEDIFF_X
        global TITLEBAR_BOTNAME_X
        global COLOR_TITLEBAR_TX
        global COLOR_TITLEBAR_BG

        title="FileBot v"+__version__+" by Searinox Navras"
        self.out_pos_color_txt(WINDOW_COLS/2-len(title)/2,0,14,COLOR_WINDOW_BG,title)
        self.out_pos_color_txt(TITLEBAR_BOTNAME_X,TITLEBAR_Y,COLOR_TITLEBAR_TX,COLOR_WINDOW_BG,"Bot name: <not retrieved>")
        self.out_pos_color_txt(TITLEBAR_TIMEDIFF_X,TITLEBAR_Y,COLOR_TITLEBAR_TX,COLOR_WINDOW_BG,"Local clock bias(seconds): UNKNOWN")
        self.out_pos_color_txt(TITLEBAR_ONLINESTATUS_X,TITLEBAR_Y,COLOR_TITLEBAR_TX,COLOR_WINDOW_BG,"Status: NOT STARTED")

        for i in REPORT_WIDTH_RANGE:
            self.screenbuffer_current[flat_coords(REPORT_X+i,REPORT_Y-1)]["col_tx"]=14
            self.screenbuffer_current[flat_coords(REPORT_X+i,REPORT_Y-1)]["char"]=205
            self.screenbuffer_current[flat_coords(REPORT_X+i,REPORT_Y+REPORT_HEIGHT)]["col_tx"]=14
            self.screenbuffer_current[flat_coords(REPORT_X+i,REPORT_Y+REPORT_HEIGHT)]["char"]=205

        for i in range(REPORT_HEIGHT):
            self.screenbuffer_current[flat_coords(REPORT_X-1,REPORT_Y+i)]["col_tx"]=14
            self.screenbuffer_current[flat_coords(REPORT_X-1,REPORT_Y+i)]["char"]=186
            self.screenbuffer_current[flat_coords(REPORT_X+REPORT_WIDTH,REPORT_Y+i)]["col_tx"]=14
            self.screenbuffer_current[flat_coords(REPORT_X+REPORT_WIDTH,REPORT_Y+i)]["char"]=186

        self.screenbuffer_current[flat_coords(REPORT_X-1,REPORT_Y-1)]["col_tx"]=14
        self.screenbuffer_current[flat_coords(REPORT_X-1,REPORT_Y-1)]["char"]=201

        self.screenbuffer_current[flat_coords(REPORT_X+REPORT_WIDTH,REPORT_Y-1)]["col_tx"]=14
        self.screenbuffer_current[flat_coords(REPORT_X+REPORT_WIDTH,REPORT_Y-1)]["char"]=187

        self.screenbuffer_current[flat_coords(REPORT_X-1,REPORT_Y+REPORT_HEIGHT)]["col_tx"]=14
        self.screenbuffer_current[flat_coords(REPORT_X-1,REPORT_Y+REPORT_HEIGHT)]["char"]=200

        self.screenbuffer_current[flat_coords(REPORT_X+REPORT_WIDTH,REPORT_Y+REPORT_HEIGHT)]["col_tx"]=14
        self.screenbuffer_current[flat_coords(REPORT_X+REPORT_WIDTH,REPORT_Y+REPORT_HEIGHT)]["char"]=188

        for i in REPORT_WIDTH_RANGE:
            for j in range(REPORT_HEIGHT):
                self.screenbuffer_current[flat_coords(REPORT_X+i,REPORT_Y+j)]["col_tx"]=COLOR_REPORT_TX
                self.screenbuffer_current[flat_coords(REPORT_X+i,REPORT_Y+j)]["col_bg"]=COLOR_REPORT_BG
                self.screenbuffer_current[flat_coords(REPORT_X+i,REPORT_Y+j)]["char"]=32

        center_text=int(WINDOW_COLS/2-len(COMMAND_BOX_TITLE)/2)
        self.out_pos_color_txt(center_text,INPUT_FIELD_Y-1,COLOR_COMMAND_TITLE_TX,COLOR_WINDOW_BG,COMMAND_BOX_TITLE)

        return

def screen_prepare(self):
        global REPORT_OUTPUT_REFRESH_INTERVAL_SECONDS
        global LOCK_SCREEN_REPORT
        global SCREEN_REPORT_EDITED
        global TITLEBAR_Y
        global COLOR_TITLEBAR_TX
        global COLOR_WINDOW_BG
        global TITLEBAR_BOTNAME_X
        global TITLEBAR_TIMEDIFF_X
        global TITLEBAR_ONLINESTATUS_X

        got_input=self.active_console.get_console_input()["input"]

        if self.last_input_field!=got_input:
            self.last_input_field=got_input
            self.recompile_input(got_input)

        if OS_uptime()-self.last_report_refresh_time>=REPORT_OUTPUT_REFRESH_INTERVAL_SECONDS:
            LOCK_SCREEN_REPORT.acquire()
            get_screen_report=SCREEN_REPORT_LINES
            get_report_edited=SCREEN_REPORT_EDITED
            SCREEN_REPORT_EDITED=False
            LOCK_SCREEN_REPORT.release()

            if get_report_edited==True:
                self.last_report_lines=get_screen_report
                self.recompile_report(get_screen_report)

            self.last_report_refresh_time=OS_uptime()

        general_data=get_new_general_data()
        for data_type in general_data:
            value=general_data[data_type]
            if data_type=="bot_name":
                self.out_pos_color_txt(TITLEBAR_BOTNAME_X+len("Bot name: "),TITLEBAR_Y,11,COLOR_WINDOW_BG,value+" "*(32-len(value)))
            elif data_type=="time_difference":
                time_difference=float(value[1:])
                time_color=10
                if time_difference>=30:
                    time_color=14
                if time_difference>=60:
                    time_color=12
                self.out_pos_color_txt(TITLEBAR_TIMEDIFF_X+len("Local clock bias(seconds): "),TITLEBAR_Y,time_color,COLOR_WINDOW_BG,value+" "*(10-len(value)))
            elif data_type=="online":
                if value=="yes":
                    out="ONLINE"
                    colorstate=10
                else:
                    out="OFFLINE"
                    colorstate=12
                self.out_pos_color_txt(TITLEBAR_ONLINESTATUS_X+len("Status: "),TITLEBAR_Y,colorstate,COLOR_WINDOW_BG,out+" "*(11-len(out)))

        return