1
0
Fork 0
This bot is used for one and only task - post pictures from my personal archive. https://t.me/prashuryashur
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

475 lines
22 KiB

from os import sep, remove, getpid, listdir
from random import choice
from shutil import move
from sys import argv, exit
from threading import Thread
from time import time
from traceback import format_exc
from pathlib import Path
from argparse import ArgumentParser
from modules.logging import logWrite
from modules.utils import configGet, jsonLoad, jsonSave, killProc, locale
# Args =====================================================================================================================================
parser = ArgumentParser(
prog = "Telegram Poster",
description = "Bot for posting some of your stuff and also receiving submissions."
)
parser.add_argument("-m", "--move-sent", action="store_true")
parser.add_argument("-c", "--cleanup", action="store_true")
parser.add_argument("--confirm", action="store_true")
parser.add_argument("-i", "--cleanup-index", action="store_true")
parser.add_argument("-n", "--norun", action="store_true")
args = parser.parse_args()
if args.move_sent:
for entry in jsonLoad(configGet("index", "locations"))["sent"]:
try:
move(configGet("queue", "locations")+sep+entry, configGet("sent", "locations")+sep+entry)
except FileNotFoundError:
logWrite(locale("move_sent_doesnt_exist", "console", locale=configGet("locale")).format(entry))
except Exception as exp:
logWrite(locale("move_sent_doesnt_exception", "console", locale=configGet("locale")).format(entry, exp))
logWrite(locale("move_sent_completed", "console", locale=configGet("locale")))
if args.cleanup:
if args.confirm:
index = jsonLoad(configGet("index", "locations"))
for entry in index["sent"]:
try:
try:
remove(configGet("queue", "locations")+sep+entry)
except FileNotFoundError:
pass
try:
remove(configGet("sent", "locations")+sep+entry)
except FileNotFoundError:
pass
except Exception as exp:
logWrite(locale("cleanup_exception", "console", locale=configGet("locale")).format(entry, exp))
jsonSave(index, jsonLoad(configGet("index", "locations")))
logWrite(locale("cleanup_completed", "console", locale=configGet("locale")))
else:
logWrite(locale("cleanup_unathorized", "console", locale=configGet("locale")))
if args.cleanup_index:
if args.confirm:
index = jsonLoad(configGet("index", "locations"))
index["sent"] = []
jsonSave(index, jsonLoad(configGet("index", "locations")))
logWrite(locale("cleanup_index_completed", "console", locale=configGet("locale")))
else:
logWrite(locale("cleanup_index_unathorized", "console", locale=configGet("locale")))
if args.norun:
logWrite(locale("passed_norun", "console", locale=configGet("locale_log")))
exit()
#===========================================================================================================================================
# Import ===================================================================================================================================
try:
import schedule # type: ignore
from pyrogram import Client, filters, idle # type: ignore
from pyrogram.types import ChatPermissions, ReplyKeyboardMarkup, InlineKeyboardMarkup, InlineKeyboardButton, BotCommand, BotCommandScopeChat # type: ignore
from pyrogram.raw.types import UpdateChannelMessageForwards, InputChannel, InputPeerChannel
from pyrogram.raw.functions.stats import GetMessagePublicForwards
except ModuleNotFoundError:
print(locale("deps_missing", "console", locale=configGet("locale")), flush=True)
exit()
#===========================================================================================================================================
pid = getpid()
app = Client("duptsiaposter", bot_token=configGet("bot_token", "bot"), api_id=configGet("api_id", "bot"), api_hash=configGet("api_hash", "bot"))
# Work in progress
# def check_forwards(app):
# try:
# index = jsonLoad(configGet("index", "locations"))
# channel = app.get_chat(configGet("channel", "posting"))
# peer = app.resolve_peer(configGet("channel", "posting"))
# print(peer, flush=True)
# posts_list = [i for i in range(index["last_id"]-100,index["last_id"])]
# last_posts = app.get_messages(configGet("channel", "posting"), message_ids=posts_list)
# for post in last_posts:
# post_forwards = GetMessagePublicForwards(channel=peer, msg_id=post.id, offset_peer=peer, offset_rate=0, offset_id=0, limit=100)
# print(post_forwards, flush=True)
# for forward in post_forwards:
# print(forward, flush=True)
# except Exception as exp:
# logWrite("Could not get last posts forwards due to {0} with traceback {1}".format(str(exp), traceback.format_exc()), debug=True)
# if configGet("error", "reports"):
# app.send_message(configGet("admin"), traceback.format_exc()) # type: ignore
# pass
def send_content():
# Send post to channel
try:
index = jsonLoad(configGet("index", "locations"))
list_queue = listdir(configGet("queue", "locations"))
for file in list_queue:
if not file in index["sent"]:
ext_match = False
for ext in configGet("photo", "posting", "extensions"):
if file.endswith(ext):
ext_match = True
ext_type = "photo"
break
for ext in configGet("video", "posting", "extensions"):
if file.endswith(ext):
ext_match = True
ext_type = "video"
break
if not ext_match:
list_queue.remove(file)
else:
list_queue.remove(file)
if len(list_queue) > 0:
candidate_file = choice(list_queue)
candidate = configGet("queue", "locations")+sep+candidate_file
else:
logWrite(locale("post_empty", "console", locale=configGet("locale")))
if configGet("error", "reports"):
app.send_message(configGet("admin"), locale("post_empty", "message", locale=configGet("locale"))) # type: ignore
return
if candidate_file in index["captions"]:
caption = index["captions"][candidate_file]
else:
caption = ""
if configGet("enabled", "caption"):
if configGet("link", "caption") != None:
caption = f"{caption}\n\n[{configGet('text', 'caption')}]({configGet('link', 'caption')})"
else:
caption = f"{caption}\n\n{configGet('text', 'caption')}"
else:
caption = caption
if ext_type == "photo": # type: ignore
if configGet("enabled", "caption"):
if configGet("link", "caption") != None:
sent = app.send_photo(configGet("channel", "posting"), candidate, caption=caption, disable_notification=configGet("silent", "posting")) # type: ignore
else:
sent = app.send_photo(configGet("channel", "posting"), candidate, caption=caption, disable_notification=configGet("silent", "posting")) # type: ignore
else:
sent = app.send_photo(configGet("channel", "posting"), candidate, caption=caption, disable_notification=configGet("silent", "posting")) # type: ignore
elif ext_type == "video": # type: ignore
if configGet("enabled", "caption"):
if configGet("link", "caption") != None:
sent = app.send_video(configGet("channel", "posting"), candidate, caption=caption, disable_notification=configGet("silent", "posting")) # type: ignore
else:
sent = app.send_video(configGet("channel", "posting"), candidate, caption=caption, disable_notification=configGet("silent", "posting")) # type: ignore
else:
sent = app.send_video(configGet("channel", "posting"), candidate, caption=caption, disable_notification=configGet("silent", "posting")) # type: ignore
else:
return
index["sent"].append(candidate_file)
index["last_id"] = sent.id
jsonSave(index, configGet("index", "locations"))
if configGet("move_sent", "posting"):
move(candidate, configGet("sent", "locations")+sep+candidate_file)
logWrite(locale("post_sent", "console", locale=configGet("locale")).format(candidate, ext_type, str(configGet("channel", "posting")), caption.replace("\n", "%n"), str(configGet("silent", "posting")))) # type: ignore
if configGet("sent", "reports"):
app.send_message(configGet("admin"), f"Posted `{candidate_file}`", disable_web_page_preview=True, reply_markup=InlineKeyboardMarkup([
[InlineKeyboardButton(locale("post_view", "button", locale=configGet("locale")), url=sent.link)] # type: ignore
])) # type: ignore
except Exception as exp:
logWrite(locale("post_exception", "console", locale=configGet("locale")).format(str(exp), format_exc()))
if configGet("error", "reports"):
app.send_message(configGet("admin"), locale("post_exception", "message", locale=configGet("locale")).format(exp, traceback.format_exc())) # type: ignore
pass
# Work in progress
# Check last posts forwards
# check_forwards(app)
if configGet("submit", "mode"):
@app.on_message(~ filters.scheduled & filters.command(["start"], prefixes="/"))
def cmd_start(app, msg):
if msg.from_user.id not in jsonLoad(configGet("blocked", "locations")):
msg.reply_text(locale("start", "message", locale=msg.from_user.language_code))
if configGet("submit", "mode"):
@app.on_message(~ filters.scheduled & filters.command(["rules", "help"], prefixes="/"))
def cmd_rules(app, msg):
if msg.from_user.id not in jsonLoad(configGet("blocked", "locations")):
msg.reply_text(locale("rules", "message", locale=msg.from_user.language_code))
# Work in progress
# @app.on_message(~ filters.scheduled & filters.command(["forwards"], prefixes="/"))
# def cmd_forwards(app, msg):
# check_forwards(app)
@app.on_message(~ filters.scheduled & filters.command(["kill", "die", "reboot"], prefixes=["", "/"]))
def cmd_kill(app, msg):
if msg.from_user.id == configGet("admin"):
logWrite(locale("shutdown", "console", locale=configGet("locale")).format(str(pid)))
msg.reply_text(locale("shutdown", "message", locale=configGet("locale")).format(str(pid)))
killProc(pid)
# Submission =====================================================================================================================================
def subLimit(user):
submit = jsonLoad(configGet("submit", "locations"))
submit[str(user.id)] = time()
jsonSave(submit, configGet("submit", "locations"))
def subLimited(user):
if user.id == configGet("admin"):
return False
else:
submit = jsonLoad(configGet("submit", "locations"))
if str(user.id) in submit:
if (time() - submit[str(user.id)]) < configGet("timeout", "submission"):
return True
else:
return False
else:
return False
def subBlock(user):
blocked = jsonLoad(configGet("blocked", "locations"))
if user not in blocked:
blocked.append(user)
jsonSave(blocked, configGet("blocked", "locations"))
def subUnblock(user):
blocked = jsonLoad(configGet("blocked", "locations"))
if user in blocked:
blocked.remove(user)
jsonSave(blocked, configGet("blocked", "locations"))
if configGet("submit", "mode"):
@app.on_message(~ filters.scheduled & filters.photo | filters.video | filters.animation | filters.document)
def get_submission(_, msg):
try:
if msg.from_user.id not in jsonLoad(configGet("blocked", "locations")):
user_locale = msg.from_user.language_code
if not subLimited(msg.from_user):
if msg.document != None:
if msg.document.mime_type not in configGet("mime_types", "submission"):
msg.reply_text(locale("mime_not_allowed", "message", locale=user_locale), quote=True)
return
if msg.document.file_size > configGet("file_size", "submission"):
msg.reply_text(locale("document_too_large", "message", locale=user_locale).format(str(configGet("file_size", "submission")/1024/1024)), quote=True)
return
if msg.video != None:
if msg.video.file_size > configGet("file_size", "submission"):
msg.reply_text(locale("document_too_large", "message", locale=user_locale).format(str(configGet("file_size", "submission")/1024/1024)), quote=True)
return
buttons = [
[
InlineKeyboardButton(text=locale("sub_yes", "button", locale=configGet("locale")), callback_data=f"sub_yes_{msg.from_user.id}_{msg.id}")
]
]
if msg.caption != None:
caption = str(msg.caption)
buttons[0].append(
InlineKeyboardButton(text=locale("sub_yes_caption", "button", locale=configGet("locale")), callback_data=f"sub_yes_{msg.from_user.id}_{msg.id}_caption")
)
buttons[0].append(
InlineKeyboardButton(text=locale("sub_no", "button", locale=configGet("locale")), callback_data=f"sub_no_{msg.from_user.id}_{msg.id}")
)
else:
caption = ""
buttons[0].append(
InlineKeyboardButton(text=locale("sub_no", "button", locale=configGet("locale")), callback_data=f"sub_no_{msg.from_user.id}_{msg.id}")
)
caption += locale("sub_by", "message", locale=locale(configGet("locale")))
if msg.from_user.first_name != None:
caption += f" {msg.from_user.first_name}"
if msg.from_user.last_name != None:
caption += f" {msg.from_user.last_name}"
if msg.from_user.username != None:
caption += f" (@{msg.from_user.username})"
if msg.from_user.phone_number != None:
caption += f" ({msg.from_user.phone_number})"
msg.copy(configGet("admin"), caption=caption, reply_markup=InlineKeyboardMarkup(buttons))
if msg.from_user.id != configGet("admin"):
buttons += [
[
InlineKeyboardButton(text=locale("sub_block", "button", locale=configGet("locale")), callback_data=f"sub_block_{msg.from_user.id}")
],
[
InlineKeyboardButton(text=locale("sub_unblock", "button", locale=configGet("locale")), callback_data=f"sub_unblock_{msg.from_user.id}")
]
]
msg.reply_text(locale("sub_sent", "message", locale=user_locale), quote=True)
subLimit(msg.from_user)
else:
msg.reply_text(locale("sub_cooldown", "message", locale=user_locale).format(str(configGet("timeout", "submission"))))
except AttributeError:
logWrite(f"from_user in function get_submission does not seem to contain id")
@app.on_callback_query(filters.regex("sub_yes_[\s\S]*_[\s\S]*")) # type: ignore
def callback_query_yes(app, clb): # type: ignore
fullclb = clb.data.split("_")
user_locale = clb.from_user.language_code
try:
submission = app.get_messages(int(fullclb[2]), int(fullclb[3]))
except:
clb.answer(text=locale("sub_msg_unavail", "message", locale=user_locale), show_alert=True)
return
try:
media = app.download_media(submission, file_name=configGet("queue", "locations")+sep)
if clb.data.endswith("_caption"):
index = jsonLoad(configGet("index", "locations"))
index["captions"][Path(media).name] = submission.caption
jsonSave(index, configGet("index", "locations"))
except:
clb.answer(text=locale("sub_media_unavail", "message", locale=user_locale), show_alert=True)
return
submission.reply_text(locale("sub_yes", "message", locale=submission.from_user.language_code), quote=True)
clb.answer(text=locale("sub_yes", "callback", locale=user_locale).format(fullclb[2]), show_alert=True)
@app.on_callback_query(filters.regex("sub_no_[\s\S]*_[\s\S]*")) # type: ignore
def callback_query_no(app, clb): # type: ignore
fullclb = clb.data.split("_")
user_locale = clb.from_user.language_code
try:
submission = app.get_messages(int(fullclb[2]), int(fullclb[3]))
except:
clb.answer(text=locale("sub_msg_unavail", "message", locale=user_locale), show_alert=True)
return
submission.reply_text(locale("sub_no", "message", locale=submission.from_user.language_code), quote=True)
clb.answer(text=locale("sub_no", "callback", locale=user_locale).format(fullclb[2]), show_alert=True)
@app.on_callback_query(filters.regex("sub_block_[\s\S]*")) # type: ignore
def callback_query_block(app, clb): # type: ignore
fullclb = clb.data.split("_")
user_locale = clb.from_user.language_code
app.send_message(int(fullclb[2]), locale("sub_msg_unavail", "message", locale=configGet("locale")))
subBlock(int(fullclb[2]))
clb.answer(text=locale("sub_block", "callback", locale=user_locale).format(fullclb[2]), show_alert=True)
@app.on_callback_query(filters.regex("sub_unblock_[\s\S]*")) # type: ignore
def callback_query_unblock(app, clb): # type: ignore
fullclb = clb.data.split("_")
user_locale = clb.from_user.language_code
app.send_message(int(fullclb[2]), locale("sub_msg_unavail", "message", locale=configGet("locale")))
subUnblock(int(fullclb[2]))
clb.answer(text=locale("sub_unblock", "callback", locale=user_locale).format(fullclb[2]), show_alert=True)
#===========================================================================================================================================
# Work in progress
# Handle new forwards
# @app.on_raw_update()
# def fwd_got(app, update, users, chats):
# if isinstance(update, UpdateChannelMessageForwards):
# logWrite(f'Forward count increased to {update["forwards"]} on post {update["id"]} in channel {update["channel_id"]}')
# logWrite(str(users), debug=True)
# logWrite(str(chats), debug=True)
# # else:
# # logWrite(f"Got raw update of type {type(update)} with contents {update}", debug=True)
if configGet("post", "mode"):
for entry in configGet("time", "posting"):
schedule.every().day.at(entry).do(send_content)
def background_task():
try:
while True:
try:
schedule.run_pending()
time.sleep(1)
except:
pass
except Exception as exp:
logWrite(locale("exception_occured", "console", locale=configGet("locale")).format(exp))
except KeyboardInterrupt:
logWrite(locale("keyboard_interrupt", "console", locale=configGet("locale")))
if configGet("shutdown", "reports"):
app.send_message(configGet("admin"), locale("shutdown", "message", locale=configGet("locale")).format(str(pid))) # type: ignore
killProc(pid)
if __name__ == "__main__":
logWrite(locale("startup", "console", locale=configGet("locale")).format(str(pid)))
app.start() # type: ignore
if configGet("startup", "reports"):
app.send_message(configGet("admin"), locale("startup", "message", locale=configGet("locale")).format(str(pid))) # type: ignore
if configGet("post", "mode"):
t = Thread(target=background_task)
t.start()
if configGet("submit", "mode"):
# Registering user commands
for entry in listdir(configGet("locale", "locations")):
if entry.endswith(".json"):
commands_list = []
for command in configGet("commands"):
commands_list.append(BotCommand(command, locale(command, "commands", locale=entry.replace(".json", ""))))
app.set_bot_commands(commands_list, language_code=entry.replace(".json", "")) # type: ignore
# Registering user commands for fallback locale
commands_list = []
for command in configGet("commands"):
commands_list.append(BotCommand(command, locale(command, "commands", locale=configGet("locale_fallback"))))
app.set_bot_commands(commands_list) # type: ignore
# Registering admin commands
commands_admin_list = []
if configGet("submit", "mode"):
for command in configGet("commands"):
commands_admin_list.append(BotCommand(command, locale(command, "commands", locale=configGet("locale"))))
for command in configGet("commands_admin"):
commands_admin_list.append(BotCommand(command, locale(command, "commands_admin", locale=configGet("locale"))))
app.set_bot_commands(commands_admin_list, scope=BotCommandScopeChat(chat_id=configGet("admin"))) # type: ignore
idle()
app.send_message(configGet("admin"), locale("shutdown", "message", locale=configGet("locale")).format(str(pid))) # type: ignore
logWrite(locale("shutdown", "console", locale=configGet("locale")).format(str(pid)))
killProc(pid)