yasirarism/MissKatyPyro
1
0
Fork 0
mirror of https://github.com/yasirarism/MissKatyPyro.git synced 2025-12-29 09:44:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

'Refactored by Sourcery' (#62)

Browse source 
Co-authored-by: Sourcery AI <>
This commit is contained in:
sourcery-ai[bot] 2023-06-16 23:11:49 +07:00 committed by GitHub
parent 370795d4c7
commit 22b0e3ba35
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 109 additions and 96 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
database/gban_db.py
View file

@ -4,9 +4,7 @@ gbansdb = dbname.gban
async def is_gbanned_user(user_id: int) -> bool:
user = await gbansdb.find_one({"user_id": user_id})
if not user:
return False
return True
return bool(user)
async def add_gban_user(user_id: int):
is_gbanned = await is_gbanned_user(user_id)

4
database/karma_db.py
View file

@ -48,9 +48,7 @@ async def update_karma(chat_id: int, name: str, karma: dict):
async def is_karma_on(chat_id: int) -> bool:
chat = await karmadb.find_one({"chat_id_toggle": chat_id})
if not chat:
return True
return False
return not chat
async def karma_on(chat_id: int):

11
misskaty/helper/human_read.py
View file

@ -34,12 +34,11 @@ def get_readable_time(seconds: int) -> str:
def get_readable_bitrate(bitrate_kbps):
if bitrate_kbps > 10000:
bitrate = str(round(bitrate_kbps / 1000, 2)) + " " + "Mb/s"
else:
bitrate = str(round(bitrate_kbps, 2)) + " " + "kb/s"
return bitrate
return (
f"{str(round(bitrate_kbps / 1000, 2))} Mb/s"
if bitrate_kbps > 10000
else f"{str(round(bitrate_kbps, 2))} kb/s"
)
def get_readable_time2(seconds: int) -> str:

28
misskaty/helper/kuso_utils.py
View file

@ -38,12 +38,11 @@ async def kusonimeBypass(url: str, slug=None):
e = traceback.format_exc()
LOGGER.error(e)
title, season, tipe, status_anime, ep, score, duration, rilis = "None", "None", "None", "None", 0, 0, 0, "None"
num = 1
genre = []
for _genre in soup.select("#venkonten > div.vezone > div.venser > div.venutama > div.lexot > div.info > p:nth-child(2)"):
gen = _genre.text.split(":").pop().strip().split(", ")
genre = gen
for smokedl in soup.find("div", {"class": "dlbodz"}).find_all("div", {"class": "smokeddlrh"}):
for num, smokedl in enumerate(soup.find("div", {"class": "dlbodz"}).find_all("div", {"class": "smokeddlrh"}), start=1):
mendata = {"name": title, "links": []}
for smokeurl in smokedl.find_all("div", {"class": "smokeurlrh"}):
quality = smokeurl.find("strong").text
@ -54,12 +53,25 @@ async def kusonimeBypass(url: str, slug=None):
links.append({"client": client, "url": url})
mendata["links"].append({"quality": quality, "link_download": links})
data.append(mendata)
num += 1
result.update({"error": False, "title": title, "thumb": thumb, "genre": genre, "genre_string": ", ".join(genre), "status_anime": status_anime, "season": season, "tipe": tipe, "ep": ep, "score": score, "duration": duration, "rilis": rilis, "data": data})
result |= {
"error": False,
"title": title,
"thumb": thumb,
"genre": genre,
"genre_string": ", ".join(genre),
"status_anime": status_anime,
"season": season,
"tipe": tipe,
"ep": ep,
"score": score,
"duration": duration,
"rilis": rilis,
"data": data,
}
except Exception:
err = traceback.format_exc()
LOGGER.error(err)
result.update({"error": True, "error_message": err})
result |= {"error": True, "error_message": err}
finally:
await http.delete(_url)
return result
@ -68,7 +80,8 @@ async def kusonimeBypass(url: str, slug=None):
async def byPassPh(url: str, name: str):
kusonime = await kusonimeBypass(url)
results = {"error": True, "error_message": kusonime}
template = """
if not kusonime["error"]:
template = """
<img src={{{thumb}}}>
<p><b>Title</b> : <code>{{title}}</code></p>
@ -92,13 +105,12 @@ async def byPassPh(url: str, name: str):
<br>
{{/data}}
""".strip()
if not kusonime["error"]:
html = chevron.render(template, kusonime)
telegraph = Telegraph()
if not telegraph.get_access_token():
await telegraph.create_account(short_name=BOT_USERNAME)
page = await telegraph.create_page(f"{kusonime.get('title')} By {escape(name)}", html_content=html)
results.update({"error": False, "url": "https://telegra.ph/{}".format(page["path"])})
results |= {"error": False, "url": f'https://telegra.ph/{page["path"]}'}
del results["error_message"]
return results

10
misskaty/helper/mediainfo_paste.py
View file

@ -161,7 +161,7 @@ def html_builder(title: str, text: str) -> str:
infobox = "<span class='container infobox'>"
subtitlebox = "<span class='container subtitlebox'>"
icon = "<img class='icons' src={icon_url} width='35px' height='35px' alt='' >"
html_msg = "<body>" + heading.format(content=title)
html_msg = f"<body>{heading.format(content=title)}"
for line in text.splitlines():
if ":" not in line and bool(line):
@ -194,14 +194,10 @@ def html_builder(title: str, text: str) -> str:
html_msg += subtitlebox if "Text #" in line else infobox
elif ":" in line:
if "Attachments" in line:
pass
elif "ErrorDetectionType" in line:
pass
else:
if "Attachments" not in line and "ErrorDetectionType" not in line:
html_msg += f"<div><code>{line.strip()}</code></div>"
elif not bool(line):
else:
html_msg += "</span>"
html_msg += "</span>"

14
misskaty/helper/subscene_helper.py
View file

@ -34,9 +34,17 @@ async def down_page(url):
releases = ""
for i in range(2):
r = release[i].text.strip()
releases = releases + f"\n{r}"
releases = f"{releases}\n{r}"
except Exception:
releases = ""
response = {"title": title, "imdb": imdb, "poster": poster, "author_name": author_name, "author_url": author_link, "download_url": download_url, "comments": comments, "releases": releases}
return response
return {
"title": title,
"imdb": imdb,
"poster": poster,
"author_name": author_name,
"author_url": author_link,
"download_url": download_url,
"comments": comments,
"releases": releases,
}

19
misskaty/plugins/locks.py
View file

@ -93,9 +93,9 @@ async def tg_lock(message, permissions: list, perm: str, lock: bool):
if perm not in permissions:
return await message.reply_text("Already locked.")
permissions.remove(perm)
elif perm in permissions:
return await message.reply_text("Already Unlocked.")
else:
if perm in permissions:
return await message.reply_text("Already Unlocked.")
permissions.append(perm)
permissions = {perm: True for perm in list(set(permissions))}
@ -126,12 +126,7 @@ async def locks_func(_, message):
permissions = await current_chat_permissions(chat_id)
if parameter in data:
await tg_lock(
message,
permissions,
data[parameter],
bool(state == "lock"),
)
await tg_lock(message, permissions, data[parameter], state == "lock")
elif parameter == "all" and state == "lock":
await app.set_chat_permissions(chat_id, ChatPermissions())
await message.reply_text(f"Locked Everything in {message.chat.title}")
@ -161,10 +156,7 @@ async def locktypes(_, message):
if not permissions:
return await message.reply_text("No Permissions.")
perms = ""
for i in permissions:
perms += f"__**{i}**__\n"
perms = "".join(f"__**{i}**__\n" for i in permissions)
await message.reply_text(perms)
@ -180,8 +172,7 @@ async def url_detector(_, message):
if user.id in mods or user.id in SUDO:
return
check = get_urls_from_text(text)
if check:
if check := get_urls_from_text(text):
permissions = await current_chat_permissions(chat_id)
if "can_add_web_page_previews" not in permissions:
try:

52
misskaty/plugins/media_extractor.py
View file

@ -120,31 +120,35 @@ async def ceksub(self: Client, ctx: Message, strings):
@use_chat_lang()
async def convertsrt(self: Client, ctx: Message, strings):
reply = ctx.reply_to_message
if reply and reply.document and (reply.document.file_name and reply.document.file_name.endswith((".vtt", ".ass", ".srt"))):
msg = await ctx.reply_msg(strings("convert_str"), quote=True)
if not os.path.exists("downloads"):
os.makedirs("downloads")
dl = await reply.download(file_name="downloads/")
filename = dl.split("/", 3)[3]
LOGGER.info(f"ConvertSub: {filename} by {ctx.from_user.first_name if ctx.from_user else ctx.sender_chat.title} [{ctx.from_user.id if ctx.from_user else ctx.sender_chat.id}]")
suffix = "srt" if ctx.command[0] == "converttosrt" else "ass"
(await shell_exec(f"{FF_MPEG_NAME} -i '{dl}' 'downloads/{filename}.{suffix}'"))[0]
c_time = time()
await ctx.reply_document(
f"downloads/{filename}.{suffix}",
caption=strings("capt_conv_sub").format(nf=filename, bot=self.me.username),
thumb="assets/thumb.jpg",
progress=progress_for_pyrogram,
progress_args=(strings("up_str"), msg, c_time, self.me.dc_id),
)
await msg.delete_msg()
try:
os.remove(dl)
os.remove(f"downloads/{filename}.{suffix}")
except:
pass
else:
if (
not reply
or not reply.document
or not reply.document.file_name
or not reply.document.file_name.endswith((".vtt", ".ass", ".srt"))
):
return await ctx.reply_msg(strings("conv_sub_help").format(cmd=ctx.command[0]), del_in=6)
msg = await ctx.reply_msg(strings("convert_str"), quote=True)
if not os.path.exists("downloads"):
os.makedirs("downloads")
dl = await reply.download(file_name="downloads/")
filename = dl.split("/", 3)[3]
LOGGER.info(f"ConvertSub: {filename} by {ctx.from_user.first_name if ctx.from_user else ctx.sender_chat.title} [{ctx.from_user.id if ctx.from_user else ctx.sender_chat.id}]")
suffix = "srt" if ctx.command[0] == "converttosrt" else "ass"
(await shell_exec(f"{FF_MPEG_NAME} -i '{dl}' 'downloads/{filename}.{suffix}'"))[0]
c_time = time()
await ctx.reply_document(
f"downloads/{filename}.{suffix}",
caption=strings("capt_conv_sub").format(nf=filename, bot=self.me.username),
thumb="assets/thumb.jpg",
progress=progress_for_pyrogram,
progress_args=(strings("up_str"), msg, c_time, self.me.dc_id),
)
await msg.delete_msg()
try:
os.remove(dl)
os.remove(f"downloads/{filename}.{suffix}")
except:
pass
@app.on_callback_query(filters.regex(r"^streamextract#"))

53
misskaty/plugins/ocr.py
View file

@ -30,26 +30,33 @@ __HELP__ = "/ocr [reply to photo] - Read Text From Image"
@use_chat_lang()
async def ocr(self: Client, ctx: Message, strings):
reply = ctx.reply_to_message
if reply and (reply.sticker or reply.photo or reply.document and reply.document.mime_type.startswith("image")):
msg = await ctx.reply_msg(strings("read_ocr"), quote=True)
try:
file_path = await reply.download()
if reply.sticker:
file_path = await reply.download(f"ocr_{ctx.from_user.id if ctx.from_user else ctx.sender_chat.id}.jpg")
response = await Telegraph().upload_file(file_path)
url = f"https://telegra.ph{response[0]['src']}"
req = (
await http.get(
f"https://script.google.com/macros/s/AKfycbwURISN0wjazeJTMHTPAtxkrZTWTpsWIef5kxqVGoXqnrzdLdIQIfLO7jsR5OQ5GO16/exec?url={url}",
follow_redirects=True,
)
).json()
await msg.edit_msg(strings("result_ocr").format(result=req["text"]))
if os.path.exists(file_path):
os.remove(file_path)
except Exception as e:
await msg.edit_msg(str(e))
if os.path.exists(file_path):
os.remove(file_path)
else:
return await ctx.reply_msg(strings("no_photo").format(cmd=ctx.command[0]), quote=True)
if (
not reply
or not reply.sticker
and not reply.photo
and (
not reply.document
or not reply.document.mime_type.startswith("image")
)
):
return await ctx.reply_msg(strings("no_photo").format(cmd=ctx.command[0]), quote=True)
msg = await ctx.reply_msg(strings("read_ocr"), quote=True)
try:
file_path = await reply.download()
if reply.sticker:
file_path = await reply.download(f"ocr_{ctx.from_user.id if ctx.from_user else ctx.sender_chat.id}.jpg")
response = await Telegraph().upload_file(file_path)
url = f"https://telegra.ph{response[0]['src']}"
req = (
await http.get(
f"https://script.google.com/macros/s/AKfycbwURISN0wjazeJTMHTPAtxkrZTWTpsWIef5kxqVGoXqnrzdLdIQIfLO7jsR5OQ5GO16/exec?url={url}",
follow_redirects=True,
)
).json()
await msg.edit_msg(strings("result_ocr").format(result=req["text"]))
if os.path.exists(file_path):
os.remove(file_path)
except Exception as e:
await msg.edit_msg(str(e))
if os.path.exists(file_path):
os.remove(file_path)

10
misskaty/plugins/web_scraper.py
View file

@ -411,11 +411,11 @@ async def getSame(msg, query, current_page, strings):
try:
index = int(current_page - 1)
PageLen = len(SCRAP_DICT[msg.id][0])
sameresult = ""
for c, i in enumerate(SCRAP_DICT[msg.id][0][index], start=1):
sameresult += f"<b>{index*6+c}. <a href='{i['url']}'>{i['title']}</a>\n<b>Status:</b> {i['sta']}\n</b>Rating:</b> {i['rate']}\n\n"
IGNORE_CHAR = "[]"
sameresult = "".join(i for i in sameresult if not i in IGNORE_CHAR)
sameresult = "".join(
f"<b>{index * 6 + c}. <a href='{i['url']}'>{i['title']}</a>\n<b>Status:</b> {i['sta']}\n</b>Rating:</b> {i['rate']}\n\n"
for c, i in enumerate(SCRAP_DICT[msg.id][0][index], start=1)
)
sameresult = "".join(i for i in sameresult if i not in "[]")
return sameresult, PageLen
except (IndexError, KeyError):
await msg.edit_msg(strings("no_result"), del_in=5)