?一，小伙伴么，當你有的時候玩TG，你自己建一個群或者頻道，有很多人的時候，會找不到聊天記錄，或者文件，那么你就可以用這個機器人來幫你實現。

? ? ?1，通過關鍵詞來檢索你想要的內容。這個時候你就會想打造屬于你的 telegramBot，直接發送你要搜索的關鍵字，就可以定位到你想要的內容，搜索支持 Lucene 語法。

? ? ?2， 機器人同時可以為群組、頻道、個人提供聊天記錄搜索服務。

? ? ?3， 工作原理是使用 Telegram Client Api 獲取頻道內所有信息，并持續監聽新信息。將所有信息歸檔Elasticsearch 搜索引擎，用戶可以在 Bot 前端執行搜索。

二、那么如何搭建呢？

前提條件

申請 Telegram MTProto API ID：?https://my.telegram.org/app

申請 Telegram Bot ID：@BotFather

準備一個 Telegram 賬號

安裝 Python3：???????Download Python | Python.org

下載源代碼：

2安裝依賴：?pip install -r requirements.txt

修改 main.py 中的配置或使用環境變量

• API_ID：Telegram MTProto API ID
• API_HASH：Telegram MTProto API ID
• BOT_TOKEN：從 BotFather 獲取的 bot token
• CHAT_ID：你要搜索的 chat 的 ID，可以使用?@getidsbot?獲取。
• ADMIN_ID：管理員的 ID，可以使用?@getidsbot?獲取。
• 先創建一個?session?文件夾（mkdir session），運行?python main.py?提示輸入手機號和驗證碼即可，session文件夾里面會生成幾個數據庫文件。
• 部署
• 把 session 文件夾和源碼部署到服務器。
• 修改 docker-compose.yml 中的環境變量
• 使用 docker-compose 部署：docker-compose up -d
• 啟動完成后用管理員的賬號（之前配置的 ADMIN_ID）向 Bot 發送命令?/download_history?下載歷史記錄。
• 源代碼如下。
• from telethon import TelegramClient, events, Button import socks import asyncio import html import osREDIS_HOST = "REDIS_HOST" in os.environ and os.environ["REDIS_HOST"] or 'localhost' REDIS_PORT = "REDIS_PORT" in os.environ and os.environ["REDIS_PORT"] or 6379 ELASTIC_URL = "ELASTIC_URL" in os.environ and os.environ["ELASTIC_URL"] or 'http://localhost:9200/' API_ID = "API_ID" in os.environ and os.environ["API_ID"] or 123456 API_HASH = "API_HASH" in os.environ and os.environ["API_HASH"] or 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxx' BOT_TOKEN = "BOT_TOKEN" in os.environ and os.environ["BOT_TOKEN"] or '123456789:xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx' CHAT_ID = "CHAT_ID" in os.environ and os.environ["CHAT_ID"] or '-1001254246410' ADMIN_ID = "ADMIN_ID" in os.environ and os.environ["ADMIN_ID"] or '345796292'from elasticsearch import Elasticsearch es = Elasticsearch([ELASTIC_URL])import redis db = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)# https://docs.telethon.dev/en/latest/basic/signing-in.html api_id = str(API_ID) api_hash = API_HASH bot_token = BOT_TOKEN# proxy = (socks.SOCKS5, '127.0.0.1', 7777) proxy = Nonechat_id = int(CHAT_ID) admin_id = int(ADMIN_ID)welcome_message = ''' 這里是 你 的搜索 Bot，直接發送你要搜索的內容即可。搜索支持 Lucene 語法。 例如： `每日速覽` `+每日速覽 +date:2019-12-25` `+每日速覽 +date:[2019-12-25 TO 2019-12-30]` '''share_id = chat_id < 0 and chat_id * -1 - 1000000000000 or chat_id elastic_index = "chat" + str(chat_id)mapping = { "properties":{ "content": { "type": "text", "analyzer": "ik_max_word", "search_analyzer": "ik_smart" }, "url": { "type": "text" }, "date": { "type": "date" } } }def ensureElasticIndex(index, mapping): if not es.indices.exists(index=elastic_index): es.indices.create(index=elastic_index) es.indices.put_mapping(index=elastic_index, body=mapping)def deleteElasticIndex(index): if es.indices.exists(index=elastic_index): es.indices.delete(index=elastic_index)def search(q, from_, size=10): ensureElasticIndex(index=elastic_index, mapping=mapping) return es.search(index=elastic_index, q=q, df="content", size=10, from_=from_, body={ "highlight" : { "pre_tags" : ["<b>"], "post_tags" : ["</b>"], "fields" : { "content" : { "fragment_size" : 15, "number_of_fragments" : 3, "fragmenter": "span" } } } })def renderRespondText(result, from_): total = result['hits']['total']['value'] respond = '搜素到%d個結果：\n' % (total) for i in range(len(result['hits']['hits'])): hit = result['hits']['hits'][i] content = 'highlight' in hit and hit['highlight']['content'][0] or hit['_source']['content'][0:15]respond += '%d. <a href="%s">%s</a>\n' % (from_ + i + 1, hit['_source']['url'], content) respond += '耗時%.3f秒。' % (result['took'] / 1000) return responddef renderRespondButton(result, from_): total = result['hits']['total']['value'] return [ [ Button.inline('上一頁??', str(max(from_ - 10, 0))), Button.inline('??下一頁', str(min(from_ + 10, total // 10 * 10))), ] ]@events.register(events.NewMessage) async def ClientMessageHandler(event): if event.chat_id == chat_id and event.raw_text and len(event.raw_text.strip()) >= 0: es.index(index=elastic_index, body={"content": html.escape(event.raw_text).replace('\n',' '), "date": int(event.date.timestamp() * 1000), "url": "https://t.me/c/%s/%s" % (share_id, event.id)}, id=event.id)@events.register(events.CallbackQuery) async def BotCallbackHandler(event): if event.data: from_i = int(event.data) q = db.get('msg-' + str(event.message_id) + '-q') if q: result = search(q, from_i) respond = renderRespondText(result, from_i) buttons = renderRespondButton(result, from_i) msg = await event.edit(respond, parse_mode='html', buttons=buttons)await event.answer()async def downloadHistory(): deleteElasticIndex(index=elastic_index) ensureElasticIndex(index=elastic_index, mapping=mapping) async for message in client.iter_messages(chat_id): if message.chat_id == chat_id and message.raw_text and len(message.raw_text.strip()) >= 0: print(message.id) es.index( index=elastic_index, body={"content": html.escape(message.raw_text).replace('\n',' '), "date": int(message.date.timestamp() * 1000), "url": "https://t.me/c/%s/%s" % (share_id, message.id)}, id=message.id )@events.register(events.NewMessage) async def BotMessageHandler(event): if event.raw_text.startswith('/start'): await event.respond(welcome_message, parse_mode='markdown') elif event.raw_text.startswith('/download_history') and event.chat_id == admin_id: # 下載所有歷史記錄 await event.respond('開始下載歷史記錄', parse_mode='markdown') await downloadHistory() await event.respond('下載完成', parse_mode='markdown') else: from_i = 0 q = event.raw_text result = search(q, from_i) respond = renderRespondText(result, from_i) buttons = renderRespondButton(result, from_i) msg = await event.respond(respond, parse_mode='html', buttons=buttons)db.set('msg-' + str(msg.id) + '-q', q)loop = asyncio.get_event_loop()client = TelegramClient('session/client', api_id, api_hash, connection_retries=None, proxy=proxy, loop=loop) client.add_event_handler(ClientMessageHandler) client.start()bot = TelegramClient('session/bot', api_id, api_hash, connection_retries=None, proxy=proxy, loop=loop) bot.add_event_handler(BotMessageHandler) bot.add_event_handler(BotCallbackHandler) bot.start(bot_token=bot_token)try: loop.run_forever() except KeyboardInterrupt: pass

總結

以上是生活随笔為你收集整理的Telegram 搜索机器人BOT的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。