This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
Telegram/modules/handlers/analytics_group.py

256 lines
8.8 KiB
Python

from datetime import datetime
from polyglot.detect import Detector
from pyrogram import filters
from pyrogram.client import Client
from pyrogram.enums import MessageEntityType, PollType
from pyrogram.types import Message
from app import app
from modules import custom_filters
from modules.database import col_analytics_group
from modules.logging import logWrite
from modules.utils import configGet
@app.on_message(
custom_filters.enabled_general
& ~filters.scheduled
& filters.chat(configGet("users", "groups"))
)
async def msg_destination_group(app: Client, msg: Message):
analytics_entry = {
"id": msg.id,
"user": msg.from_user.id,
"date": datetime.now(),
"reply": {
"id": msg.reply_to_message_id,
"top_id": msg.reply_to_top_message_id,
"user": None
if msg.reply_to_message is None
else msg.reply_to_message.from_user.id,
},
"forward": {
"id": msg.forward_from_message_id,
"chat": None if msg.forward_from_chat is None else msg.forward_from_chat.id,
"user": None if msg.forward_from is None else msg.forward_from.id,
"date": msg.forward_date,
},
"media_spoilered": msg.has_media_spoiler,
"entities": {"links": [], "mentions": []},
"text": None,
"language": None,
"language_confidence": None,
"animation": None,
"audio": None,
"contact": None,
"document": None,
"location": None,
"photo": None,
"poll": None,
"sticker": None,
"venue": None,
"video": None,
"videonote": None,
"voice": None,
}
if msg.text is not None or msg.caption is not None:
text = msg.text if msg.text is not None else msg.caption
analytics_entry["text"] = text
if msg.entities is not None or msg.caption_entities is not None:
entities = (
msg.entities if msg.entities is not None else msg.caption_entities
)
for entity in entities:
if entity.type == MessageEntityType.TEXT_LINK:
analytics_entry["entities"]["links"].append(entity.url)
elif entity.type == MessageEntityType.URL:
analytics_entry["entities"]["links"].append(
text[entity.offset : entity.offset + entity.length]
)
elif entity.type == MessageEntityType.TEXT_MENTION:
analytics_entry["entities"]["mentions"].append(entity.user.id)
elif entity.type == MessageEntityType.MENTION:
analytics_entry["entities"]["mentions"].append(
text[entity.offset : entity.offset + entity.length]
)
lang = Detector(text, quiet=True).language
analytics_entry["language"] = lang.code
analytics_entry["language_confidence"] = lang.confidence
if lang.code == "ru":
logWrite(
f"Message '{text}' from {msg.from_user.first_name} ({msg.from_user.id}) is fucking russian [confidence {lang.confidence}]"
)
if msg.animation is not None:
analytics_entry["animation"] = {
"id": msg.animation.file_id,
"duration": msg.animation.duration,
"height": msg.animation.height,
"width": msg.animation.width,
"file_name": msg.animation.file_name,
"mime_type": msg.animation.mime_type,
}
if msg.audio is not None:
analytics_entry["audio"] = {
"id": msg.audio.file_id,
"title": msg.audio.title,
"performer": msg.audio.performer,
"duration": msg.audio.duration,
"file_name": msg.audio.file_name,
"file_size": msg.audio.file_size,
"mime_type": msg.audio.mime_type,
}
if msg.contact is not None:
analytics_entry["contact"] = {
"id": msg.contact.user_id,
"first_name": msg.contact.first_name,
"last_name": msg.contact.last_name,
"phone_number": msg.contact.phone_number,
"vcard": msg.contact.vcard,
}
if msg.document is not None:
analytics_entry["document"] = {
"id": msg.document.file_id,
"file_name": msg.document.file_name,
"file_size": msg.document.file_size,
"mime_type": msg.document.mime_type,
}
if msg.location is not None:
analytics_entry["location"] = {
"longitude": msg.location.longitude,
"latitude": msg.location.latitude,
}
if msg.photo is not None:
thumbnails = []
for thumbail in msg.photo.thumbs:
thumbnails.append(
{
"id": thumbail.file_id,
"height": thumbail.height,
"width": thumbail.width,
"file_size": thumbail.file_size,
}
)
analytics_entry["photo"] = {
"id": msg.photo.file_id,
"height": msg.photo.height,
"width": msg.photo.width,
"file_size": msg.photo.file_size,
"thumbnails": thumbnails,
}
if msg.poll is not None:
options = []
for option in msg.poll.options:
options.append(option.text)
analytics_entry["poll"] = {
"id": msg.poll.id,
"question": msg.poll.question,
"open_period": msg.poll.open_period,
"close_date": msg.poll.close_date,
"options": options,
"correct_option": msg.poll.correct_option_id,
"explanation": msg.poll.explanation,
"anonymous": msg.poll.is_anonymous,
"multiple_answers": msg.poll.allows_multiple_answers,
"quiz": True if msg.poll.type == PollType.QUIZ else False,
}
if msg.sticker is not None:
thumbnails = []
for thumbail in msg.sticker.thumbs:
thumbnails.append(
{
"id": thumbail.file_id,
"height": thumbail.height,
"width": thumbail.width,
"file_size": thumbail.file_size,
}
)
analytics_entry["sticker"] = {
"id": msg.sticker.file_id,
"emoji": msg.sticker.emoji,
"set_name": msg.sticker.set_name,
"animated": msg.sticker.is_animated,
"video": msg.sticker.is_video,
"height": msg.sticker.height,
"width": msg.sticker.width,
"file_name": msg.sticker.file_name,
"file_size": msg.sticker.file_size,
"mime_type": msg.sticker.mime_type,
"thumbnails": thumbnails,
}
if msg.venue is not None:
analytics_entry["venue"] = {
"title": msg.venue.title,
"address": msg.venue.address,
"longitude": msg.venue.location.longitude,
"latitude": msg.venue.location.latitude,
"foursquare_id": msg.venue.foursquare_id,
"foursquare_type": msg.venue.foursquare_type,
}
if msg.video is not None:
thumbnails = []
for thumbail in msg.video.thumbs:
thumbnails.append(
{
"id": thumbail.file_id,
"height": thumbail.height,
"width": thumbail.width,
"file_size": thumbail.file_size,
}
)
analytics_entry["video"] = {
"id": msg.video.file_id,
"duration": msg.video.duration,
"height": msg.video.height,
"width": msg.video.width,
"file_name": msg.video.file_name,
"file_size": msg.video.file_size,
"mime_type": msg.video.mime_type,
"thumbnails": thumbnails,
}
if msg.video_note is not None:
thumbnails = []
for thumbail in msg.video_note.thumbs:
thumbnails.append(
{
"id": thumbail.file_id,
"height": thumbail.height,
"width": thumbail.width,
"file_size": thumbail.file_size,
}
)
analytics_entry["videonote"] = {
"id": msg.video_note.file_id,
"duration": msg.video_note.duration,
"length": msg.video_note.length,
"file_size": msg.video_note.file_size,
"mime_type": msg.video_note.mime_type,
"thumbnails": thumbnails,
}
if msg.voice is not None:
analytics_entry["voice"] = {
"id": msg.voice.file_id,
"duration": msg.voice.duration,
"file_size": msg.voice.file_size,
"mime_type": msg.voice.mime_type,
}
col_analytics_group.insert_one(analytics_entry)