Instantly share code, notes, and snippets.

@blueset /README.md
Last active Sep 30, 2018

Embed
What would you like to do?
Filter all telegram stickers messages in groups for 60 seconds.
  1. Clone https://github.com/vysheng/tg/, and install all necessary dependencies
  2. Apply documents.patch, then ./configure && make
  3. pip3 install pytg
  4. Modify main.py and change tgdir to the correct path.
  5. Log in your tg-cli
  6. nohup ./bin/telegram-cli --json -R -W -P 4458 --permanent-peer-ids --permanent-peer-ids > /dev/null &
  7. nohup python3 kagami.py > /dev/null &
diff --git a/json-tg.c b/json-tg.c
index a58e70a..8395d6d 100644
--- a/json-tg.c
+++ b/json-tg.c
@@ -233,6 +233,41 @@ json_t *json_pack_media (struct tgl_message_media *M) {
case tgl_message_media_video:
case tgl_message_media_document_encr:
assert (json_object_set (res, "type", json_string ("document")) >= 0);
+ if (M->type == tgl_message_media_document){
+ assert (M->document);
+ if (M->document->flags & TGLDF_IMAGE) {
+ assert (json_object_set (res, "media", json_string ("image")) >= 0);
+ } else if (M->document->flags & TGLDF_AUDIO) {
+ assert (json_object_set (res, "media", json_string ("audio")) >= 0);
+ } else if (M->document->flags & TGLDF_VIDEO) {
+ assert (json_object_set (res, "media", json_string ("video")) >= 0);
+ } else if (M->document->flags & TGLDF_STICKER) {
+ assert (json_object_set (res, "media", json_string ("sticker")) >= 0);
+ } else {
+ assert (json_object_set (res, "media", json_string ("document")) >= 0);
+ }
+ if (M->document->caption && strlen (M->document->caption)) {
+ assert (json_object_set (res, "dcaption", json_string (M->document->caption)) >= 0);
+ }
+
+ if (M->document->mime_type) {
+ assert (json_object_set (res, "mime", json_string (M->document->mime_type)) >= 0);
+ }
+
+ if (M->document->w && M->document->h) {
+ assert (json_object_set (res, "width", json_integer (M->document->w)) >= 0);
+ assert (json_object_set (res, "height", json_integer (M->document->h)) >= 0);
+ }
+
+ if (M->document->duration) {
+ assert (json_object_set (res, "duration", json_integer (M->document->duration)) >= 0);
+ }
+ assert (json_object_set (res, "size", json_integer (M->document->size)) >= 0);
+
+ if (M->caption) {
+ assert (json_object_set (res, "caption", json_string (M->caption)) >= 0);
+ }
+ }
break;
case tgl_message_media_unsupported:
assert (json_object_set (res, "type", json_string ("unsupported")) >= 0);
from pytg import Telegram from pytg.receiver import Receiver
from pytg.sender import Sender
from pytg.utils import coroutine
import time
import threading
tgdir = "/root/tg/"
#t = Telegram(telegram=tgdir + "bin/telegram-cli",
# pubkey_file=tgdir + "tg-server.pub")
def rm_msg(sender, msg_id, msg):
print("sticker!", msg_id)
f = msg.get('sender', {'first_name': ''})
warn = f['first_name']
warn += " " + f.get('last_name', '') if f.get('last_name', '') else ''
warn += " (@%s)" % f.get('username', '') if f.get('username', '') else ''
warn += ', no stickers. Warn issued.'
sender.reply(msg_id, warn)
print(warn)
time.sleep(60)
sender.raw("delete_msg %s" % msg_id)
print("removed!", msg_id)
@coroutine
def loop(sender):
try:
while True:
msg = (yield)
if msg.get('flags', 0) == 256 and msg.get('media', {}).get('caption', '') == "sticker.webp":
threading.Thread(target=rm_msg, args=(sender, msg['id'], msg)).start()
except Exception as e:
print(e)
receiver = Receiver("127.0.0.1", 4458)
sender = Sender("127.0.0.1", 4458)
receiver.start()
receiver.message(loop(sender))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment