Telegram消息通过企业微信机器人发送警告通知的功能

程序的主要功能是从 Telegram 频道中抓取消息数据，并将其保存到 MongoDB 数据库中，同时通过 Redis 来存储和管理消息的偏移数据，用于跟踪消息的处理进度。它还包含将消息通过企业微信机器人发送警告通知的功能。

1. 具体功能总结：

1.1. MongoDB 数据库连接与更新

程序使用 pymongo 连接 MongoDB 数据库，提供了 update 方法来将抓取到的 Telegram 消息数据保存到 MongoDB 中，采用 upsert 模式（如果消息已存在则更新，否则插入）。

1.2. Redis 用于存储偏移信息

通过 Redis 来记录处理的 Telegram 消息的偏移量，确保从上次处理过的消息开始获取历史记录，避免重复处理消息。

1.3. Telegram 客户端配置与消息抓取

使用 telethon 的 TelegramClient 来连接 Telegram API，并通过异步的 iter_messages 方法抓取指定频道的消息。通过 format_message 方法将消息格式化为需要的结构化数据（包括消息文本、发送者信息、群组信息等），并返回一个字典格式的数据。抓取的消息按照消息 ID 或者日期进行排序，以获取指定的历史记录。

1.4. 消息保存与处理

消息经过格式化后，会通过 mongo_cli.update() 保存到 MongoDB 中。同时，使用 redis_cli.set 将当前处理的消息 ID 存入 Redis，以便下次继续处理新的消息。

1.5. 通过企业微信机器人发送通知

robot_warning 函数用于通过企业微信机器人发送通知。在某些场景下，抓取到的消息数据会被发送到企业微信中，作为系统的警告通知功能。

1.6. 定时任务调度

使用 apscheduler 的 AsyncIOScheduler 设置一个每隔 5 分钟运行一次的任务，调用 load_history_to_save 函数来抓取新的消息并进行处理。使用 next_run_time=datetime.datetime.now() 来立即启动第一个任务执行。

1.7. 异步加载 Telegram 历史消息

load_history_to_save 函数会遍历所有对话（群组或频道），通过异步的方式抓取历史消息并存储到 MongoDB 中，间隔 10 秒处理每个群组。使用 iter_dialogs 方法获取所有对话，依次处理每个群组的历史消息。

2. 总结：

该程序主要用于从 Telegram 群组或频道中抓取历史消息，并通过 MongoDB 和 Redis 进行数据存储与管理，结合定时调度任务自动获取最新的消息。同时，提供了一个企业微信通知机制来发送消息抓取的结果。

  1import asyncio
  2import datetime
  3import json
  4
  5import pymongo
  6import requests
  7from apscheduler.schedulers.asyncio import AsyncIOScheduler
  8from redis import Redis
  9from telethon import TelegramClient
 10# https://my.telegram.org/apps
 11from telethon.utils import get_display_name
 12
 13# 数据库基本信息
 14db_configs = {
 15    'type': 'mongo',
 16    'host': '127.0.0.1',
 17    'port': '27017',
 18    "user": "",
 19    "password": "",
 20    'db_name': 'new_spider'
 21}
 22
 23
 24class Mongo():
 25    def __init__(self):
 26        self.db_name = db_configs.get("db_name")
 27        self.host = db_configs.get("host")
 28        self.port = db_configs.get("port")
 29        self.client = pymongo.MongoClient(f'mongodb://{self.host}:{self.port}', connect=False, maxPoolSize=10)
 30        self.username = db_configs.get("user")
 31        self.password = db_configs.get("passwd")
 32        if self.username and self.password:
 33            self.db = self.client[self.db_name].authenticate(self.username, self.password)
 34        self.db = self.client[self.db_name]
 35
 36    def update(self, item, col="tg_spider"):
 37        if not isinstance(item, list):
 38            item = [item]
 39
 40        for each_item in item:
 41            coll = self.db[col]
 42            coll.update_one({"message_id": each_item["message_id"]}, {'$set': each_item}, upsert=True)
 43
 44
 45redis_cli = Redis.from_url("redis://@localhost:6379")
 46last_offset_date = datetime.datetime.strptime("2022-10-01", "%Y-%m-%d")  # 设置开始时间
 47api_id = xxxx
 48
 49api_hash = xxxxxx
 50
 51channel_dict = {}
 52
 53client = TelegramClient('ztelegrm', api_id, api_hash, connection_retries=15, retry_delay=3)
 54robot_url = xxxxx
 55
 56mongo_cli = Mongo()
 57
 58
 59def robot_warning(text, url):
 60    """
 61    企业微信机器人
 62    text: 文本
 63    url: 链接
 64
 65    """
 66    if isinstance(text, dict) or isinstance(text, list):
 67        text = json.dumps(text, ensure_ascii=False)
 68    headers = {
 69        "Content-Type": "application/json"
 70    }
 71    data = {
 72        "msgtype": "text",
 73        "text": {
 74            "content": text,
 75        }
 76    }
 77    return requests.post(url, headers=headers, data=json.dumps(data))
 78
 79
 80async def format_message(message, chat_id):
 81    if not message.text or message.media:
 82        return None
 83
 84    chat = await message.get_chat()
 85
 86    chat_display_name = get_display_name(chat)
 87    sender_user = await message.get_sender()
 88
 89    content = message.text
 90    talker = chat_id
 91    tg_id = getattr(sender_user, 'username', '')
 92    create_time = message.date.strftime("%s")
 93    group_name = chat_display_name
 94    first_name = getattr(sender_user, 'first_name', '')
 95    last_name = getattr(sender_user, 'last_name', '')
 96    if tg_id is None:
 97        tg_id = "未知"
 98    if last_name is None and first_name is None:
 99        user_name = "未知"
100    else:
101        user_name = f"{first_name}_{last_name}"
102    send_data = f"{content}|$#$#|{talker}|$#$#|{tg_id}|$#$#|{create_time}|$#$#|{group_name}|$#$#|{user_name}"
103    redis_cli.set(f"tg_cache_id_{chat_id}", message.id)
104    # return send_data
105
106    doc_data = {
107        "timestamp": message.date.strftime("%s"),
108        "sender": {
109            "username": getattr(sender_user, 'username', ''),
110            "firstName": getattr(sender_user, 'first_name', ''),
111            "lastName": getattr(sender_user, 'last_name', '')
112        },
113        "channel": chat_display_name,
114        "channel_id": chat_id,
115        "text": message.text,
116        "message_id": message.id
117    }
118    return doc_data
119
120
121async def get_channel_dict():
122    async for d in client.iter_dialogs():
123        channelId = d.entity.id
124        channelName = d.name
125        # channel_dict[channelId] = channelName
126        yield channelId, channelName
127
128
129async def load_history_to_save():
130    channel_dict = get_channel_dict()
131    async for channel_info in channel_dict:
132        channel_id = channel_info[0]
133        min_id = redis_cli.get(f"tg_cache_id_{channel_id}")
134        if min_id is None:
135            min_id = 0
136        print(channel_id, channel_info[1])
137        entity = await client.get_entity(channel_id)
138        await asyncio.sleep(10)
139        message_list = []
140        # 每次查50条，limit=50
141        if min_id == 0:
142            # 这个地方含义是获取2022-10-01这天的数据,从旧数据到新数据的顺序，这样message id逐渐增大。
143            async for message in client.iter_messages(entity, reverse=True, offset_date=last_offset_date, limit=50):
144                doc_data = await format_message(message, channel_id)
145                if doc_data:
146                    message_list.append(doc_data)
147        else:
148            # 根据message id获取数据，id逐渐增大。
149            async for message in client.iter_messages(entity, reverse=True, min_id=int(min_id),
150                                                      limit=50):
151                doc_data = await format_message(message, channel_id)
152                if doc_data:
153                    message_list.append(doc_data)
154        # robot_warning(message_list, robot_url)
155        mongo_cli.update(message_list)
156
157        print("发送完成")
158
159
160def start():
161    with client:
162        scheduler = AsyncIOScheduler()
163        scheduler.add_job(load_history_to_save, 'interval', minutes=5, max_instances=1,
164                          next_run_time=datetime.datetime.now())
165        scheduler.start()
166        client.loop.run_forever()
167
168
169if __name__ == '__main__':
170    start()