Telegram 搜索机器人BOT

一，小伙伴么，当你有的时候玩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相关推荐

telegram 创建机器人 bot 简介
今天看到一个电报群里说利用BOT来签到,很好奇就想学学,来吧 ! 创建一个BOT!!! 1.先搜索BotFather 我这里搜索到好几个没有一个可用的,输入/help 返回俄文错误信息. 后来在官网 ...
telegram怎么搜索机器人_telegram机器人
telegram是国外常用的聊天功能,功能很强大,除了聊天还可以玩游戏,传文件,视频,声音,投票,群组.当然,这些跟qq,微信都很类似了.但是它毕竟是国外最常用的IM之一.telegram有个聊天机器 ...
机器人坐标系建立_如何在30分钟内建立一个简单的搜索机器人
机器人坐标系建立 by Quinn Langille 奎因·兰吉尔(Quinn Langille) 如何在30分钟内建立一个简单的搜索机器人 (How to Build A Simple Search ...
微信公众号与小程序（二十——关于我如何莫名其妙建了一个电影搜索机器人这件事）
没有小程序了购买小程序网课后的确奋发学了一阵子然后有一天我想了一下做这个的前景,就算我真搞出一个壁纸小程序,又有几个人会关注呢?有利润吗?有价值吗? 费劲看微信阅读文档,整半天也理解不了哎,搞不 ...
抢鞋软件bot服务器系统,抢鞋机器人bot app
抢鞋机器人bot是一款抢购鞋子的神器,大家都知道买一双耐克鞋非常的不容易,不仅要预约购鞋资格,还要去抢单,但是有了抢鞋机器人bot大家就在也不用为抢不到鞋子而发愁了,各种高端鞋子帮你秒杀抢购! 抢鞋机 ...
抢鞋软件bot服务器系统,抢鞋机器人bot安卓版
抢鞋机器人bot在这可以利用机器人帮助自己强到想要购买的运动鞋,在这里会有不同品牌的鞋子是可以去购买的,在这里会有详细的操作说明大家是能够通过上面的操作说明学会如何完成自己的购买操作,在这里是买鞋是会 ...
Telegram 群组创建机器人bot
文章目录创建机器人获取群组chatId 发送消息 Telegram是一款跨平台的即时通讯软件,用户可以相互交换加密与自毁消息,发送照片.影片等所有类型文件. 创建机器人 Telegram软件中直接 ...
三款telegram中群搜索机器人推荐
Telegram(简称TG或电报)是跨平台的即时通信软件,其客户端是自由及开放源代码软件,但服务器端是专有软件.用户可以相互交换加密与自毁消息(类似于"阅后即焚"),发送照片.影片 ...
telegram 创建机器人
目录第一步第二步第三步第四步第五步第一步登录Telegram,并找到@BotFather,使用/new bot指令,并按照指示为你的Bot取个昵称以及用户名(Bot的用户名必需以bot结 ...

Telegram 搜索机器人BOT

Telegram 搜索机器人BOT相关推荐

最新文章

热门文章