Improved linting and removed unused imports
This commit is contained in:
Profitroll
parent
807e629ae7
commit
b766d0c52c
|
|
@ -2,9 +2,7 @@ from os import path, remove, sep
|
|||
from shutil import rmtree
|
||||
from typing import Union
|
||||
from pyrogram.client import Client
|
||||
from pyrogram.types import Message, CallbackQuery
|
||||
from pyrogram.enums.parse_mode import ParseMode
|
||||
from pyrogram.session.session import Session
|
||||
from pyrogram.types import Message
|
||||
from classes.exceptions import SubmissionDuplicatesError, SubmissionUnavailableError
|
||||
from modules.api_client import upload_pic
|
||||
from modules.database import col_submitted
|
||||
|
|
|
|||
|
|
@ -9,16 +9,28 @@ class PosterUser():
|
|||
self.id = id
|
||||
|
||||
def is_blocked(self) -> bool:
|
||||
"""Check if user is banned from submitting content.
|
||||
|
||||
### Returns:
|
||||
`bool`: Must be `True` if banned and `False` if not
|
||||
"""
|
||||
return False if col_banned.find_one({"user": self.id}) is None else True
|
||||
|
||||
def block(self) -> None:
|
||||
"""Ban user from using command and submitting content."""
|
||||
if col_banned.find_one({"user": self.id}) is None:
|
||||
col_banned.insert_one({"user": self.id, "date": datetime.now()})
|
||||
|
||||
def unblock(self) -> None:
|
||||
"""Allow user to use command and submit posts again."""
|
||||
col_banned.find_one_and_delete({"user": self.id})
|
||||
|
||||
def is_limited(self) -> bool:
|
||||
"""Check if user is on a cooldown after submitting something.
|
||||
|
||||
### Returns:
|
||||
`bool`: Must be `True` if on the cooldown and `False` if not
|
||||
"""
|
||||
if self.id in app.admins:
|
||||
return False
|
||||
else:
|
||||
|
|
@ -28,5 +40,6 @@ class PosterUser():
|
|||
return True if (datetime.now() - db_record["cooldown"]).total_seconds() < configGet("timeout", "submission") else False
|
||||
|
||||
def limit(self) -> None:
|
||||
"""Restart user's cooldown. Used after post has been submitted."""
|
||||
if col_users.find_one_and_update({"user": self.id}, {"$set": {"cooldown": datetime.now()}}) is None:
|
||||
col_users.insert_one({"user": self.id, "cooldown": datetime.now()})
|
||||
|
|
@ -3,7 +3,7 @@ from classes.poster_client import PosterClient
|
|||
from pyrogram.types import BotCommand, BotCommandScopeChat
|
||||
from modules.utils import configGet, locale
|
||||
|
||||
async def register_commands(app: PosterClient):
|
||||
async def register_commands(app: PosterClient) -> None:
|
||||
|
||||
if configGet("submit", "mode"):
|
||||
# Registering user commands
|
||||
|
|
|
|||
|
|
@ -16,7 +16,7 @@ with open(getcwd()+path.sep+"config.json", "r", encoding='utf8') as file:
|
|||
file.close()
|
||||
|
||||
# Check latest log size
|
||||
def checkSize(debug=False):
|
||||
def checkSize(debug=False) -> None:
|
||||
|
||||
global log_folder
|
||||
|
||||
|
|
@ -39,7 +39,7 @@ def checkSize(debug=False):
|
|||
pass
|
||||
|
||||
# Append string to log
|
||||
def logAppend(message, debug=False):
|
||||
def logAppend(message, debug=False) -> None:
|
||||
|
||||
global log_folder
|
||||
|
||||
|
|
@ -56,7 +56,7 @@ def logAppend(message, debug=False):
|
|||
log.close()
|
||||
|
||||
# Print to stdout and then to log
|
||||
def logWrite(message, debug=False):
|
||||
def logWrite(message, debug=False) -> None:
|
||||
# save to log file and rotation is to be done
|
||||
logAppend(f'{message}', debug=debug)
|
||||
print(f"{message}", flush=True)
|
||||
|
|
@ -5,7 +5,6 @@ from traceback import format_exc
|
|||
from uuid import uuid4
|
||||
from PIL import Image
|
||||
|
||||
from bson import ObjectId
|
||||
from classes.poster_client import PosterClient
|
||||
from requests import get
|
||||
|
||||
|
|
@ -15,7 +14,7 @@ from modules.logger import logWrite
|
|||
from modules.utils import configGet, locale
|
||||
|
||||
|
||||
async def send_content(app: PosterClient):
|
||||
async def send_content(app: PosterClient) -> None:
|
||||
|
||||
try:
|
||||
|
||||
|
|
@ -118,138 +117,4 @@ async def send_content(app: PosterClient):
|
|||
try:
|
||||
rmtree(path.join(configGet("tmp", "locations"), tmp_dir), ignore_errors=True)
|
||||
except:
|
||||
pass
|
||||
|
||||
|
||||
# async def send_content_old(app: PosterClient):
|
||||
|
||||
# # Send post to channel
|
||||
# try:
|
||||
|
||||
# index = jsonLoad(configGet("index", "locations"))
|
||||
|
||||
# if configGet("api_based", "mode"):
|
||||
|
||||
# try:
|
||||
# pic = random_pic()
|
||||
# except:
|
||||
# logWrite(locale("post_empty", "console", locale=configGet("locale")))
|
||||
# if configGet("error", "reports"):
|
||||
# await app.send_message(configGet("admin"), locale("post_empty", "message", locale=configGet("locale")))
|
||||
# return
|
||||
|
||||
# token = authorize()
|
||||
|
||||
# response = get(f'{configGet("address", "posting", "api")}/photos/{pic[0]}', headers={"Authorization": f"Bearer {token}"}, stream=True)
|
||||
|
||||
# with open(configGet("tmp", "locations")+sep+pic[0]+".jpg", 'wb') as out_file:
|
||||
# copyfileobj(response.raw, out_file)
|
||||
|
||||
# del response
|
||||
|
||||
# candidate = configGet("tmp", "locations")+sep+pic[0]+".jpg"
|
||||
# candidate_file = pic[1]
|
||||
# ext_type = "photo"
|
||||
|
||||
# if not configGet("api_based", "mode"):
|
||||
|
||||
# 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"):
|
||||
# await app.send_message(configGet("admin"), locale("post_empty", "message", locale=configGet("locale")))
|
||||
# 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":
|
||||
|
||||
# if configGet("enabled", "caption"):
|
||||
# if configGet("link", "caption") != None:
|
||||
# sent = await app.send_photo(configGet("channel", "posting"), candidate, caption=caption, disable_notification=configGet("silent", "posting"))
|
||||
# else:
|
||||
# sent = await app.send_photo(configGet("channel", "posting"), candidate, caption=caption, disable_notification=configGet("silent", "posting"))
|
||||
# else:
|
||||
# sent = await app.send_photo(configGet("channel", "posting"), candidate, caption=caption, disable_notification=configGet("silent", "posting"))
|
||||
|
||||
# elif ext_type == "video":
|
||||
|
||||
# if configGet("enabled", "caption"):
|
||||
# if configGet("link", "caption") != None:
|
||||
# sent = await app.send_video(configGet("channel", "posting"), candidate, caption=caption, disable_notification=configGet("silent", "posting"))
|
||||
# else:
|
||||
# sent = await app.send_video(configGet("channel", "posting"), candidate, caption=caption, disable_notification=configGet("silent", "posting"))
|
||||
# else:
|
||||
# sent = await app.send_video(configGet("channel", "posting"), candidate, caption=caption, disable_notification=configGet("silent", "posting"))
|
||||
|
||||
# else:
|
||||
# return
|
||||
|
||||
# if configGet("api_based", "mode"):
|
||||
# remove(configGet("tmp", "locations")+sep+pic[0]+".jpg")
|
||||
# move_pic(pic[0])
|
||||
|
||||
# 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"))))
|
||||
|
||||
# if configGet("sent", "reports"):
|
||||
# await 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)]
|
||||
# ]))
|
||||
|
||||
# except Exception as exp:
|
||||
# logWrite(locale("post_exception", "console", locale=configGet("locale")).format(str(exp), format_exc()))
|
||||
# if configGet("error", "reports"):
|
||||
# await app.send_message(configGet("admin"), locale("post_exception", "message", locale=configGet("locale")).format(exp, format_exc()))
|
||||
# pass
|
||||
|
||||
|
||||
# Work in progress
|
||||
# Check last posts forwards
|
||||
# check_forwards(app)
|
||||
pass
|
||||
|
|
@ -9,10 +9,11 @@ from sys import exit
|
|||
from os import sep, kill
|
||||
from os import name as osname
|
||||
from traceback import print_exc
|
||||
from typing import Any
|
||||
|
||||
from modules.logger import logWrite
|
||||
|
||||
def jsonLoad(filename):
|
||||
def jsonLoad(filename: str) -> Any:
|
||||
"""Loads arg1 as json and returns its contents"""
|
||||
with open(filename, "r", encoding='utf8') as file:
|
||||
try:
|
||||
|
|
@ -26,7 +27,7 @@ def jsonLoad(filename):
|
|||
file.close()
|
||||
return output
|
||||
|
||||
def jsonSave(contents, filename):
|
||||
def jsonSave(contents: Any, filename: str) -> None:
|
||||
"""Dumps dict/list arg1 to file arg2"""
|
||||
try:
|
||||
with open(filename, "w", encoding='utf8') as file:
|
||||
|
|
@ -108,10 +109,14 @@ except ModuleNotFoundError:
|
|||
print(locale("deps_missing", "console", locale=configGet("locale")), flush=True)
|
||||
exit()
|
||||
|
||||
def killProc(pid):
|
||||
def killProc(pid: int) -> None:
|
||||
"""Kill process by its PID. Meant to be used to kill the main process of bot itself.
|
||||
|
||||
### Args:
|
||||
* pid (`int`): PID of the target
|
||||
"""
|
||||
if osname == "posix":
|
||||
from signal import SIGKILL
|
||||
kill(pid, SIGKILL)
|
||||
else:
|
||||
p = Process(pid)
|
||||
p.kill()
|
||||
Process(pid).kill()
|
||||
|
|
@ -1,12 +1,8 @@
|
|||
from os import getpid
|
||||
|
||||
from pyrogram import filters
|
||||
from classes.poster_client import PosterClient
|
||||
from pyrogram.types import Message
|
||||
|
||||
from classes.poster_client import PosterClient
|
||||
from modules.app import app
|
||||
from modules.logger import logWrite
|
||||
from modules.utils import configGet, killProc, locale
|
||||
|
||||
|
||||
@app.on_message(~ filters.scheduled & filters.command(["import"], prefixes=["", "/"]))
|
||||
|
|
|
|||
|
|
@ -175,9 +175,6 @@ async def get_submission(app: PosterClient, msg: Message):
|
|||
[
|
||||
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}")
|
||||
# ]
|
||||
]
|
||||
|
||||
PosterUser(msg.from_user.id).limit()
|
||||
|
|
|
|||
Loading…
Reference in New Issue