Telegram_luanon
unknown
plain_text
3 years ago
5.2 kB
14
Indexable
import os import time from telethon import TelegramClient, sync, events from telethon.tl.functions.channels import JoinChannelRequest, LeaveChannelRequest from .utils import Utils, TelegramTask from .exceptions import NoValue class Telegram: def __init__(self, phone=None, api_ịd=None, api_hash=None, delay=None): self.Utils = Utils() self.tasks = [] self.consts = {} self.tag = "Telegram" self.phone = phone or self.raise_error("noValue", "phone") self.api_id = api_ịd or 1339492 self.api_hash = api_hash or "262d46fcf6756c14123ee6910c6ee5f3" self.delay = delay or 2 self.check_file() def raise_error(self, ex_type=None, ex_name=None): if ex_type == "noValue": raise NoValue("Value {0} bị thiếu, vui lòng truyền vào value {0}".format(ex_name)) def check_file(self): if not os.path.exists("session"): os.makedirs("session") if not os.listdir("session"): print("Bỏ tệp +84966xxx.session vào thư mục session để chạy tool") exit() def sess_connect(self): try: client = TelegramClient("session/%s" %self.phone, self.api_id, self.api_hash) client.connect() if not client.is_user_authorized(): self.Utils.wrong(self.tag, "Không có kể nối: %s" %self.phone) return False return client except Exception as e: traceback = "Sess_connect: %s - %s - %s" %(e.__traceback__.tb_frame.f_code.co_filename, e.__traceback__.tb_lineno, e) self.Utils.wrong(self.tag, traceback) return False def add_task(self, entity=None, task_answer=None, task_type=None, task_delay=None, const=None, command=None): self.tasks.append({"entity": entity, "task_answer": task_answer, "task_type": task_type, "task_delay": task_delay or self.delay, "const": const, "command": command}) def run(self): done = False for _ in range(3): try: sess = self.sess_connect() if sess == False: continue for task in self.tasks: entity = task["entity"] entity = sess.get_entity(entity) if entity else entity task_answer = task["task_answer"] task_type = task["task_type"] task_delay = task["task_delay"] const = task["const"] command = task["command"] argcount = command.__code__.co_argcount if command else 0 if task_type == "SMSG": if const: if const in self.consts.keys(): text = self.consts[const] else: if "cache" in self.consts.keys(): text = self.consts["cache"] else: text = task_answer else: text = task_answer if argcount == 1: text = command(text) sess.send_message(entity, text) elif task_type == "GMSG": if isinstance(task_answer, (list, tuple)): limit, index = task_answer msg = sess.get_messages(entity, limit=limit)[index] else: limit, index = 1, 0 msg = sess.get_messages(entity, limit=limit)[index] text = msg.message if argcount == 1: text = command(text) if const: self.consts.update({const: text}) else: self.consts.update({"cache": text}) elif task_type == "GIMG": if isinstance(task_answer, (list, tuple)): limit, index = task_answer img = sess.get_messages(entity, limit=limit)[index] else: limit, index = 1, 0 img = sess.get_messages(entity, limit=limit)[index] if img.media: img = img.download_media("captcha.jpg") if argcount == 1: img = command(img) if const: self.consts.update({const: img}) else: self.consts.update({"cache": img}) elif task_type == "C": if isinstance(task_answer, (list, tuple)): row, col = task_answer sess.get_messages(entity, limit=1)[0].click(row, col) elif isinstance(task_answer, int): sess.get_messages(entity, limit=1)[0].click(task_answer) else: self.Utils.wrong(self.tag, "Task_type: '%s' nhận được câu trả lời sai '%s' thay vì int, list, tuple" %(task_type, type(task_answer))) elif task_type in ["J", "L"]: if isinstance(task_answer, str): if task_type == "J": sess(JoinChannelRequest(task_answer)) else: sess(LeaveChannelRequest(task_answer)) elif isinstance(task_answer, (list, tuple)): for group in task_answer: if task_type == "J": sess(JoinChannelRequest(group)) else: sess(LeaveChannelRequest(group)) else: self.Utils.wrong(self.tag, "Task_type: '%s' nhận được câu trả lời sai '%s' thay vì str, list, tuple" %(task_type, type(task_answer))) else: self.Utils.wrong(self.tag, "Task_type: '%s' không phải giá trị đúng" %task_type) time.sleep(task_delay) continue done = True sess.disconnect() self.Utils.wrong(self.tag, "%s: Done" %self.phone) break except Exception as e: traceback = "Run: %s - %s - %s" %(e.__traceback__.tb_frame.f_code.co_filename, e.__traceback__.tb_lineno, e) self.Utils.wrong(self.tag, traceback) continue return done
Editor is loading...